1 files changed, 7 insertions, 8 deletions

diff --git a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/service/JsonMessageParser.java b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/service/JsonMessageParser.java
index eed0f0bd..708865fa 100644
--- a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/service/JsonMessageParser.java
+++ b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/service/JsonMessageParser.java
@@ -1,6 +1,6 @@
 /*-
  * ============LICENSE_START========================================================================
- * Copyright (C) 2018 NOKIA Intellectual Property, 2018-2019 Nordix Foundation. All rights reserved.
+ * Copyright (C) 2018, 2020 NOKIA Intellectual Property, 2018-2019 Nordix Foundation. All rights reserved.
  * ==================================================================================================
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -96,18 +96,17 @@ public class JsonMessageParser {
      * @return a <code>Flux</code> containing messages.
      */
 
-    public Flux<FileReadyMessage> getMessagesFromJson(Mono<JsonElement> rawMessage) {
-        return rawMessage.flatMapMany(this::createMessageData);
+    public Flux<FileReadyMessage> getMessagesFromJson(Flux<JsonElement> rawMessage) {
+        return rawMessage.flatMap(this::createMessageData);
     }
 
     Optional<JsonObject> getJsonObjectFromAnArray(JsonElement element) {
-        JsonParser jsonParser = new JsonParser();
         if (element.isJsonPrimitive()) {
-            return Optional.of(jsonParser.parse(element.getAsString()).getAsJsonObject());
+            return Optional.of(JsonParser.parseString(element.getAsString()).getAsJsonObject());
         } else if (element.isJsonObject()) {
             return Optional.of((JsonObject) element);
         } else {
-            return Optional.of(jsonParser.parse(element.toString()).getAsJsonObject());
+            return Optional.of(JsonParser.parseString(element.toString()).getAsJsonObject());
         }
     }
 
@@ -117,9 +116,9 @@ public class JsonMessageParser {
     }
 
     /**
-     * Extract info from string and create a Flux of {@link FileReadyMessage}.
+     * Extract info from jsonElement and create a Flux of {@link FileReadyMessage}.
      *
-     * @param rawMessage - results from DMaaP
+     * @param jsonElement - result from DMaaP
      * @return reactive Flux of FileReadyMessages
      */
     private Flux<FileReadyMessage> createMessageData(JsonElement jsonElement) {