Modified KafkaConsumerRestDepositor JSON processing

README.md

@@ -86,3 +86,10 @@ mvn test
 ```
 
 It should be noted that Maven & Java are required to run the unit tests. If you do not have Maven or Java installed, you can reopen the project in the provided dev container and run the tests from there.
+
+# Object data consumption
+The KafkaConsumerRestDepositor will accept any string as input to be passed into the SDW. If provided a JSON object, the tokens of "encodedMsg" and "estimatedRemovalDate" will be passed through directly to the SDW in the form of the following:
+{depositRequests:[{"encodeType": STRING ,"encodedMsg": STRING, "estimatedRemovalDate": STRING}]}
+
+If provided a string of non-json form, the value of "encodedMsg" will inherit the passed value and information will be passed to the SDW in the form of the following:
+{depositRequests:[{"encodeType": STRING ,"encodedMsg": STRING}]}

src/main/java/jpo/sdw/depositor/consumerdepositors/KafkaConsumerRestDepositor.java

@@ -30,15 +30,14 @@ public static boolean loop() {
    private RestDepositor<String> restDepositor;
    private KafkaConsumer<String, String> kafkaConsumer;
    private JSONObject jsonMsgList;
-   private JSONObject jsonMsg;
+   private String encodeType;
 
    public KafkaConsumerRestDepositor(KafkaConsumer<String, String> kafkaConsumer, RestDepositor<String> restDepositor,
          String encodeType) {
       this.setKafkaConsumer(kafkaConsumer);
       this.setRestDepositor(restDepositor);
       this.jsonMsgList = new JSONObject();
-      this.jsonMsg = new JSONObject();
-      this.jsonMsg.put("encodeType", encodeType);
+      this.encodeType = encodeType;
    }
 
    @Override
@@ -47,10 +46,13 @@ public void run(String... topics) {
       while (LoopController.loop()) { // NOSONAR (used for unit testing)
          ConsumerRecords<String, String> records = this.getKafkaConsumer().poll(Duration.ofMillis(100));
          JSONArray jsonRequests = new JSONArray();
+
          for (ConsumerRecord<String, String> record : records) {
             logger.info("Depositing message {}", record);
-            this.jsonMsg.put("encodedMsg", record.value());
-            jsonRequests.put(new JSONObject(jsonMsg.toString()));
+
+            JSONObject depositMessage = prepareJSONObject(record.value());
+
+            jsonRequests.put(new JSONObject(depositMessage.toString()));
          }
          if (records.count() != 0) {
             this.jsonMsgList.put("depositRequests", jsonRequests);
@@ -59,6 +61,22 @@ public void run(String... topics) {
       }
    }
 
+   private JSONObject prepareJSONObject(String record) {
+      // Old version treated record as value for static field "encodedMsg"
+      // Try/Catch around new optional json object to attach meta data. Fall back if
+      // record isn't json.
+      JSONObject deposit = new JSONObject();
+      deposit.put("encodeType", this.encodeType);
+      try {
+         JSONObject recordObj = new JSONObject(record);
+         deposit.put("encodedMsg", recordObj.getString("encodedMsg"));
+         deposit.put("estimatedRemovalDate", recordObj.opt("estimatedRemovalDate"));
+      } catch (Exception e) {
+         deposit.put("encodedMsg", record);
+      }
+      return deposit;      
+   }
+
    public RestDepositor<String> getRestDepositor() {
       return restDepositor;
    }

src/test/java/jpo/sdw/depositor/consumerdepositors/KafkaConsumerRestDepositorTest.java

@@ -16,6 +16,7 @@
 import org.apache.kafka.clients.consumer.ConsumerRecords;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.apache.kafka.common.TopicPartition;
+import org.json.JSONObject;
 import org.junit.Test;
 
 import jpo.sdw.depositor.consumerdepositors.KafkaConsumerRestDepositor.LoopController;
@@ -43,7 +44,7 @@ public class KafkaConsumerRestDepositorTest {
    public void runShouldDepositMessage() {
 
       List<ConsumerRecord<String, String>> crList = new ArrayList<ConsumerRecord<String, String>>();
-      crList.add(new ConsumerRecord<String, String>("key", 0, 0, "value", null));
+      crList.add(new ConsumerRecord<String, String>("key", 0, 0, "value", "Message"));
 
       Map<TopicPartition, List<ConsumerRecord<String, String>>> recordsMap = new HashMap<TopicPartition, List<ConsumerRecord<String, String>>>();
       recordsMap.put(new TopicPartition("string", 0), crList);
@@ -74,6 +75,76 @@ public boolean loop(Invocation inv) {
       testKafkaConsumerRestDepositor.run("testTopic");
    }
 
+    @Test
+   public void runShouldDepositJSONMessage() {
+
+      List<ConsumerRecord<String, String>> crList = new ArrayList<ConsumerRecord<String, String>>();
+      crList.add(new ConsumerRecord<String, String>("key", 0, 0, "value", "{\"encodedMsg\":\"C4400000000680C0DE3\"}"));
+
+      Map<TopicPartition, List<ConsumerRecord<String, String>>> recordsMap = new HashMap<TopicPartition, List<ConsumerRecord<String, String>>>();
+      recordsMap.put(new TopicPartition("string", 0), crList);
+
+      final ConsumerRecords<String, String> testConsumerRecords = new ConsumerRecords<String, String>(recordsMap);
+
+      new MockUp<LoopController>() {
+         @Mock
+         public boolean loop(Invocation inv) {
+            if (inv.getInvocationIndex() == 0) {
+               return true;
+            } else {
+               return false;
+            }
+         }
+      };
+
+      new Expectations() {
+         {
+            injectableKafkaConsumer.poll((Duration)any);
+            result = testConsumerRecords;
+
+            injectableRestDepositor.deposit("{\"depositRequests\":[{\"encodeType\":\"\",\"encodedMsg\":\"C4400000000680C0DE3\"}]}");
+            times = 1;
+         }
+      };
+
+      testKafkaConsumerRestDepositor.run("testTopic");
+   }
+
+   @Test
+   public void runShouldDepositJSONEstimatedRemovalDateMessage() {
+
+      List<ConsumerRecord<String, String>> crList = new ArrayList<ConsumerRecord<String, String>>();
+      crList.add(new ConsumerRecord<String, String>("key", 0, 0, "value", "{\"encodedMsg\":\"C4400000000680C0DE3\",\"estimatedRemovalDate\":\"2023-12-01T17:47:11-05:15\"}"));
+
+      Map<TopicPartition, List<ConsumerRecord<String, String>>> recordsMap = new HashMap<TopicPartition, List<ConsumerRecord<String, String>>>();
+      recordsMap.put(new TopicPartition("string", 0), crList);
+
+      final ConsumerRecords<String, String> testConsumerRecords = new ConsumerRecords<String, String>(recordsMap);
+
+      new MockUp<LoopController>() {
+         @Mock
+         public boolean loop(Invocation inv) {
+            if (inv.getInvocationIndex() == 0) {
+               return true;
+            } else {
+               return false;
+            }
+         }
+      };
+
+      new Expectations() {
+         {
+            injectableKafkaConsumer.poll((Duration)any);
+            result = testConsumerRecords;
+
+            injectableRestDepositor.deposit("{\"depositRequests\":[{\"encodeType\":\"\",\"encodedMsg\":\"C4400000000680C0DE3\",\"estimatedRemovalDate\":\"2023-12-01T17:47:11-05:15\"}]}");
+            times = 1;
+         }
+      };
+
+      testKafkaConsumerRestDepositor.run("testTopic");
+   }
+
    @Test
    public void loopControllerShouldAlwaysReturnTrue() {
       assertTrue(LoopController.loop());
@@ -93,4 +164,4 @@ public void testConstructorIsPrivate()
       }
    }
 
-}
+}