Repository: Vantiq/vantiq-extension-sources (language: python)

Issue:
camelComponent: Add equivalency definitions so that we can augment component headers with expected names
Many camel components provide headers that include common information, unfortunately using component specific header names.  For example, different components have headers that contain the content type (_e.g._, `image/jpeg`, `application/json`), but each component seems to provide this in a header named in a component specific manner. 

The Vantiq component can provide this information to/from Vantiq, but the Vantiq component cannot _know_ the "other end" of the Camel app to which it is connected, or the specific header name that that other component uses.

Instead, we propose offering a source configuration option that says to look for header XXX and add another header with that value before sending it back to Vantiq.  Or, augment the headers, adding XXX from YYY before sending to the camel app.  In this way, the creator of the connector source can define this behavior, and their Vantiq app can find these header values (_e.g._, contentType) in a place that they understand.


Discussion / hints:

Fix patch:
diff --git a/camelComponent/README.md b/camelComponent/README.md
index 9e258d1c..74d9855d 100644
--- a/camelComponent/README.md
+++ b/camelComponent/README.md
@@ -94,6 +94,54 @@ By adding the `structuredMessageHeader` component endpoint option with a value o
 
 A schema type definition for this message structure is available in `src/main/resources/types/com/vantiq/extsrc/camelcomp/message.json`.
 
+#### Header Duplication
+
+There are numerous components in the Apache Camel universe, and many include Camel Headers that convey the same 
+information. For example, a number of components include headers whose name is of the form `Camel<componentn 
+name>ContentType` (_e.g._, `CamelAzureServiceBusContentType` to name but one). 
+
+If the Vantiq application wants to deal with these, it does not want to know all possible variations here -- that is,
+after all, the purpose of an integration system: to manage these discrepancies.  To ease the handling of these 
+situations, the Vantiq Camel Component can duplicate (NOT replace) these header values into alternative names. When 
+this is requested, the original set of headers will be sent or received, but the duplicates with values will be sent 
+or received as well.
+
+To do this, the Camel system must be configured with a Bean of type `io.vantiq.extsrc.camel.HeaderDuplicationBean`, 
+containing a Map<String, String> where the keys of the map represent the header names to be duplicated, and the 
+values are the new header names into which the header value will be duplicated. (Note: we do this via a configured 
+bean since the number of headers could be large, and having URI of great length is sometimes problematic.)
+
+To use this, add the `headerDuplicationBeanName` component/endpoint option with the value being the name of the bean 
+containing the map described above. Note that this header duplication is performed only when the URI parameter 
+`structuredMessageHeader` is present with the value `true`. In other cases, it is silently ignored.
+
+For example, using a Java example, if you want to duplicate the `CamelComponentContentType` header into the 
+header `VantiqContentType`, we would configure things as follows:
+
+```java
+    ...
+    HeaderDuplicationBean hdBean = new HeaderDuplicationBean();
+    Map<String, String> hdMap = Map.of("CamelComponentContentType", "VantiqContentType");
+    hdBean.setHeaderDuplicationMap(hdMap);
+    context.getRegistry().bind("someBeanName", hdBean);
+    
+    // Now, a route endpoint could be defined such as
+    String vantiqEndpoint = "vantiq://<host>:<port>/&sourceName=xxx?accessToken=YYY" +
+        "?structuredMessageHeader=true" +
+        "?headerDuplicationBeanName=someBeanName"
+```
+
+When this definition is in place, messages arriving (via the Azure ServiceBus) to that endpoint will contain the 
+headers `CamelComponentContentType` & `VantiqContentType` with the same values. If the `CamelComponentContentType` 
+header is not present, no duplication will happen.
+
+Note that if multiple headers are specified to duplicate to a single duplicate (e.g., both header foo & bar
+are both supposed to be duplicated to the header baz), the results are unpredictable.  Something will happen, 
+but the results depend upon the order on which the headers are processed, and may vary from instance to
+instance.  Callers should take care to avoid these situations.  There may be cases where components have 
+alternatives that are used, both of which are desired to duplicate to the same header, so we allow such a 
+definition. But the caller should take care with such definitions.
+
 ## Exchanges to Vantiq
 
 The Vantiq Component expects messages in an exchange to arrive in the form of a Java Map.  However, messages arriving
