Merge pull request #18 from btison/outbox

ERDEMO-105: use outbox for all Kafka messages from process service

pom.xml

@@ -19,6 +19,7 @@
     <quarkus.platform.version>1.10.3.Final</quarkus.platform.version>
     <smallrye.version>2.5.0</smallrye.version>
     <surefire-plugin.version>3.0.0-M5</surefire-plugin.version>
+    <commons-text.version>1.8</commons-text.version>
   </properties>
   <dependencyManagement>
     <dependencies>
@@ -80,6 +81,11 @@
       <groupId>io.micrometer</groupId>
       <artifactId>micrometer-registry-prometheus</artifactId>
     </dependency>
+    <dependency>
+      <groupId>org.apache.commons</groupId>
+      <artifactId>commons-text</artifactId>
+      <version>${commons-text.version}</version>
+    </dependency>
     <dependency>
       <groupId>io.quarkus</groupId>
       <artifactId>quarkus-junit5</artifactId>

src/main/java/com/redhat/emergency/response/incident/consumer/serialization/QuotedStringDeserializer.java

@@ -0,0 +1,47 @@
+package com.redhat.emergency.response.incident.consumer.serialization;
+
+import java.io.UnsupportedEncodingException;
+import java.util.Map;
+
+import org.apache.commons.text.StringEscapeUtils;
+import org.apache.kafka.common.errors.SerializationException;
+import org.apache.kafka.common.serialization.Deserializer;
+
+public class QuotedStringDeserializer implements Deserializer<String> {
+
+    private String encoding = "UTF8";
+
+    @Override
+    public void configure(Map<String, ?> configs, boolean isKey) {
+        String propertyName = isKey ? "key.deserializer.encoding" : "value.deserializer.encoding";
+        Object encodingValue = configs.get(propertyName);
+        if (encodingValue == null)
+            encodingValue = configs.get("deserializer.encoding");
+        if (encodingValue instanceof String)
+            encoding = (String) encodingValue;
+    }
+
+    @Override
+    public String deserialize(String topic, byte[] data) {
+        try {
+            if (data == null) {
+                return null;
+            }
+            else {
+                String deserialized = new String(data, encoding);
+                if (deserialized.startsWith("\"")) {
+                    String unescaped = StringEscapeUtils.unescapeJson(deserialized);
+                    return unescaped.substring(1, unescaped.length()-1);
+                }
+                return deserialized;
+            }
+        } catch (UnsupportedEncodingException e) {
+            throw new SerializationException("Error when deserializing byte[] to string due to unsupported encoding " + encoding);
+        }
+    }
+
+    @Override
+    public void close() {
+        // nothing to do
+    }
+}

src/main/resources/application.properties

@@ -15,7 +15,7 @@ quarkus.kafka.health.enabled=true
 # Configure the Kafka source
 mp.messaging.incoming.incident-command.connector=smallrye-kafka
 mp.messaging.incoming.incident-command.key.deserializer=org.apache.kafka.common.serialization.StringDeserializer
-mp.messaging.incoming.incident-command.value.deserializer=org.apache.kafka.common.serialization.StringDeserializer
+mp.messaging.incoming.incident-command.value.deserializer=com.redhat.emergency.response.incident.consumer.serialization.QuotedStringDeserializer
 mp.messaging.incoming.incident-command.cloud-events=true
 mp.messaging.incoming.incident-command.request.timeout.ms=30000
 mp.messaging.incoming.incident-command.enable.auto.commit=false