Migrate TimDepositController to use Spring Kafka

jpo-ode-svcs/src/main/java/us/dot/its/jpo/ode/traveler/TimDepositController.java

@@ -27,13 +27,13 @@
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.ResponseEntity;
+import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.web.bind.annotation.CrossOrigin;
 import org.springframework.web.bind.annotation.PostMapping;
 import org.springframework.web.bind.annotation.PutMapping;
 import org.springframework.web.bind.annotation.RequestBody;
 import org.springframework.web.bind.annotation.RestController;
 import us.dot.its.jpo.ode.coder.OdeTimDataCreatorHelper;
-import us.dot.its.jpo.ode.kafka.OdeKafkaProperties;
 import us.dot.its.jpo.ode.kafka.topics.Asn1CoderTopics;
 import us.dot.its.jpo.ode.kafka.topics.JsonTopics;
 import us.dot.its.jpo.ode.kafka.topics.PojoTopics;
@@ -57,8 +57,6 @@
 import us.dot.its.jpo.ode.util.JsonUtils;
 import us.dot.its.jpo.ode.util.JsonUtils.JsonUtilsException;
 import us.dot.its.jpo.ode.util.XmlUtils;
-import us.dot.its.jpo.ode.wrapper.MessageProducer;
-import us.dot.its.jpo.ode.wrapper.serdes.OdeTimSerializer;
 
 /**
  * The REST controller for handling TIM creation requests.
@@ -80,8 +78,8 @@ public class TimDepositController {
   private final SerialId serialIdJ2735;
   private final SerialId serialIdOde;
 
-  private final MessageProducer<String, String> stringMsgProducer;
-  private final MessageProducer<String, OdeObject> timProducer;
+  private final KafkaTemplate<String, String> kafkaTemplate;
+  private final KafkaTemplate<String, OdeObject> timDataKafkaTemplate;
 
   private final boolean dataSigningEnabledSDW;
 
@@ -102,12 +100,13 @@ public TimDepositControllerException(String errMsg) {
    * Spring Autowired constructor for the REST controller to properly initialize.
    */
   @Autowired
-  public TimDepositController(OdeKafkaProperties odeKafkaProperties,
-                              Asn1CoderTopics asn1CoderTopics,
+  public TimDepositController(Asn1CoderTopics asn1CoderTopics,
                               PojoTopics pojoTopics,
                               JsonTopics jsonTopics,
                               TimIngestTrackerProperties ingestTrackerProperties,
-                              SecurityServicesProperties securityServicesProperties) {
+                              SecurityServicesProperties securityServicesProperties,
+                              KafkaTemplate<String, String> kafkaTemplate,
+                              KafkaTemplate<String, OdeObject> timDataKafkaTemplate) {
     super();
 
     this.asn1CoderTopics = asn1CoderTopics;
@@ -116,12 +115,8 @@ public TimDepositController(OdeKafkaProperties odeKafkaProperties,
     this.serialIdJ2735 = new SerialId();
     this.serialIdOde = new SerialId();
 
-    this.stringMsgProducer =
-        MessageProducer.defaultStringMessageProducer(odeKafkaProperties.getBrokers(),
-            odeKafkaProperties.getKafkaType(), odeKafkaProperties.getDisabledTopics());
-    this.timProducer = new MessageProducer<>(odeKafkaProperties.getBrokers(),
-        odeKafkaProperties.getKafkaType(), null,
-        OdeTimSerializer.class.getName(), odeKafkaProperties.getDisabledTopics());
+    this.kafkaTemplate = kafkaTemplate;
+    this.timDataKafkaTemplate = timDataKafkaTemplate;
 
     this.dataSigningEnabledSDW = securityServicesProperties.getIsSdwSigningEnabled();
 
@@ -247,10 +242,10 @@ public synchronized ResponseEntity<String> depositTim(String jsonString, Request
     }
 
     OdeTimData odeTimData = new OdeTimData(timMetadata, timDataPayload);
-    timProducer.send(pojoTopics.getTimBroadcast(), null, odeTimData);
+    timDataKafkaTemplate.send(pojoTopics.getTimBroadcast(), null, odeTimData);
 
     String obfuscatedTimData = TimTransmogrifier.obfuscateRsuPassword(odeTimData.toJson());
-    stringMsgProducer.send(jsonTopics.getTimBroadcast(), null, obfuscatedTimData);
+    kafkaTemplate.send(jsonTopics.getTimBroadcast(), serialIdJ2735.getStreamId(), obfuscatedTimData);
 
     // Now that the message has been published to OdeBroadcastTim topic, it should
     // be
@@ -311,13 +306,13 @@ public synchronized ResponseEntity<String> depositTim(String jsonString, Request
 
         String obfuscatedJ2735Tim = TimTransmogrifier.obfuscateRsuPassword(j2735Tim);
         // publish Broadcast TIM to a J2735 compliant topic.
-        stringMsgProducer.send(jsonTopics.getJ2735TimBroadcast(), null, obfuscatedJ2735Tim);
+        kafkaTemplate.send(jsonTopics.getJ2735TimBroadcast(), serialIdJ2735.getStreamId(), obfuscatedJ2735Tim);
         // publish J2735 TIM also to general un-filtered TIM topic with streamID as key
-        stringMsgProducer.send(jsonTopics.getTim(), serialIdJ2735.getStreamId(),
-            obfuscatedJ2735Tim); // Write XML to the encoder input topic at the end to ensure the correct order
+        kafkaTemplate.send(jsonTopics.getTim(), serialIdJ2735.getStreamId(), obfuscatedJ2735Tim);
+        // Write XML to the encoder input topic at the end to ensure the correct order
         // of operations to pair
         // each message to an OdeTimJson streamId key
-        stringMsgProducer.send(asn1CoderTopics.getEncoderInput(), null, xmlMsg);
+        kafkaTemplate.send(asn1CoderTopics.getEncoderInput(), serialIdJ2735.getStreamId(), xmlMsg);
       }
 
       serialIdOde.increment();