Subscription for Individual Events (#80)

Added rules to create an aggregated object for each received event.

Author: saif-ericsson

pom.xml

@@ -282,4 +282,4 @@
 
         </plugins>
     </build>
-</project>
+</project>

src/main/java/com/ericsson/ei/controller/SubscriptionControllerImpl.java

@@ -144,6 +144,7 @@ public ResponseEntity<SubscriptionResponse> updateSubscriptions(@RequestBody Lis
         }
 
     }
+
     @Override
     @CrossOrigin
     @ApiOperation(value = "Removes the subscription from the database")

src/main/java/com/ericsson/ei/handlers/ObjectHandler.java

@@ -66,6 +66,10 @@ public class ObjectHandler {
     @Setter
     @Autowired
     private SubscriptionHandler subscriptionHandler;
+    
+    @Getter
+    @Value("${aggregated.collection.ttlValue}")
+    private int ttlValue;
 
     public boolean insertObject(String aggregatedObject, RulesObject rulesObject, String event, String id) {
         if (id == null) {
@@ -75,7 +79,10 @@ public boolean insertObject(String aggregatedObject, RulesObject rulesObject, St
         }
         JsonNode document = prepareDocumentForInsertion(id, aggregatedObject);
         log.debug("ObjectHandler: Aggregated Object document to be inserted: " + document.toString());
-        
+
+        mongoDbHandler.createTTLIndex(databaseName, collectionName, "Time", ttlValue);
+ 
+
         boolean result = mongoDbHandler.insertDocument(databaseName, collectionName, document.toString());
         if (result)
             eventToObjectMap.updateEventToObjectMapInMemoryDB(rulesObject, event, id);

src/main/java/com/ericsson/ei/mongodbhandler/MongoDBHandler.java

@@ -17,6 +17,7 @@
 package com.ericsson.ei.mongodbhandler;
 
 import java.util.ArrayList;
+import java.util.List;
 import java.util.concurrent.TimeUnit;
 
 import javax.annotation.PostConstruct;
@@ -87,21 +88,22 @@ public void createConnection(String host, int port) {
     public boolean insertDocument(String dataBaseName, String collectionName, String input) {
         try {
             MongoCollection<Document> collection = getMongoCollection(dataBaseName, collectionName);
-            final Document dbObjectInput = Document.parse(input);
-            collection.insertOne(dbObjectInput);
-            log.info("Object : " + input);
-            log.info("inserted successfully in ");
-            log.info("collection : " + collectionName + "and db : " + dataBaseName);
-            return true;
+            if (collection != null) {
+                final Document dbObjectInput = Document.parse(input);
+                collection.insertOne(dbObjectInput);
+                log.debug("Object : " + input);
+                log.debug("inserted successfully in ");
+                log.debug("collection : " + collectionName + "and db : " + dataBaseName);
+                return true;
+            }
         } catch (MongoWriteException e) {
             log.error(e.getMessage(), e);
         }
         return false;
     }
 
     /**
-     * This method is used for the retrieve the all documents from the
-     * collection
+     * This method is used for the retrieve the all documents from the collection
      * 
      * @param dataBaseName
      * @param collectionName
@@ -111,15 +113,17 @@ public ArrayList<String> getAllDocuments(String dataBaseName, String collectionN
         ArrayList<String> result = new ArrayList<>();
         try {
             MongoCollection<Document> collection = getMongoCollection(dataBaseName, collectionName);
-            collection.find(new BasicDBObject()).forEach((Block<Document>) document -> {
-                result.add(JSON.serialize(document));
-            });
-            if (result.size() != 0) {
-                log.debug("getAllDocuments() :: database: " + dataBaseName + " and collection: " + collectionName
-                        + " fetched No of :" + result.size());
-            } else {
-                log.debug("getAllDocuments() :: database: " + dataBaseName + "and collection: " + collectionName
-                        + " documents are not found");
+            if (collection != null) {
+                collection.find(new BasicDBObject()).forEach((Block<Document>) document -> {
+                    result.add(JSON.serialize(document));
+                });
+                if (result.size() != 0) {
+                    log.debug("getAllDocuments() :: database: " + dataBaseName + " and collection: " + collectionName
+                            + " fetched No of :" + result.size());
+                } else {
+                    log.debug("getAllDocuments() :: database: " + dataBaseName + "and collection: " + collectionName
+                            + " documents are not found");
+                }
             }
         } catch (Exception e) {
             log.error(e.getMessage(), e);
@@ -143,15 +147,17 @@ public ArrayList<String> find(String dataBaseName, String collectionName, String
 
         try {
             MongoCollection<Document> collection = getMongoCollection(dataBaseName, collectionName);
-            collection.find(BasicDBObject.parse(condition)).forEach((Block<Document>) document -> {
-                result.add(JSON.serialize(document));
-            });
-            if (result.size() != 0) {
-                log.debug("find() :: database: " + dataBaseName + " and collection: " + collectionName
-                        + " fetched No of :" + result.size());
-            } else {
-                log.debug("find() :: database: " + dataBaseName + " and collection: " + collectionName
-                        + " documents are not found");
+            if (collection != null) {
+                collection.find(BasicDBObject.parse(condition)).forEach((Block<Document>) document -> {
+                    result.add(JSON.serialize(document));
+                });
+                if (result.size() != 0) {
+                    log.debug("find() :: database: " + dataBaseName + " and collection: " + collectionName
+                            + " fetched No of :" + result.size());
+                } else {
+                    log.debug("find() :: database: " + dataBaseName + " and collection: " + collectionName
+                            + " documents are not found");
+                }
             }
         } catch (Exception e) {
             log.error(e.getMessage(), e);
@@ -161,8 +167,8 @@ public ArrayList<String> find(String dataBaseName, String collectionName, String
     }
 
     /**
-     * This method is used for update the document in collection and remove the
-     * lock in one query. Lock is needed for multi process execution
+     * This method is used for update the document in collection and remove the lock
+     * in one query. Lock is needed for multi process execution
      * 
      * @param dataBaseName
      * @param collectionName
@@ -175,12 +181,14 @@ public ArrayList<String> find(String dataBaseName, String collectionName, String
     public boolean updateDocument(String dataBaseName, String collectionName, String input, String updateInput) {
         try {
             MongoCollection<Document> collection = getMongoCollection(dataBaseName, collectionName);
-            final Document dbObjectInput = Document.parse(input);
-            final Document dbObjectUpdateInput = Document.parse(updateInput);
-            UpdateResult updateMany = collection.replaceOne(dbObjectInput, dbObjectUpdateInput);
-            log.debug("updateDocument() :: database: " + dataBaseName + " and collection: " + collectionName
-                    + " is document Updated :" + updateMany.wasAcknowledged());
-            return updateMany.wasAcknowledged();
+            if (collection != null) {
+                final Document dbObjectInput = Document.parse(input);
+                final Document dbObjectUpdateInput = Document.parse(updateInput);
+                UpdateResult updateMany = collection.replaceOne(dbObjectInput, dbObjectUpdateInput);
+                log.debug("updateDocument() :: database: " + dataBaseName + " and collection: " + collectionName
+                        + " is document Updated :" + updateMany.wasAcknowledged());
+                return updateMany.wasAcknowledged();
+            }
         } catch (Exception e) {
             log.error(e.getMessage(), e);
         }
@@ -189,9 +197,9 @@ public boolean updateDocument(String dataBaseName, String collectionName, String
     }
 
     /**
-     * This method is used for lock and return the document that matches the
-     * input condition in one query. Lock is needed for multi process execution.
-     * This method is executed in a loop.
+     * This method is used for lock and return the document that matches the input
+     * condition in one query. Lock is needed for multi process execution. This
+     * method is executed in a loop.
      * 
      * @param dataBaseName
      * @param collectionName
@@ -204,13 +212,15 @@ public boolean updateDocument(String dataBaseName, String collectionName, String
     public Document findAndModify(String dataBaseName, String collectionName, String input, String updateInput) {
         try {
             MongoCollection<Document> collection = getMongoCollection(dataBaseName, collectionName);
-            final Document dbObjectInput = Document.parse(input);
-            final Document dbObjectUpdateInput = Document.parse(updateInput);
-            Document result = collection.findOneAndUpdate(dbObjectInput, dbObjectUpdateInput);
-            if (result != null) {
-                log.debug("updateDocument() :: database: " + dataBaseName + " and collection: " + collectionName
-                        + " updated successfully");
-                return result;
+            if (collection != null) {
+                final Document dbObjectInput = Document.parse(input);
+                final Document dbObjectUpdateInput = Document.parse(updateInput);
+                Document result = collection.findOneAndUpdate(dbObjectInput, dbObjectUpdateInput);
+                if (result != null) {
+                    log.debug("updateDocument() :: database: " + dataBaseName + " and collection: " + collectionName
+                            + " updated successfully");
+                    return result;
+                }
             }
         } catch (Exception e) {
             log.error(e.getMessage(), e);
@@ -231,16 +241,18 @@ public Document findAndModify(String dataBaseName, String collectionName, String
     public boolean dropDocument(String dataBaseName, String collectionName, String condition) {
         try {
             MongoCollection<Document> collection = getMongoCollection(dataBaseName, collectionName);
-            final Document dbObjectCondition = Document.parse(condition);
-            DeleteResult deleteMany = collection.deleteMany(dbObjectCondition);
-            if (deleteMany.getDeletedCount() > 0) {
-                log.debug("database" + dataBaseName + " and collection: " + collectionName + " deleted No.of records "
-                        + deleteMany.getDeletedCount());
-                return true;
-            } else {
-                log.debug("database " + dataBaseName + " and collection: " + collectionName
-                        + " No documents found to delete");
-                return false;
+            if (collection != null) {
+                final Document dbObjectCondition = Document.parse(condition);
+                DeleteResult deleteMany = collection.deleteMany(dbObjectCondition);
+                if (deleteMany.getDeletedCount() > 0) {
+                    log.debug("database" + dataBaseName + " and collection: " + collectionName
+                            + " deleted No.of records " + deleteMany.getDeletedCount());
+                    return true;
+                } else {
+                    log.debug("database " + dataBaseName + " and collection: " + collectionName
+                            + " No documents found to delete");
+                    return false;
+                }
             }
         } catch (Exception e) {
             log.error(e.getMessage(), e);
@@ -265,12 +277,15 @@ public void createTTLIndex(String dataBaseName, String collectionName, String fi
     }
 
     private MongoCollection<Document> getMongoCollection(String dataBaseName, String collectionName) {
+        if (mongoClient == null)
+            return null;
         MongoDatabase db = mongoClient.getDatabase(dataBaseName);
-        if (!db.listCollectionNames().into(new ArrayList<String>()).contains(collectionName)) {
-            log.info("The requested database(" + dataBaseName + ") / collection(" + collectionName
+        List<String> collectionList = db.listCollectionNames().into(new ArrayList<String>());
+        if (!collectionList.contains(collectionName)) {
+            log.debug("The requested database(" + dataBaseName + ") / collection(" + collectionName
                     + ") not available in mongodb, Creating ........");
             db.createCollection(collectionName);
-            log.info("done....");
+            log.debug("done....");
         }
         MongoCollection<Document> collection = db.getCollection(collectionName);
         return collection;