musictrigger/src/MusicTrigger.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
  23 import java.util.Arrays;
  24 import java.util.Collection;
  25 import java.util.Collections;
  26 import java.util.HashMap;
  27 import java.util.Iterator;
  28 import java.util.List;
  29 import java.nio.ByteBuffer;
  30 import java.nio.charset.Charset;
  31 import java.nio.charset.StandardCharsets;
  32 import java.util.ArrayList;
  33 import java.util.Map;
  34
  35 import javax.ws.rs.core.HttpHeaders;
  36 import javax.ws.rs.core.MediaType;
  37
  38 import org.apache.cassandra.config.ColumnDefinition;
  39 import org.apache.cassandra.db.Clustering;
  40 import org.apache.cassandra.db.Mutation;
  41 import org.apache.cassandra.db.partitions.Partition;
  42 import org.apache.cassandra.db.rows.Cell;
  43 import org.apache.cassandra.db.rows.Row;
  44 import org.apache.cassandra.db.rows.Unfiltered;
  45 import org.apache.cassandra.db.rows.UnfilteredRowIterator;
  46 import org.apache.cassandra.triggers.ITrigger;
  47 import org.slf4j.Logger;
  48 import org.slf4j.LoggerFactory;
  49
  50 import com.sun.jersey.api.client.Client;
  51 import com.sun.jersey.api.client.ClientResponse;
  52 import com.sun.jersey.api.client.WebResource;
  53
  54 public class MusicTrigger implements ITrigger {
  55
  56         private static final Logger logger = LoggerFactory.getLogger(MusicTrigger.class);
  57
  58
  59     public Collection<Mutation> augment(Partition partition)
  60     {
  61         boolean isDelete = false;
  62         logger.info("Step 1: "+partition.partitionLevelDeletion().isLive());
  63         if(partition.partitionLevelDeletion().isLive()) {
  64
  65         } else {
  66             // Partition Level Deletion
  67                 isDelete = true;
  68         }
  69         logger.info("MusicTrigger isDelete: " + isDelete);
  70         String ksName = partition.metadata().ksName;
  71         String tableName = partition.metadata().cfName;
  72         logger.info("MusicTrigger Table: " + tableName);
  73         boolean isInsert = checkQueryType(partition);
  74         org.json.simple.JSONObject obj = new org.json.simple.JSONObject();
  75
  76         String operation = null;
  77         if(isDelete)
  78                 operation = "delete";
  79         else if(isInsert)
  80                 operation = "insert";
  81         else
  82                 operation = "update";
  83         Map<String, Object> changeMap = new HashMap<>();
  84
  85         obj.put("operation", operation);
  86         obj.put("keyspace", ksName);
  87         obj.put("table_name", tableName);
  88         obj.put("full_table", ksName+"."+tableName);
  89         obj.put("primary_key", partition.metadata().getKeyValidator().getString(partition.partitionKey().getKey()));
  90         List<String> updateList = new ArrayList<>();
  91         //obj.put("message_id", partition.metadata().getKeyValidator().getString(partition.partitionKey().getKey()));
  92         if("update".equals(operation)) {
  93                 try {
  94                     UnfilteredRowIterator it = partition.unfilteredIterator();
  95                     while (it.hasNext()) {
  96                         Unfiltered un = it.next();
  97                         Clustering clt = (Clustering) un.clustering();
  98                         Iterator<Cell> cells = partition.getRow(clt).cells().iterator();
  99                         Iterator<ColumnDefinition> columns = partition.getRow(clt).columns().iterator();
 100
 101                         while(columns.hasNext()){
 102                             ColumnDefinition columnDef = columns.next();
 103                             Cell cell = cells.next();
 104
 105                             String data = null;
 106                             if(cell.column().type.toString().equals("org.apache.cassandra.db.marshal.UTF8Type")) {
 107                                 logger.info(">> type is String");
 108                                 data = new String(cell.value().array()); // If cell type is text
 109                             } else if(cell.column().type.toString().equals("org.apache.cassandra.db.marshal.Int32Type")) {
 110                                 //ByteBuffer wrapped = ByteBuffer.wrap(cell.value()); // big-endian by default
 111                                 int num = fromByteArray(cell.value().array());
 112                                 logger.info(">> type is Integer1 :: "+num);
 113                                 data = String.valueOf(num);
 114                             }
 115
 116                             logger.info("Inside triggers loop: "+columnDef.name+" : "+data);
 117                             //changeMap.put(ksName+"."+tableName+"."+columnDef.name,data);
 118                             updateList.add(ksName+"."+tableName+":"+columnDef.name+":"+data);
 119                             changeMap.put("field_value",ksName+"."+tableName+":"+columnDef.name+":"+data);
 120
 121                         }
 122                     }
 123                 } catch (Exception e) {
 124
 125                 }
 126                 obj.put("updateList", updateList);
 127         } else {
 128                 changeMap.put("field_value", ksName+"."+tableName);
 129         }
 130
 131         obj.put("changeValue", changeMap);
 132         logger.info("Sending response: "+obj.toString());
 133         try {
 134             notifyMusic(obj.toString());
 135         } catch(Exception e) {
 136             e.printStackTrace();
 137             logger.error("Notification failed..."+e.getMessage());
 138         }
 139         return Collections.emptyList();
 140     }
 141
 142     private int fromByteArray(byte[] bytes) {
 143         return bytes[0] << 24 | (bytes[1] & 0xFF) << 16 | (bytes[2] & 0xFF) << 8 | (bytes[3] & 0xFF);
 144     }
 145
 146     private boolean checkQueryType(Partition partition) {
 147         UnfilteredRowIterator it = partition.unfilteredIterator();
 148         while (it.hasNext()) {
 149             Unfiltered unfiltered = it.next();
 150             Row row = (Row) unfiltered;
 151             if (isInsert(row)) {
 152                 return true;
 153             }
 154         }
 155         return false;
 156     }
 157
 158     private boolean isInsert(Row row) {
 159         return row.primaryKeyLivenessInfo().timestamp() != Long.MIN_VALUE;
 160     }
 161
 162         private void notifyMusic(String request) {
 163                 System.out.println("notifyMusic...");
 164                 Client client = Client.create();
 165                 WebResource webResource = client.resource("http://localhost:8080/MUSIC/rest/v2/admin/callbackOps");
 166
 167                 JSONObject data = new JSONObject();
 168                 data.setData(request);
 169
 170                 ClientResponse response = webResource.accept("application/json").type("application/json")
 171                 .post(ClientResponse.class, data);
 172
 173                 if(response.getStatus() != 200){
 174                         System.out.println("Exception while notifying MUSIC...");
 175         }
 176                 /*response.getHeaders().put(HttpHeaders.CONTENT_TYPE, Arrays.asList(MediaType.APPLICATION_JSON));
 177                 response.bufferEntity();
 178                 String x = response.getEntity(String.class);
 179                 System.out.println("Response: "+x);*/
 180
 181         }
 182
 183         /*public Collection<Mutation> augment(Partition partition) {
 184
 185                 String tableName = partition.metadata().cfName;
 186         System.out.println("Table: " + tableName);
 187
 188         JSONObject obj = new JSONObject();
 189         obj.put("message_id", partition.metadata().getKeyValidator().getString(partition.partitionKey().getKey()));
 190
 191
 192             try {
 193                 UnfilteredRowIterator it = partition.unfilteredIterator();
 194                 while (it.hasNext()) {
 195                     Unfiltered un = it.next();
 196                     Clustering clt = (Clustering) un.clustering();
 197                     Iterator<Cell> cls = partition.getRow(clt).cells().iterator();
 198                 Iterator<ColumnDefinition> columns = partition.getRow(clt).columns().iterator();
 199
 200                     while(cls.hasNext()){
 201                         Cell cell = cls.next();
 202                         String data = new String(cell.value().array()); // If cell type is text
 203                         System.out.println(cell + " : " +data);
 204
 205                     }
 206                     while(columns.hasNext()){
 207                     ColumnDefinition columnDef = columns.next();
 208                     Cell cell = cls.next();
 209                     String data = new String(cell.value().array()); // If cell type is text
 210                     obj.put(columnDef.toString(), data);
 211                 }
 212                 }
 213             } catch (Exception e) {
 214             }
 215
 216             System.out.println(obj.toString());
 217
 218         return Collections.emptyList();
 219         }*/
 220
 221 }