diff --git a/camelComponent/src/main/java/io/vantiq/extsrc/camel/HeaderDuplicationBean.java b/camelComponent/src/main/java/io/vantiq/extsrc/camel/HeaderDuplicationBean.java
new file mode 100644
index 00000000..12638e76
--- /dev/null
+++ b/camelComponent/src/main/java/io/vantiq/extsrc/camel/HeaderDuplicationBean.java
@@ -0,0 +1,24 @@
+/*
+ * Copyright (c) 2023 Vantiq, Inc.
+ *
+ * All rights reserved.
+ *
+ * SPDX: MIT
+ */
+
+package io.vantiq.extsrc.camel;
+
+import java.util.Map;
+
+public class HeaderDuplicationBean {
+    
+    private Map<String, String> headerDuplicationMap;
+    
+    public Map<String, String> getHeaderDuplicationMap() {
+        return headerDuplicationMap;
+    }
+    
+    public void setHeaderDuplicationMap(Map<String, String> heMap) {
+        this.headerDuplicationMap = heMap;
+    }
+}
diff --git a/camelComponent/src/main/java/io/vantiq/extsrc/camel/VantiqConsumer.java b/camelComponent/src/main/java/io/vantiq/extsrc/camel/VantiqConsumer.java
index fb043eb5..643fb0e4 100644
--- a/camelComponent/src/main/java/io/vantiq/extsrc/camel/VantiqConsumer.java
+++ b/camelComponent/src/main/java/io/vantiq/extsrc/camel/VantiqConsumer.java
@@ -19,7 +19,6 @@
 import io.vantiq.extjsdk.ExtensionWebSocketClient;
 import io.vantiq.extjsdk.Handler;
 
-import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.ExecutorService;
 import java.util.stream.Collectors;
