package io.muoncore.extension.amqp;

import io.muoncore.transport.TransportInboundMessage;
import io.muoncore.transport.TransportMessage;
import io.muoncore.transport.TransportOutboundMessage;

import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.stream.Collectors;

public class AmqpMessageTransformers {

    public static QueueListener.QueueMessage outboundToQueue(String queue, TransportOutboundMessage message) {

        Map<String, String> headers = new HashMap<>();
        headers.putAll(message.getMetadata());
        headers.put("id", message.getId());
        headers.put("sourceService", message.getSourceServiceName());
        headers.put("targetService", message.getTargetServiceName());
        headers.put("protocol", message.getProtocol());

        String delimitedContentTypes = message.getSourceAvailableContentTypes().stream()
                .collect(Collectors.joining("|"));
        headers.put("sourceAvailableContentTypes", delimitedContentTypes);
        headers.put("channelOperation", message.getChannelOperation().toString());

        return new QueueListener.QueueMessage(message.getType(), queue, message.getPayload(), headers, message.getContentType());
    }

    public static TransportInboundMessage queueToInbound(QueueListener.QueueMessage message) {
        Map metadata = new HashMap<>();
        metadata.putAll(message.getHeaders());

        String targetService = "unknown";

        if (message.getHeaders().get("targetService") != null) {
            targetService = message.getHeaders().get("targetService");
        }

        return new TransportInboundMessage(
                message.getEventType(),
                message.getHeaders().get("id").toString(),
                targetService,
                message.getHeaders().get("sourceService").toString(),
                message.getHeaders().get("protocol").toString(),
                metadata,
                message.getContentType(), message.getBody(),
                Arrays.asList(message.getHeaders().get("sourceAvailableContentTypes").split("|")),
                TransportMessage.ChannelOperation.valueOf(message.getHeaders().get("channelOperation")));
    }

}

@@ -281,66 +281,222 @@
 til.
-stream.Collectors;%0A%0Apublic class AmqpMessageTransformers %7B
+logging.Level;%0Aimport java.util.logging.Logger;%0Aimport java.util.stream.Collectors;%0A%0Apublic class AmqpMessageTransformers %7B%0A%0A    private static Logger LOG = Logger.getLogger(ServiceQueue.class.getCanonicalName());%0A
 %0A%0A  
@@ -1610,16 +1610,654 @@
 nown%22;%0A%0A
+        if (!isvalid(%22id%22, message)) %7B%0A            LOG.log(Level.SEVERE, %22Header missing id%22);%0A        %7D%0A        if (!isvalid(%22sourceService%22, message)) %7B%0A            LOG.log(Level.SEVERE, %22Header missing sourceService%22);%0A        %7D%0A        if (!isvalid(%22protocol%22, message)) %7B%0A            LOG.log(Level.SEVERE, %22Header missing protocol%22);%0A        %7D%0A        if (!isvalid(%22sourceAvailableContentTypes%22, message)) %7B%0A            LOG.log(Level.SEVERE, %22Header missing sourceAvailableContentTypes%22);%0A        %7D%0A        if (!isvalid(%22channelOperation%22, message)) %7B%0A            LOG.log(Level.SEVERE, %22Header missing channelOperation%22);%0A        %7D%0A%0A
         
@@ -3002,11 +3002,154 @@
 ;%0A    %7D%0A
+    %0A    static boolean isvalid(String name, QueueListener.QueueMessage message) %7B%0A        return message.getHeaders().get(name) != null;%0A    %7D
 %0A%7D%0A

Add simple validation