diff --git a/kontor-scripts/add_link_stomp.py b/kontor-scripts/add_link_stomp.py index 025093f..f74b69d 100644 --- a/kontor-scripts/add_link_stomp.py +++ b/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") diff --git a/kontor-spring/src/main/java/de/thpeetz/kontor/integration/routes/QueueMediaLink.java b/kontor-spring/src/main/java/de/thpeetz/kontor/integration/routes/QueueMediaLink.java new file mode 100644 index 0000000..74fba5a --- /dev/null +++ b/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"); + } +} diff --git a/kontor-spring/src/main/java/de/thpeetz/kontor/integration/services/CheckLinkProcessor.java b/kontor-spring/src/main/java/de/thpeetz/kontor/integration/services/CheckLinkProcessor.java new file mode 100644 index 0000000..8d82795 --- /dev/null +++ b/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 myMap = objectMapper.readValue(messageBody, new TypeReference>() {}); + // 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); + } + } +}