add CheckLinkProcessor

kontor-scripts/add_link_stomp.py

@@ -15,7 +15,7 @@ parser.add_argument("--config", "-c", default="kontor-docker")
 parser.add_argument("--verbose", "-v", action="count", default=0)
 parser.add_argument("--server", "-s", default="127.0.0.1")
 parser.add_argument("--port", "-p", default="61616")
-parser.add_argument("--destination", "-d", default="media.link.add")
+parser.add_argument("--destination", "-d", default="media.link")
 args = parser.parse_args()
 
 
@@ -33,6 +33,6 @@ if __name__ == "__main__":
 
     link: Link = Link(url=args.url)
     json_bytes = msgspec.json.encode(link)
-    conn.send(body=json_bytes, destination=args.destination)
+    conn.send(body=json_bytes.decode(), destination=args.destination)
 
     logger.info("kontor.add_link finished")

kontor-spring/src/main/java/de/thpeetz/kontor/integration/routes/QueueMediaLink.java

@@ -0,0 +1,42 @@
+package de.thpeetz.kontor.integration.routes;
+
+import org.apache.camel.builder.RouteBuilder;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import de.thpeetz.kontor.integration.services.CheckLinkProcessor;
+import de.thpeetz.kontor.services.MediaFileService;
+
+@Component
+public class QueueMediaLink extends RouteBuilder {
+
+    @Autowired
+    private final MediaFileService mediaFileService;
+
+    @Autowired
+    public QueueMediaLink(MediaFileService mediaFileService) {
+        this.mediaFileService = mediaFileService;
+    }
+
+    @Override
+    public void configure() throws Exception {
+        from("jms:queue:media.link")
+            .errorHandler(deadLetterChannel("jms:queue:DLQ")
+                .maximumRedeliveries(0)
+                .useOriginalMessage()
+                .onPrepareFailure(exchange -> {
+                    exchange.getIn().setHeader("FailureReason", "processing failed");
+                })
+            )
+            .routeId("media.link")
+            .log("${body}")
+            .trace(true)
+            .process(new CheckLinkProcessor(mediaFileService))
+            .choice()
+                .when(exchangeProperty("linkId").isNotNull())
+                    .to("jms:queue:media.link.add")
+                .otherwise()
+                    .to("jms:queue:DLQ")
+            .to("jms:queue:media.link.add.processed");
+    }
+}

kontor-spring/src/main/java/de/thpeetz/kontor/integration/services/CheckLinkProcessor.java

@@ -0,0 +1,50 @@
+package de.thpeetz.kontor.integration.services;
+
+// import java.nio.charset.StandardCharsets;
+// import java.util.HashMap;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.json.simple.JSONObject;
+import org.json.simple.parser.JSONParser;
+import org.json.simple.parser.ParseException;
+
+// import com.fasterxml.jackson.core.type.TypeReference;
+// import com.fasterxml.jackson.databind.ObjectMapper;
+
+import de.thpeetz.kontor.services.MediaFileService;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class CheckLinkProcessor implements Processor {
+
+    private final MediaFileService mediaFileService;
+
+    public CheckLinkProcessor(MediaFileService mediaFileService) {
+        this.mediaFileService = mediaFileService;
+    }
+
+    @Override
+    public void process(Exchange exchange) throws Exception {
+        String messageBody = exchange.getIn().getBody(String.class);
+        log.info("message body: {}", messageBody);
+
+        // ObjectMapper objectMapper = new ObjectMapper();
+        // HashMap<String, String> myMap = objectMapper.readValue(messageBody, new TypeReference<HashMap<String, String>>() {});
+        // String url = myMap.get("url");
+        // log.info("found url: {}", url);
+        // exchange.getIn().setHeader("linkId", null);
+        // exchange.setProperty("linkId", null);
+
+        JSONParser parser = new JSONParser();
+        try {
+            JSONObject jsonObject = (JSONObject) parser.parse(messageBody);
+            String url = (String)jsonObject.get("url");
+            log.info("found url: {}", url);
+            exchange.getIn().setHeader("linkId", null);
+        } catch (ParseException pe) {
+            log.info("parse exception: {}", pe.toString());
+            exchange.getIn().setHeader("linkId", null);
+        }
+    }
+}