src/main/java/org/onap/music/conductor/conditionals/MusicContional.java

   1 /*
   2  * ============LICENSE_START==========================================
   3  * org.onap.music
   4  * ===================================================================
   5  *  Copyright (c) 2017 AT&T Intellectual Property
   6  * ===================================================================
   7  *  Licensed under the Apache License, Version 2.0 (the "License");
   8  *  you may not use this file except in compliance with the License.
   9  *  You may obtain a copy of the License at
  10  *
  11  *     http://www.apache.org/licenses/LICENSE-2.0
  12  *
  13  *  Unless required by applicable law or agreed to in writing, software
  14  *  distributed under the License is distributed on an "AS IS" BASIS,
  15  *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16  *  See the License for the specific language governing permissions and
  17  *  limitations under the License.
  18  *
  19  * ============LICENSE_END=============================================
  20  * ====================================================================
  21  */
  22 package org.onap.music.conductor.conditionals;
  23
  24 import java.io.PrintWriter;
  25 import java.io.StringWriter;
  26 import java.util.HashMap;
  27 import java.util.Map;
  28
  29 import javax.ws.rs.core.Response.Status;
  30
  31 import org.codehaus.jettison.json.JSONObject;
  32 import org.onap.music.datastore.PreparedQueryObject;
  33 import org.onap.music.eelf.logging.EELFLoggerDelegate;
  34 import org.onap.music.eelf.logging.format.AppMessages;
  35 import org.onap.music.eelf.logging.format.ErrorSeverity;
  36 import org.onap.music.eelf.logging.format.ErrorTypes;
  37 import org.onap.music.lockingservice.MusicLockState;
  38 import org.onap.music.main.MusicCore;
  39 import org.onap.music.main.MusicUtil;
  40 import org.onap.music.main.ResultType;
  41 import org.onap.music.main.ReturnType;
  42 import org.onap.music.response.jsonobjects.JsonResponse;
  43 import org.onap.music.rest.RestMusicDataAPI;
  44
  45 import com.datastax.driver.core.ColumnDefinitions;
  46 import com.datastax.driver.core.DataType;
  47 import com.datastax.driver.core.ResultSet;
  48 import com.datastax.driver.core.Row;
  49 import com.datastax.driver.core.TableMetadata;
  50
  51 public class MusicContional {
  52         private static EELFLoggerDelegate logger = EELFLoggerDelegate.getLogger(RestMusicDataAPI.class);
  53
  54         public static ReturnType conditionalInsert(String keyspace, String tablename, String casscadeColumnName,
  55                         Map<String, Object> casscadeColumnData, String primaryKey, Map<String, Object> valuesMap,
  56                         Map<String, String> status) throws Exception {
  57
  58                 Map<String, PreparedQueryObject> queryBank = new HashMap<>();
  59                 TableMetadata tableInfo = null;
  60                 tableInfo = MusicCore.returnColumnMetadata(keyspace, tablename);
  61                 DataType primaryIdType = tableInfo.getPrimaryKey().get(0).getType();
  62                 String primaryId = tableInfo.getPrimaryKey().get(0).getName();
  63                 DataType casscadeColumnType = tableInfo.getColumn(casscadeColumnName).getType();
  64                 String vector = String.valueOf(Thread.currentThread().getId() + System.currentTimeMillis());
  65
  66                 PreparedQueryObject select = new PreparedQueryObject();
  67                 select.appendQueryString("SELECT * FROM " + keyspace + "." + tablename + " where " + primaryId + " = ?");
  68                 select.addValue(MusicUtil.convertToActualDataType(primaryIdType, primaryKey));
  69                 queryBank.put(MusicUtil.SELECT, select);
  70
  71                 PreparedQueryObject update = new PreparedQueryObject();
  72                 Map<String, String> updateColumnvalues = new HashMap<>(); //casscade column values
  73                 updateColumnvalues = getValues(true, casscadeColumnData, status);
  74                 Object formatedValues = MusicUtil.convertToActualDataType(casscadeColumnType, updateColumnvalues);
  75                 update.appendQueryString("UPDATE " + keyspace + "." + tablename + " SET " + casscadeColumnName + " ="
  76                                 + casscadeColumnName + " + ? , vector_ts = ?" + " WHERE " + primaryId + " = ? ");
  77                 update.addValue(formatedValues);
  78                 update.addValue(MusicUtil.convertToActualDataType(DataType.text(), vector));
  79                 update.addValue(MusicUtil.convertToActualDataType(primaryIdType, primaryKey));
  80                 queryBank.put(MusicUtil.UPDATE, update);
  81
  82                 Map<String, String> insertColumnvalues = new HashMap<>();
  83                 insertColumnvalues = getValues(false, casscadeColumnData, status);
  84                 formatedValues = MusicUtil.convertToActualDataType(casscadeColumnType, insertColumnvalues);
  85                 PreparedQueryObject insert = extractQuery(valuesMap, tableInfo, tablename, keyspace, primaryId, primaryKey,casscadeColumnName,formatedValues);
  86                 queryBank.put(MusicUtil.INSERT, insert);
  87
  88
  89                 String key = keyspace + "." + tablename + "." + primaryKey;
  90                 String lockId = MusicCore.createLockReference(key);
  91                 long leasePeriod = MusicUtil.getDefaultLockLeasePeriod();
  92                 ReturnType lockAcqResult = MusicCore.acquireLockWithLease(key, lockId, leasePeriod);
  93
  94                 try {
  95                         if (lockAcqResult.getResult().equals(ResultType.SUCCESS)) {
  96                                 ReturnType criticalPutResult = conditionalInsertAtomic(lockId, keyspace, tablename, primaryKey,
  97                                                 queryBank);
  98                                 MusicCore.releaseLock(lockId, true);
  99                                 if (criticalPutResult.getMessage().contains("insert"))
 100                                         criticalPutResult
 101                                                         .setMessage("Insert values: ");
 102                                 else if (criticalPutResult.getMessage().contains("update"))
 103                                         criticalPutResult
 104                                                         .setMessage("Update values: " + updateColumnvalues);
 105                                 return criticalPutResult;
 106
 107                         } else {
 108                                 MusicCore.releaseLock(lockId, true);
 109                                 return lockAcqResult;
 110                         }
 111                 } catch (Exception e) {
 112                         MusicCore.releaseLock(lockId, true);
 113                         return new ReturnType(ResultType.FAILURE, e.getMessage());
 114                 }
 115
 116         }
 117
 118         public static ReturnType conditionalInsertAtomic(String lockId, String keyspace, String tableName,
 119                         String primaryKey, Map<String, PreparedQueryObject> queryBank) {
 120
 121                 ResultSet results = null;
 122
 123                 try {
 124
 125                         MusicLockState mls = MusicCore.getLockingServiceHandle()
 126                                         .getLockState(keyspace + "." + tableName + "." + primaryKey);
 127                         if (mls.getLockHolder().equals(lockId) == true) {
 128                                 try {
 129                                         results = MusicCore.getDSHandle().executeCriticalGet(queryBank.get(MusicUtil.SELECT));
 130                                 } catch (Exception e) {
 131                                         return new ReturnType(ResultType.FAILURE, e.getMessage());
 132                                 }
 133                                 if (results.all().isEmpty()) {
 134                                         MusicCore.getDSHandle().executePut(queryBank.get(MusicUtil.INSERT), "critical");
 135                                         return new ReturnType(ResultType.SUCCESS, "insert");
 136                                 } else {
 137                                         MusicCore.getDSHandle().executePut(queryBank.get(MusicUtil.UPDATE), "critical");
 138                                         return new ReturnType(ResultType.SUCCESS, "update");
 139                                 }
 140                         } else {
 141                                 return new ReturnType(ResultType.FAILURE,
 142                                                 "Cannot perform operation since you are the not the lock holder");
 143                         }
 144
 145                 } catch (Exception e) {
 146                         StringWriter sw = new StringWriter();
 147                         e.printStackTrace(new PrintWriter(sw));
 148                         String exceptionAsString = sw.toString();
 149                         return new ReturnType(ResultType.FAILURE,
 150                                         "Exception thrown while doing the critical put, check sanctity of the row/conditions:\n"
 151                                                         + exceptionAsString);
 152                 }
 153
 154         }
 155
 156         public static ReturnType update(Map<String,PreparedQueryObject> queryBank, String keyspace, String tableName, String primaryKey,String primaryKeyValue,String planId,String cascadeColumnName,Map<String,String> cascadeColumnValues) {
 157
 158                 String key = keyspace + "." + tableName + "." + primaryKeyValue;
 159                 String lockId = MusicCore.createLockReference(key);
 160                 long leasePeriod = MusicUtil.getDefaultLockLeasePeriod();
 161                 ReturnType lockAcqResult = MusicCore.acquireLockWithLease(key, lockId, leasePeriod);
 162
 163                 try {
 164
 165                         if (lockAcqResult.getResult().equals(ResultType.SUCCESS)) {
 166                                 ReturnType updateResult= updateAtomic(lockId, keyspace, tableName, primaryKey,primaryKeyValue, queryBank,planId,cascadeColumnValues,cascadeColumnName);
 167                                 MusicCore.releaseLock(lockId, true);
 168                                 return updateResult;
 169
 170                         } else {
 171                                 MusicCore.releaseLock(lockId, true);
 172                                 return lockAcqResult;
 173                         }
 174
 175                 } catch (Exception e) {
 176                         MusicCore.releaseLock(lockId, true);
 177                         return new ReturnType(ResultType.FAILURE, e.getMessage());
 178
 179                 }
 180         }
 181
 182         public static ReturnType updateAtomic(String lockId, String keyspace, String tableName, String primaryKey,String primaryKeyValue,
 183                         Map<String,PreparedQueryObject> queryBank,String planId,Map<String,String> cascadeColumnValues,String casscadeColumnName) {
 184                 try {
 185
 186                         MusicLockState mls = MusicCore.getLockingServiceHandle()
 187                                         .getLockState(keyspace + "." + tableName + "." + primaryKeyValue);
 188                         if (mls.getLockHolder().equals(lockId) == true) {
 189                                 Row row  = MusicCore.getDSHandle().executeCriticalGet(queryBank.get(MusicUtil.SELECT)).one();
 190
 191                                 if(row != null) {
 192                                         Map<String, String> updatedValues = cascadeColumnUpdateSpecific(row, cascadeColumnValues, casscadeColumnName, planId);
 193                                         JSONObject json = new JSONObject(updatedValues);
 194                                         PreparedQueryObject update = new PreparedQueryObject();
 195                                         String vector_ts = String.valueOf(Thread.currentThread().getId() + System.currentTimeMillis());
 196                                         update.appendQueryString("UPDATE " + keyspace + "." + tableName + " SET " + casscadeColumnName + "['" + planId
 197                                                         + "'] = ?, vector_ts = ? WHERE " + primaryKey + " = ?");
 198                                         update.addValue(MusicUtil.convertToActualDataType(DataType.text(), json.toString()));
 199                                         update.addValue(MusicUtil.convertToActualDataType(DataType.text(), vector_ts));
 200                                         update.addValue(MusicUtil.convertToActualDataType(DataType.text(), primaryKeyValue));
 201                                         try {
 202                                                 MusicCore.getDSHandle().executePut(update, "critical");
 203                                         } catch (Exception ex) {
 204                                                 return new ReturnType(ResultType.FAILURE, ex.getMessage());
 205                                         }
 206                                 }else {
 207                                         return new ReturnType(ResultType.FAILURE,"Cannot find data related to key: "+primaryKey);
 208                                 }
 209
 210                                 MusicCore.getDSHandle().executePut(queryBank.get(MusicUtil.UPSERT), "critical");
 211                                 return new ReturnType(ResultType.SUCCESS, "update success");
 212
 213                         } else {
 214                                 return new ReturnType(ResultType.FAILURE,
 215                                                 "Cannot perform operation since you are the not the lock holder");
 216                         }
 217
 218                 } catch (Exception e) {
 219                         StringWriter sw = new StringWriter();
 220                         e.printStackTrace(new PrintWriter(sw));
 221                         String exceptionAsString = sw.toString();
 222                         return new ReturnType(ResultType.FAILURE,
 223                                         "Exception thrown while doing the critical put, check sanctity of the row/conditions:\n"
 224                                                         + exceptionAsString);
 225                 }
 226
 227         }
 228
 229         @SuppressWarnings("unchecked")
 230         public static Map<String, String> getValues(boolean isExists, Map<String, Object> casscadeColumnData,
 231                         Map<String, String> status) {
 232
 233                 Map<String, String> value = new HashMap<>();
 234                 Map<String, String> returnMap = new HashMap<>();
 235                 Object key = casscadeColumnData.get("key");
 236                 String setStatus = "";
 237                 value = (Map<String, String>) casscadeColumnData.get("value");
 238
 239                 if (isExists)
 240                         setStatus = status.get("exists");
 241                 else
 242                         setStatus = status.get("nonexists");
 243
 244                 value.put("status", setStatus);
 245                 JSONObject valueJson = new JSONObject(value);
 246                 returnMap.put(key.toString(), valueJson.toString());
 247                 return returnMap;
 248
 249         }
 250
 251
 252         public static PreparedQueryObject extractQuery(Map<String, Object> valuesMap, TableMetadata tableInfo, String tableName,
 253                         String keySpaceName,String primaryKeyName,String primaryKey,String casscadeColumn,Object casscadeColumnValues) throws Exception {
 254
 255                 PreparedQueryObject queryObject = new PreparedQueryObject();
 256                 StringBuilder fieldsString = new StringBuilder("(vector_ts"+",");
 257                 StringBuilder valueString = new StringBuilder("(" + "?" + ",");
 258                 String vector = String.valueOf(Thread.currentThread().getId() + System.currentTimeMillis());
 259                 queryObject.addValue(vector);
 260                 if(casscadeColumn!=null && casscadeColumnValues!=null) {
 261                         fieldsString.append("" +casscadeColumn+" ," );
 262                   valueString.append("?,");
 263                   queryObject.addValue(casscadeColumnValues);
 264                 }
 265
 266                 int counter = 0;
 267                 for (Map.Entry<String, Object> entry : valuesMap.entrySet()) {
 268
 269                         fieldsString.append("" + entry.getKey());
 270             Object valueObj = entry.getValue();
 271             if (primaryKeyName.equals(entry.getKey())) {
 272                 primaryKey = entry.getValue() + "";
 273                 primaryKey = primaryKey.replace("'", "''");
 274             }
 275             DataType colType = null;
 276             try {
 277                 colType = tableInfo.getColumn(entry.getKey()).getType();
 278             } catch(NullPointerException ex) {
 279                 logger.error(EELFLoggerDelegate.errorLogger,ex.getMessage() +" Invalid column name : "+entry.getKey(), AppMessages.INCORRECTDATA  ,ErrorSeverity.CRITICAL, ErrorTypes.DATAERROR);
 280
 281             }
 282
 283             Object formattedValue = null;
 284             try {
 285               formattedValue = MusicUtil.convertToActualDataType(colType, valueObj);
 286             } catch (Exception e) {
 287               logger.error(EELFLoggerDelegate.errorLogger,e.getMessage());
 288           }
 289
 290                         valueString.append("?");
 291             queryObject.addValue(formattedValue);
 292
 293
 294                         if (counter == valuesMap.size() - 1) {
 295                 fieldsString.append(")");
 296                 valueString.append(")");
 297             } else {
 298                 fieldsString.append(",");
 299                 valueString.append(",");
 300             }
 301             counter = counter + 1;
 302         }
 303         queryObject.appendQueryString("INSERT INTO " + keySpaceName + "." + tableName + " "
 304                 + fieldsString + " VALUES " + valueString);
 305                 return queryObject;
 306         }
 307
 308         public static Object getColValue(Row row, String colName, DataType colType) {
 309                 switch (colType.getName()) {
 310                 case VARCHAR:
 311                         return row.getString(colName);
 312                 case UUID:
 313                         return row.getUUID(colName);
 314                 case VARINT:
 315                         return row.getVarint(colName);
 316                 case BIGINT:
 317                         return row.getLong(colName);
 318                 case INT:
 319                         return row.getInt(colName);
 320                 case FLOAT:
 321                         return row.getFloat(colName);
 322                 case DOUBLE:
 323                         return row.getDouble(colName);
 324                 case BOOLEAN:
 325                         return row.getBool(colName);
 326                 case MAP:
 327                         return row.getMap(colName, String.class, String.class);
 328                 default:
 329                         return null;
 330                 }
 331         }
 332
 333         @SuppressWarnings("unchecked")
 334         public static Map<String, String> cascadeColumnUpdateSpecific(Row row, Map<String, String> changeOfStatus,
 335                         String cascadeColumnName, String planId) {
 336
 337                 ColumnDefinitions colInfo = row.getColumnDefinitions();
 338                 DataType colType = colInfo.getType(cascadeColumnName);
 339                 Map<String, String> values = new HashMap<>();
 340                 Object columnValue = getColValue(row, cascadeColumnName, colType);
 341
 342                 Map<String, String> finalValues = new HashMap<>();
 343                 values = (Map<String, String>) columnValue;
 344                 if (values.keySet().contains(planId)) {
 345                         String valueString = values.get(planId);
 346                         String tempValueString = valueString.replaceAll("\\{", "").replaceAll("\"", "").replaceAll("\\}", "");
 347                         String[] elements = tempValueString.split(",");
 348                         for (String str : elements) {
 349                                 String[] keyValue = str.split(":");
 350                                 if ((changeOfStatus.keySet().contains(keyValue[0].replaceAll("\\s", ""))))
 351                                 keyValue[1] = changeOfStatus.get(keyValue[0].replaceAll("\\s", ""));
 352                                 finalValues.put(keyValue[0], keyValue[1]);
 353                         }
 354                 }
 355                 return finalValues;
 356
 357         }
 358
 359 }