@@ -106,7 +105,10 @@ private void processMessage(Object msg, ExchangePattern pattern) {
                                 .stream()
                                 .filter(e -> e.getKey() instanceof String)
                                 .collect(Collectors.toMap(e -> (String) e.getKey(), Map.Entry::getValue));
-
+                    // If endpoint says to copy header values, do that here now that we have the values.
+                    if (camelHdrs.size() > 0) {
+                        camelHdrs.putAll(endpoint.duplicateHeaders(camelHdrs));
+                    }
                 }
                 if (msgAsMap.get(STRUCTURED_MESSAGE_MESSAGE_PROPERTY) != null) {
                     camelBody = msgAsMap.get(STRUCTURED_MESSAGE_MESSAGE_PROPERTY);
diff --git a/camelComponent/src/main/java/io/vantiq/extsrc/camel/VantiqEndpoint.java b/camelComponent/src/main/java/io/vantiq/extsrc/camel/VantiqEndpoint.java
index 400904d7..fc6249e2 100644
--- a/camelComponent/src/main/java/io/vantiq/extsrc/camel/VantiqEndpoint.java
+++ b/camelComponent/src/main/java/io/vantiq/extsrc/camel/VantiqEndpoint.java
@@ -34,6 +34,7 @@
 
 import java.net.URI;
 import java.net.URISyntaxException;
+import java.util.HashMap;
 import java.util.Map;
 import java.util.Properties;
 import java.util.concurrent.CompletableFuture;
@@ -96,6 +97,20 @@ public class VantiqEndpoint extends DefaultEndpoint {
     @Setter
     private boolean structuredMessageHeader;
     
+    // Note doing this as a constructed bean rather than a multivalue parameter because the number of headers might
+    // be large, and a large set thereof may violate URI length requirements. Also, it's primarily intended for use
+    // in the Camel Connector, so constructing such a things there based on source parameters is not hard.  Should a
+    // need arise, we can offer another option(s).
+    public static final String HEADER_DUPLICATION_BEAN_NAME = "headerDuplicationBeanName";
+    
+    @UriParam(name = HEADER_DUPLICATION_BEAN_NAME, description = "A bean name where the bean contains a " +
+            "headerDuplicationMap value mapping header names to a duplicate header into which the value should be " +
+            "copied.")
+    @Metadata(required = false)
+    @Getter
+    @Setter
+    private String headerDuplicationBeanName;
+    
     @Setter
     @Getter
     private ExtensionWebSocketClient vantiqClient;
@@ -110,6 +125,9 @@ public class VantiqEndpoint extends DefaultEndpoint {
     @Getter
     private String endpointName;
     
+    @Getter
+    private Map<String, String> headerDuplicationMap;
+    
     InstanceConfigUtils utils;
 
     public VantiqEndpoint() {
@@ -214,6 +232,20 @@ public void startup() throws CamelException {
             }
             CamelException failure = null;
             try {
+                if (headerDuplicationBeanName != null &&
+                        !headerDuplicationBeanName.isEmpty() && !headerDuplicationBeanName.isBlank()) {
+                    // If we have header equivalents specified, fetch them and populate our local store for our consumers &
+                    // producers
+                    HeaderDuplicationBean hdBean = getCamelContext().getRegistry().lookupByNameAndType(
+                            headerDuplicationBeanName,
+                            HeaderDuplicationBean.class);
+                    if (hdBean == null) {
+                        throw new IllegalArgumentException("No HeaderDuplicationBean named " + headerDuplicationBeanName +
+                                                                   " was found in the Camel Registry.");
+                    } else {
+                        headerDuplicationMap = hdBean.getHeaderDuplicationMap();
+                    }
+                }
                 log.debug("Attempting to connect to URL: {} from {}", getEndpointBaseUri(), getEndpointUri());
                 String vtq = getEndpointBaseUri();
                
@@ -367,6 +399,34 @@ public static String adjustVantiqTarget(String baseUrl, boolean noSsl) throws Ca
         return correctedUrl;
     }
     
+    /**
+     * Return map of headers duplicated as specified by the HEADER_EQUIVALENCE_BEAN_NAME parameter.
+     * <p>
+     * If no duplication is specified or nothing found to duplicate, return an empty map.
+     * <p>
+     * Note that if multiple headers are specified to duplicate to a single duplicate (e.g., both header foo & bar
+     * are both supposed to be duplicated to the header baz), the results are unpredictable.  Something will happen,
+     * but the results depend upon the order on which the headers are processed, and may vary from instance to
+     * instance.  Callers should take care to avoid these situations.  There may be cases where components have
+     * alternatives that are used, both of which are desired to duplicate to the same header, so we allow such a
+     * definition.
+     *
+     * @param headers Map<String, Object> specifying the headers supplied by Camel
+     * @return Map<String, Object> specifying the duplicated headers.  Returns an empty map of nothing to duplicated
+     */
+    public Map<String, Object> duplicateHeaders(Map<String, Object> headers) {
+        Map<String, Object> dupedHdrs = new HashMap<>();
+        if (headerDuplicationMap != null && headerDuplicationMap.size() > 0) {
+            headerDuplicationMap.forEach((k, v) -> {
+                if (headers.get(k) != null) {
+                    // If we have a value for a header we're expecting to duplicate, then we duplicate that
+                    // value into the duplicated header's name.
+                    dupedHdrs.put(v, headers.get(k));
+                }
+            });
+        }
+        return dupedHdrs;
+    }
     @Override
     public void doStop() {
         synchronized (startStopLock) {
diff --git a/camelComponent/src/main/java/io/vantiq/extsrc/camel/VantiqProducer.java b/camelComponent/src/main/java/io/vantiq/extsrc/camel/VantiqProducer.java
index 5995b80f..e5569164 100644
--- a/camelComponent/src/main/java/io/vantiq/extsrc/camel/VantiqProducer.java
+++ b/camelComponent/src/main/java/io/vantiq/extsrc/camel/VantiqProducer.java
@@ -17,7 +17,6 @@
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.SerializationFeature;
-import com.fasterxml.jackson.databind.exc.MismatchedInputException;
 import com.fasterxml.jackson.databind.node.ArrayNode;
 import com.fasterxml.jackson.databind.node.BinaryNode;
 import com.fasterxml.jackson.databind.node.BooleanNode;
@@ -202,6 +201,7 @@ public void process(Exchange exchange) throws Exception {
                 }
             }
             if (hdrs != null) {
+                hdrs.putAll(endpoint.duplicateHeaders(hdrs));
                 fmtMsg.put(STRUCTURED_MESSAGE_HEADERS_PROPERTY, hdrs);
             }
             Map<String, Object> m = mapper.convertValue(vMsg, new TypeReference<>() {});


Test patch:
diff --git a/camelComponent/src/test/java/io/vantiq/extsrc/camel/VantiqComponentTest.java b/camelComponent/src/test/java/io/vantiq/extsrc/camel/VantiqComponentTest.java
index a33c3e40..00e29c54 100644
--- a/camelComponent/src/test/java/io/vantiq/extsrc/camel/VantiqComponentTest.java
+++ b/camelComponent/src/test/java/io/vantiq/extsrc/camel/VantiqComponentTest.java
@@ -10,6 +10,7 @@
 
 import static io.vantiq.extsrc.camel.VantiqEndpoint.ACCESS_TOKEN_PARAM;
 import static io.vantiq.extsrc.camel.VantiqEndpoint.CONSUMER_OUTPUT_JSON_PARAM;
+import static io.vantiq.extsrc.camel.VantiqEndpoint.HEADER_DUPLICATION_BEAN_NAME;
 import static io.vantiq.extsrc.camel.VantiqEndpoint.SOURCE_NAME_PARAM;
 import static io.vantiq.extsrc.camel.VantiqEndpoint.STRUCTURED_MESSAGE_HEADER_PARAM;
 import static io.vantiq.extsrc.camel.VantiqEndpoint.STRUCTURED_MESSAGE_HEADERS_PROPERTY;
@@ -17,10 +18,8 @@
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
-import com.fasterxml.jackson.databind.node.ArrayNode;
 import com.fasterxml.jackson.databind.node.BaseJsonNode;
 import com.fasterxml.jackson.databind.node.ObjectNode;
-import com.fasterxml.jackson.databind.node.TextNode;
 import io.vantiq.extjsdk.ExtensionServiceMessage;
 import io.vantiq.extjsdk.FalseClient;
 import io.vantiq.extjsdk.FalseWebSocket;
@@ -55,10 +54,14 @@ public class VantiqComponentTest extends CamelTestSupport {
     private static final String accessToken = "someAccessToken";
     private static final String TEST_MSG_PREAMBLE = "published message from FauxVantiq #";
     private static final String TEST_MSG_KEY = "someRandomKey";
+    private static final String TEST_HEADER_BEAN_NAME = "TestHEBean";
     
     private final String routeStartUri = "direct:start";
     
     private final String routeStartStructuredUri = "direct:structuredstart";
+    
+    private final String routeStartStructuredHeaderMapUri = "direct:structuredMappedstart";
+    
     private final String routeEndUri = "mock:direct:result";
     
     private final String exceptionEndpoint = "mock:direct:error";
@@ -71,6 +74,12 @@ public class VantiqComponentTest extends CamelTestSupport {
             "&" + ACCESS_TOKEN_PARAM + "=" + accessToken +
             "&" + STRUCTURED_MESSAGE_HEADER_PARAM + "=true";
     
+    private final String vantiqEndpointStructuredHeaderMapUri = "vantiq://doesntmatter/" +
+            "?" + SOURCE_NAME_PARAM + "=" + testSourceName +
+            "&" + ACCESS_TOKEN_PARAM + "=" + accessToken +
+            "&" + STRUCTURED_MESSAGE_HEADER_PARAM + "=true" +
+            "&" + HEADER_DUPLICATION_BEAN_NAME + "=" + TEST_HEADER_BEAN_NAME;
+    
     private final String vantiqSenderUri = "vantiq://senderdoesntmatter/" +
             "?" + SOURCE_NAME_PARAM + "=" + testSourceName +
             "&" + ACCESS_TOKEN_PARAM + "=" + accessToken;
@@ -90,6 +99,13 @@ public class VantiqComponentTest extends CamelTestSupport {
             "&" + CONSUMER_OUTPUT_JSON_PARAM + "=true" +
             "&" + STRUCTURED_MESSAGE_HEADER_PARAM + "=true";
     
+    public final String vantiqStructuredJsonSenderMappedUri = "vantiq://structuredsenderdoesntmatter/" +
+            "?" + SOURCE_NAME_PARAM + "=" + testSourceName +
+            "&" + ACCESS_TOKEN_PARAM + "=" + accessToken +
+            "&" + CONSUMER_OUTPUT_JSON_PARAM + "=true" +
+            "&" + STRUCTURED_MESSAGE_HEADER_PARAM + "=true" +
+            "&" + HEADER_DUPLICATION_BEAN_NAME + "=" + TEST_HEADER_BEAN_NAME;
+    
     private final String vantiqJsonQuerierUri = "vantiq://jsonquerierdoesntmatter/" +
             "?" + SOURCE_NAME_PARAM + "=" + testSourceName +
             "&" + ACCESS_TOKEN_PARAM + "=" + accessToken +
@@ -267,6 +283,87 @@ public void testVantiqProducerStructuredJson() {
         validateExtensionMsg(lastMsg, true, null, new String[] {"bye"}, "mom");
     }
     
+    @Test
+    public void testVantiqProducerHeaderMapping() {
+        // First, grab our test environment.
+        FauxVantiqComponent vc = (FauxVantiqComponent) context.getComponent("vantiq");
+        assert vc != null;
+        // Note that we need to fetch the endpoints by URI since there are more than one of them.
+        FauxVantiqEndpoint ep = (FauxVantiqEndpoint) context.getEndpoint(vantiqEndpointStructuredHeaderMapUri);
+        FalseClient fc = ep.myClient;
+        
+        ArrayList<String> msgs =new ArrayList<>();
+        int itemCount = 10;
+        for (int i = 0; i < itemCount; i++) {
+            msgs.add("{\"hi\": \"mom " + i + "\"}");
+        }
+        
+        for (Object item: msgs) {
+            log.info("Sending msg: " + item);
+            
+            assert item != null;
+            sendBody(routeStartStructuredHeaderMapUri, item);
+            
+            //noinspection rawtypes
+            Map lastMsg = fc.getLastMessageAsMap();
+            validateExtensionMsg(lastMsg, true, null, new String[] {"hi"}, "mom");
+        }
+        
+        // Now, try a similar test sending maps
+        ArrayList<Map<String, String>> mapList = new ArrayList<>();
+        for (int i = 0; i < itemCount; i++) {
+            Map<String, String> mp = Map.of("hi", "dad " + i);
+            mapList.add(mp);
+        }
+        
+        AtomicInteger hdrCounter = new AtomicInteger(0);
+        
+        mapList.forEach( msg -> {
+            Map<String, Object> hdrs = new HashMap<>();
+            hdrs.put("theMessage", msg);
+            hdrs.put("counter", hdrCounter.getAndIncrement());
+            hdrs.put("shouldBeNull", null);
+            hdrs.put("header1", "value1");
+            hdrs.put("header2", "value2");
+    
+            Map<String, Object> expHdrs = new HashMap<>(hdrs);
+            expHdrs.put("dupHeader1", hdrs.get("header1"));
+            expHdrs.put("dupHeader2", hdrs.get("header2"));
+            sendBody(routeStartStructuredHeaderMapUri, msg, hdrs);
+            
+            //noinspection rawtypes
+            Map lastMsg = fc.getLastMessageAsMap();
+            validateExtensionMsg(lastMsg, true, expHdrs, new String[] {"hi"}, "dad");
+        });
+        
+        ObjectMapper mapper = new ObjectMapper();
+        AtomicInteger counter = new AtomicInteger(0);
+        msgs.forEach( msg -> {
+            try {
+                ObjectNode node = mapper.readValue(msg, ObjectNode.class);
+                node.put("hi", "aki " + counter.getAndIncrement());
+                Map<String, Object> hdrs = new HashMap<>();
+                hdrs.put("theMessage", msg);
+                hdrs.put("counter", hdrCounter.getAndIncrement());
+                
+                sendBody(routeStartStructuredHeaderMapUri, node, hdrs);
+                
+                //noinspection rawtypes
+                Map lastMsg = fc.getLastMessageAsMap();
+                validateExtensionMsg(lastMsg, true, hdrs, new String[] {"hi"}, "aki");
+            } catch (JsonProcessingException e) {
+                throw new RuntimeException(e);
+            }
+        });
+        
+        sendBody(routeStartStructuredHeaderMapUri, "{\"bye\": \"mom\"}");
+        
+        //noinspection rawtypes
+        Map lastMsg = fc.getLastMessageAsMap();
+        validateExtensionMsg(lastMsg, true, null, new String[] {"bye"}, "mom");
+    }
+    
+    
     void validateExtensionMsg(Map<?,?> lastMsg, Boolean isStructured, Map<String, Object> expHdrs,
                               String[] msgKeys, String msgPreamble) {
         assert lastMsg.containsKey("op");
@@ -285,7 +382,7 @@ void validateExtensionMsg(Map<?,?> lastMsg, Boolean isStructured, Map<String, Ob
                         (Map<String, Object>) msg.get(STRUCTURED_MESSAGE_HEADERS_PROPERTY);
                 assert hdrs != null;
                 expHdrs.forEach( (key, value) -> {
-                    assert hdrs.containsKey(key);
+                    assertTrue("Missing header " + key, hdrs.containsKey(key));
                     log.debug("For key {}, comparing value {} ({}) with expected {} ({}).",
                               key, hdrs.get(key),
                               hdrs.get(key) != null ? hdrs.get(key).getClass().getName() : hdrs.get(key),
@@ -595,6 +692,136 @@ public void testVantiqConsumerOutputJsonStructured() throws Exception {
         assert msgsReceived.get() == mapMsgCount + extraTestMsgs.size();
     }
     
+    @Test
+    public void testVantiqConsumerHeaderMapping() throws Exception {
+        
+        // First, grab our test environment.
+        FauxVantiqComponent vc = (FauxVantiqComponent) context.getComponent("vantiq");
+        assert vc != null;
+        // Note that we need to fetch the endpoints by URI since there are more than one of them.
+        FauxVantiqEndpoint endp = (FauxVantiqEndpoint) context.getEndpoint(vantiqStructuredJsonSenderMappedUri);
+        
+        assert endp.myClient.getListener() instanceof TestListener;
+        TestListener tl = (TestListener) endp.myClient.getListener();
+        
+        int mapMsgCount = 10;
+        List<Object> extraTestMsgs = List.of("I am a test string", List.of(Map.of("test", "message", "another", "test" +
+                " message")));
+        
+        MockEndpoint mocked = getMockEndpoint(routeEndUri);
+        mocked.expectedMinimumMessageCount(mapMsgCount + extraTestMsgs.size());
+        VantiqEndpoint ve = getMandatoryEndpoint(vantiqEndpointUri, VantiqEndpoint.class);
+        assert ve != null;
+        ObjectMapper mapper = new ObjectMapper();
+        WebSocket ws = new FalseWebSocket();
+        for (int i = 0; i < mapMsgCount + extraTestMsgs.size(); i++) {
+            ExtensionServiceMessage ep = new ExtensionServiceMessage(vantiqEndpointUri);
+            ep.op = ExtensionServiceMessage.OP_PUBLISH;
+            ep.resourceName = "SOURCES";
+            ep.resourceId = testSourceName;
+            if (i < mapMsgCount) {
+                HashMap<String, Object> msg = new HashMap<>();
+                msg.put(TEST_MSG_KEY, TEST_MSG_PREAMBLE + i);
+                HashMap<String, Object> smsg = new HashMap<>();
+                smsg.put(STRUCTURED_MESSAGE_MESSAGE_PROPERTY, msg);
+                HashMap<String, Object> hdrs = new HashMap<>();
+                hdrs.put("theMessage", msg);
+                hdrs.put("counter", i);
+                smsg.put(STRUCTURED_MESSAGE_HEADERS_PROPERTY, hdrs);
+                ep.object = smsg;
+            } else {
+                HashMap<String, Object> smsg = new HashMap<>();
+                HashMap<String, Object> hdrs = new HashMap<>();
+                hdrs.put("theMessage", extraTestMsgs.get(i - mapMsgCount));
+                hdrs.put("counter", i);
+                hdrs.put("isExtra", true);
+                hdrs.put("header1", "header1Value");
+                hdrs.put("header2", "header2Value");
+                smsg.put(STRUCTURED_MESSAGE_HEADERS_PROPERTY, hdrs);
+                smsg.put(STRUCTURED_MESSAGE_MESSAGE_PROPERTY, extraTestMsgs.get(i - mapMsgCount));
+                ep.object = smsg;
+            }
+            
+            byte[] msgBytes = mapper.writeValueAsBytes(ep);
+            
+            // Given the message bytes, simulate delivery of our WebSocket message
+            tl.onMessage(ws, new ByteString(msgBytes));
+        }
+        
+        // Consumers run in BG threads, so wait a bit for those to finish.
+        mocked.await(5L, TimeUnit.SECONDS);
+        
+        assertEquals("Mocked service expected vs. actual", mapMsgCount + extraTestMsgs.size(),
+                     mocked.getReceivedCounter());
+        List<Exchange> exchanges = mocked.getReceivedExchanges();
+        Set<String> uniqueMsgs = new HashSet<>();
+        AtomicInteger msgsReceived = new AtomicInteger();
+        exchanges.forEach(exchange -> {
+            Object exchangeBody = exchange.getIn().getBody();
+            Map<String, Object> exchangeHdrs = exchange.getIn().getHeaders();
+            assertNotNull("Null exchange body",  exchangeBody);
+            // Verify that we got JSON -- which will manifest here as a String, but we'll verify that we can convert it.
+            assertTrue("Vantiq Consumer Output wrong type: " + exchangeBody.getClass().getName(),
+                       exchangeBody instanceof BaseJsonNode);
+            msgsReceived.addAndGet(1);
+            Object msg;
+            try {
+                msg = mapper.convertValue(exchangeBody, Map.class); //readValue((String) exchangeBody, Map.class);
+            } catch (Exception e) {
+                // Maybe we asked for the wrong type
+                try {
+                    msg = mapper.convertValue(exchangeBody, String.class);
+                } catch (Exception ex) {
+                    try {
+                        msg = mapper.convertValue(exchangeBody, List.class);
+                        // readValue((String) exchangeBody,List.class);
+                    } catch (Exception exc) {
+                        throw new RuntimeException(exc);
+                    }
+                }
+            }
+            assertNotNull("Deserialized msg is null", msg);
+            if (msg instanceof Map) {
+                assert ((Map<?,?>) msg).containsKey(TEST_MSG_KEY);
+                assert ((Map<?,?>) msg).get(TEST_MSG_KEY) instanceof String;
+                assert ((String) ((Map<?,?>) msg).get(TEST_MSG_KEY)).startsWith(TEST_MSG_PREAMBLE);
+                assert exchangeHdrs.get("theMessage") instanceof Map;
+                assert (((Map<?,?>) msg).get(TEST_MSG_KEY)).equals(
+                        ((Map<?,?>) exchangeHdrs.get("theMessage")).get(TEST_MSG_KEY));
+                assert exchangeHdrs.get("counter") instanceof Number;
+                uniqueMsgs.add(((String) ((Map<?,?>) msg).get(TEST_MSG_KEY)));
+            } else if (msg instanceof String) {
+                assertEquals(msg, extraTestMsgs.get(0));
+                assert exchangeHdrs.get("theMessage") instanceof String;
+                assert msg.equals(exchangeHdrs.get("theMessage"));
+                assert exchangeHdrs.get("counter") instanceof Number;
+                assert exchangeHdrs.get("isExtra") instanceof Boolean;
+                assert (Boolean) exchangeHdrs.get("isExtra");
+                assert exchangeHdrs.get("header1") instanceof String;
+                assert exchangeHdrs.get("dupHeader1") instanceof String;
+                assert exchangeHdrs.get("header1").equals(exchangeHdrs.get("dupHeader1"));
+                assert exchangeHdrs.get("header2") instanceof String;
+                assert exchangeHdrs.get("dupHeader2") instanceof String;
+                assert exchangeHdrs.get("header2").equals(exchangeHdrs.get("dupHeader2"));
+            } else {
+                assert msg.equals(exchangeHdrs.get("theMessage"));
+                assert exchangeHdrs.get("theMessage").getClass().equals(msg.getClass());
+                assert exchangeHdrs.get("counter") instanceof Number;
+                assert exchangeHdrs.get("isExtra") instanceof Boolean;
+                assert (Boolean) exchangeHdrs.get("isExtra");
+                assert exchangeHdrs.get("header1") instanceof String;
+                assert exchangeHdrs.get("dupHeader1") instanceof String;
+                assert exchangeHdrs.get("header1").equals(exchangeHdrs.get("dupHeader1"));
+                assert exchangeHdrs.get("header2") instanceof String;
+                assert exchangeHdrs.get("dupHeader2") instanceof String;
+                assert exchangeHdrs.get("header2").equals(exchangeHdrs.get("dupHeader2"));
+    
+            }
+        });
+        assert uniqueMsgs.size() == mapMsgCount;
+        assert msgsReceived.get() == mapMsgCount + extraTestMsgs.size();
+    }
+    
     @Test
     public void testVantiqConsumerQuery() throws Exception {
         
@@ -763,7 +990,23 @@ public void testVantiqConsumerOutputJsonQuery() throws Exception {
     protected RouteBuilder createRouteBuilder() {
         return new RouteBuilder() {
             public void configure() {
-                
+                // since everything starts here, we need to instantiate this bean before startup so here as things
+                // are defined.
+                HeaderDuplicationBean hdBean = new HeaderDuplicationBean();
+                Map<String, String> heMap = Map.of("header1", "dupHeader1",
+                                                   "header2", "dupHeader2");
+                hdBean.setHeaderDuplicationMap(heMap);
+                context.getRegistry().bind(TEST_HEADER_BEAN_NAME, hdBean);
+                HeaderDuplicationBean testHdBean =
+                        context.getRegistry().lookupByNameAndType(TEST_HEADER_BEAN_NAME, HeaderDuplicationBean.class);
+                assertNotNull(testHdBean);
+                Map<String, String> testHeMap = testHdBean.getHeaderDuplicationMap();
+                assertNotNull(testHeMap);
+                assertEquals(heMap.size(), testHeMap.size());
+    
+                heMap.forEach( (k, v) -> {
+                    assertEquals(heMap.get(k), testHeMap.get(k));
+                });
                 // Override the component type to be used...
                 context.addComponent("vantiq", new FauxVantiqComponent());
                 for (String name: context.getComponentNames()) {
@@ -776,25 +1019,31 @@ public void configure() {
                 
                 from(routeStartUri)
                   .to(vantiqEndpointUri);
-                
+
                 from(routeStartStructuredUri)
                         .to(vantiqEndpointStructuredUri);
-                
+
                 from(vantiqSenderUri)
                         .to(routeEndUri);
-    
+
                 from(vantiqJsonSenderUri)
                         .to(routeEndUri);
-    
+
                 from(vantiqStructuredJsonSenderUri)
                         .to(routeEndUri);
     
+                from(vantiqStructuredJsonSenderMappedUri)
+                        .to(routeEndUri);
+                
+                from(routeStartStructuredHeaderMapUri)
+                        .to(vantiqEndpointStructuredHeaderMapUri);
+                
                 from(vantiqQuerySenderUri)
                         .setExchangePattern(ExchangePattern.InOut)
                         .to(routeEndUri)
                             .setBody(constant("{ \"Response\": \"Message\"}"))
                         .to(vantiqQuerySenderUri);
-    
+
                 from(vantiqJsonQuerierUri)
                         .setExchangePattern(ExchangePattern.InOut)
                         .to(routeEndUri)