Switch label-owner-service to fetch raw message

PawelPacholek · PawelPacholek · commit e58684c0fbae · 2025-05-09T10:39:30.000+02:00
diff --git a/label-owner-service/api/src/main/java/com/label_owner_service/api/config/ApiConfiguration.java b/label-owner-service/api/src/main/java/com/label_owner_service/api/config/ApiConfiguration.java
@@ -37,7 +37,8 @@ public PubSubInboundChannelAdapter inboundChannelAdapter(
     );
     adapter.setOutputChannel(messageChannel);
     adapter.setAckMode(AckMode.MANUAL);
-    adapter.setPayloadType(InitialOwner.class);
+    adapter.setPayloadType(String.class);
+    //adapter.setPayloadType(InitialOwner.class);
     return adapter;
   }
 
diff --git a/label-owner-service/api/src/main/java/com/label_owner_service/api/listeners/DataClassSerialization.java b/label-owner-service/api/src/main/java/com/label_owner_service/api/listeners/DataClassSerialization.java
@@ -0,0 +1,31 @@
+package com.label_owner_service.api.listeners;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.json.JsonMapper;
+
+import java.util.concurrent.Callable;
+
+public class DataClassSerialization {
+
+  static final ObjectMapper MAPPER = JsonMapper.builder().build();
+
+  public static String serialize(Object input) {
+    return unchecked(() -> MAPPER.writeValueAsString(input));
+  }
+
+  public static <T> T deserialize(String source, Class<T> type) {
+    return unchecked(() -> MAPPER.readValue(source, type));
+  }
+
+  private static <T> T unchecked(Callable<T> callable) {
+    try {
+      return callable.call();
+    } catch (RuntimeException e) {
+      throw e;
+  }   catch (Exception e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+
+}
diff --git a/label-owner-service/api/src/main/java/com/label_owner_service/api/listeners/LabelOwnerListener.java b/label-owner-service/api/src/main/java/com/label_owner_service/api/listeners/LabelOwnerListener.java
@@ -47,13 +47,22 @@ public LabelOwnerListener(LabelOwnerUseCase labelOwnerUseCase) {
 
     @ServiceActivator(inputChannel = "initialOwnerChannel")
     public void messageReceiver(
-            InitialOwner initialOwner,
+      String rawInitialOwnerMessage,
+            //InitialOwner initialOwner,
             @Header(GcpPubSubHeaders.ORIGINAL_MESSAGE) BasicAcknowledgeablePubsubMessage message
     ) {
+        InitialOwner initialOwner = parseOwner(rawInitialOwnerMessage);
         labelOwnerUseCase.labelAndSendOwner(initialOwner);
         message.ack();
     }
 
+    private InitialOwner parseOwner(String rawInitialOwnerMessage) {
+      String initialOwnerJson = rawInitialOwnerMessage
+        .replaceAll("^.*payload=\\{", "{")
+        .replaceAll("}, headers=.*$", "}");
+      return DataClassSerialization.deserialize(initialOwnerJson, InitialOwner.class);
+    }
+
 
 
   /*

Original file line number	Diff line number	Diff line change
`@@ -37,7 +37,8 @@ public PubSubInboundChannelAdapter inboundChannelAdapter(`
`37`	`37`	`);`
`38`	`38`	`adapter.setOutputChannel(messageChannel);`
`39`	`39`	`adapter.setAckMode(AckMode.MANUAL);`
`40`		`- adapter.setPayloadType(InitialOwner.class);`
	`40`	`+ adapter.setPayloadType(String.class);`
	`41`	`+ //adapter.setPayloadType(InitialOwner.class);`
`41`	`42`	`return adapter;`
`42`	`43`	`}`
`43`	`44`