Add Column mode to NGSIToMongo Processor

Author: anmunoz

nifi-ngsi-bundle/nifi-ngsi-processors/src/main/java/org/apache/nifi/processors/ngsi/AbstractMongoProcessor.java

@@ -76,7 +76,7 @@ public abstract class AbstractMongoProcessor extends AbstractProcessor {
             .displayName("Attribute Persistence")
             .description("The mode of storing the data inside of the table")
             .required(false)
-            .allowableValues("row")
+            .allowableValues("row","column")
             .defaultValue("row")
             .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
             .build();
@@ -215,8 +215,6 @@ public abstract class AbstractMongoProcessor extends AbstractProcessor {
 
     static {
         descriptors.add(URI);
-       // descriptors.add(SSL_CONTEXT_SERVICE);
-       // descriptors.add(CLIENT_AUTH);
         descriptors.add(NGSI_VERSION);
         descriptors.add(DATA_MODEL);
         descriptors.add(ATTR_PERSISTENCE);

nifi-ngsi-bundle/nifi-ngsi-processors/src/main/java/org/apache/nifi/processors/ngsi/ngsi/aggregators/MongoAggregator.java

@@ -159,24 +159,22 @@ public void aggregate(Entity entity, long creationTime,String dataModel) {
             // get the event body
                 String entityId = entity.getEntityId();
                 String entityType = entity.getEntityType();
-                // iterate on all this context element attributes, if there are attributes
+            // iterate on all this context element attributes, if there are attributes
                 ArrayList<Attributes> contextAttributes = entity.getEntityAttrs();
-                Document doc = createDoc(recvTimeTs, entityId, entityType);
+                Document doc = createDoc(recvTimeTs, entityId, entityType, dataModel);
 
-                for (Attributes contextAttribute : contextAttributes) {
+            for (Attributes contextAttribute : contextAttributes) {
                     String attrName = contextAttribute.getAttrName();
                     String attrType = contextAttribute.getAttrType();
                     String attrValue = contextAttribute.getAttrValue();
-                    doc.append("attrName", attrName)
-                            .append("attrType", attrType)
-                            .append("attrValue", attrValue);
+                    doc.append( attrName,attrValue);
                 } // for
-
                 aggregation.add(doc);
         } // aggregate
 
-        private Document createDoc(long recvTimeTs, String entityId, String entityType) {
+        private Document createDoc(long recvTimeTs, String entityId, String entityType,String dataModel) {
             Document doc = new Document("recvTime", new Date(recvTimeTs));
+            doc.append("recvTimeTs", recvTimeTs);
 
             switch (dataModel) {
                 case "db-by-service-path":