Merge pull request #2026 from telefonicaid/task/cast_date_attr_types_…

…datetime_in_sink Task/cast date attr types datetime in sink
telefonicaid · May 11, 2021 · 5d173f0 · 5d173f0
2 parents 650644d + e17130e
commit 5d173f0
Showing 1 changed file with 32 additions and 18 deletions.
diff --git a/cygnus-ngsi/src/main/java/com/telefonica/iot/cygnus/sinks/NGSIMongoSink.java b/cygnus-ngsi/src/main/java/com/telefonica/iot/cygnus/sinks/NGSIMongoSink.java
@@ -209,6 +209,21 @@ protected ArrayList<String> getKeysToCrop (boolean rowAttrPersistence){
         }
         return keysToCrop;
     }
+
+    protected BasicDBObject castDate (String key, BasicDBObject basicDBObject){
+        try {
+            LOGGER.debug("[" + this.getName() + "] Casting to Date att with key: " + key + " value: "
+                         + basicDBObject.get(key));
+            String str = basicDBObject.get(key).toString();
+            Date date = new Date(CommonUtils.getTimeInstantFromString(str));
+            basicDBObject.put(key, date);
+        } catch (Exception e) {
+            // Never reached
+            LOGGER.error("[" + this.getName() + "] Casting to Date att with key: " + key + " value: "
+                         + basicDBObject.get(key) + " Details=" + e.getMessage());
+        }
+        return basicDBObject;
+    }
 
     private void persistAggregation(NGSIGenericAggregator aggregator) throws CygnusPersistenceError {
         ArrayList<String> keysToCrop = getKeysToCrop(rowAttrPersistence);
@@ -217,25 +232,24 @@ private void persistAggregation(NGSIGenericAggregator aggregator) throws CygnusP
         ArrayList<Document> aggregation = new ArrayList<>();
         for (int i = 0 ; i < jsonObjects.size() ; i++) {
             BasicDBObject basicDBObject = BasicDBObject.parse(jsonObjects.get(i).toString());
-            for (String key : basicDBObject.keySet()) {
-                try {
-                    ArrayList<JsonElement> values;
-                    if (rowAttrPersistence) {
-                        values = aggregator.getAggregation().get(NGSIConstants.ATTR_TYPE);
-                    } else {
-                        values = aggregator.getAggregation().get(key + NGSIConstants.AUTOGENERATED_ATTR_TYPE);
-                    }
-                    if ( (values != null) &&
-                         (values.toString().contains("ISO8601") ||
-                          values.toString().contains("DateTime")) ) {
-                        LOGGER.debug("[" + this.getName() + "] Casting to Date att with key: " + key + " value: " + basicDBObject.get(key));
-                        String str = basicDBObject.get(key).toString();
-                        Date date = new Date(CommonUtils.getTimeInstantFromString(str));
-                        basicDBObject.put(key, date);
+            JsonElement attType;
+            if (rowAttrPersistence) {
+                attType = aggregator.getAggregation().get(NGSIConstants.ATTR_TYPE).get(i);
+                if ( (attType != null &&
+                      attType.isJsonPrimitive()) &&
+                     (attType.getAsString().equals("ISO8601") ||
+                      attType.getAsString().equals("DateTime")) ) {
+                    basicDBObject = castDate(NGSIConstants.ATTR_VALUE, basicDBObject);
+                }
+            } else {
+                for (String key : basicDBObject.keySet()) {
+                    attType = aggregator.getAggregation().get(key + NGSIConstants.AUTOGENERATED_ATTR_TYPE).get(i);
+                    if ( (attType != null &&
+                          attType.isJsonPrimitive()) &&
+                         (attType.getAsString().equals("ISO8601") ||
+                          attType.getAsString().equals("DateTime")) ) {
+                        basicDBObject = castDate(key, basicDBObject);
                     }
-                } catch (Exception e) {
-                    // Never reached
-                    LOGGER.error("[" + this.getName() + "] Casting to Date att with key: " + key + " value: " + basicDBObject.get(key) + " Details=" + e.getMessage());
                 }
             }
             aggregation.add(new Document(basicDBObject.toMap()));