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

Issue:
camelComponent: add structured message option
There are times when camel components (not Vantiq) use the messages headers to obtain or provide information.  

Add a URL option to instruct the Vantiq camel component to provide the message (to or from Vantiq) in format matching Camel's structure.  Something like

```js
{
    header: { ... },
    message: { ... }
}
```

where header & message will be another Json object.  We will define a schema type for this structure.

This doesn't give low-level details (for some components, the `message` part may be well-structured, but it gives us some control and allows us to provide the headers as desired.  Since the Vantiq camel component did not do things this way from the start, this is an option -- use of the option allows for backward compatibility.


Discussion / hints:

Fix patch:
diff --git a/camelComponent/README.md b/camelComponent/README.md
index b238c9a7..ccf70e67 100644
--- a/camelComponent/README.md
+++ b/camelComponent/README.md
@@ -88,6 +88,12 @@ When so specified, messages sent to a Vantiq consumer will arrive in a Camel Exc
 Messages sent to Vantiq from the component will
 arrive as Vail objects, where the property names correspond to the Map keys.
 
+### Structured Headers and Messages
+
+By adding the `structuredMessageHeader` component endpoint option with a value of `true` (see below), you can set or access the Camel message headers.  When this option is set to `true`, messages sent to or received from the Camel Component will have two properties: `headers` and `message`, both containing Vail objects. The `message` property will contain a Vail object where each property corresponds to same-named property in the underlying message.  The `headers` property will contain a Vail object where each property contains the value of the named message header.
+
+A schema type definition for this message structure is available in `src/main/resources/types/io/vantiq/extsrc/camelcomp/message.json`.
+
 ## Exchanges to Vantiq
 
 The Vantiq Component expects messages in an exchange to arrive in the form of a Java Map.  However, messages arriving
@@ -100,7 +106,7 @@ in Json format will be accepted as well.
 The URI for the Vantiq connection takes the following form:
 
 ```
-    vantiq://host[:port]?sourceName=<source name>&accessToken=<access token>[&sendPings=<boolean>][&failedMessageQueueSize=<int>][&consumerOutputJson=<boolean>]
+    vantiq://host[:port]?sourceName=<source name>&accessToken=<access token>[&sendPings=<boolean>][&failedMessageQueueSize=<int>][&consumerOutputJson=<boolean>][&structuredMessageHeader=<boolean>]
 ```
 
 ### Component Endpoint Options
@@ -114,6 +120,15 @@ The endpoint options apply to producer and consumer endpoints.  The following ar
 * **sendPings** [optional] a boolean value indicating whether to periodically ping the Vantiq server (default is false)
 * **failedMessageQueueSize** [optional] an integer value indicating how many messages to hold for sending when the connection to the Vantiq server fails.
 * **consumerOutputJson** [optional] a boolean value indicating whether messages sent from Vantiq to the Vantiq component (_i.e._, a Vantiq Consumer) should be put into an Apache Camel Exchange as a JSON message.  If false, messages are put into an exchange as a Java Map.
+* **structuredMessageHeader** [optional] a boolean value indicating whether messages sent to and from the Vantiq component (Vantiq Consumers and Producers) should be structured as
+
+    ```json
+    {
+        headers: { ... },
+        message: { ... }
+    }
+    ```  
+
 
 For example, to create an Apache Camel application that receives messages from a Vantiq system at `vantiq.example.com`
 from source `exampleSource` and access token `FD10s0x...`, we would use something like the following:
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 f4274740..afeddf8a 100644
--- a/camelComponent/src/main/java/io/vantiq/extsrc/camel/VantiqConsumer.java
+++ b/camelComponent/src/main/java/io/vantiq/extsrc/camel/VantiqConsumer.java
@@ -8,6 +8,8 @@
 
 package io.vantiq.extsrc.camel;
 
+import static io.vantiq.extsrc.camel.VantiqEndpoint.STRUCTURED_MESSAGE_HEADERS_PROPERTY;
+import static io.vantiq.extsrc.camel.VantiqEndpoint.STRUCTURED_MESSAGE_MESSAGE_PROPERTY;
 import static org.apache.camel.ExchangePattern.InOnly;
 import static org.apache.camel.ExchangePattern.InOut;
 
@@ -17,8 +19,10 @@
 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;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.camel.Exchange;
 import org.apache.camel.ExchangePattern;
@@ -60,31 +64,30 @@ protected void doStop() throws Exception {
 
         // shutdown the thread pool gracefully
         getEndpoint().getCamelContext().getExecutorServiceManager().shutdownGraceful(executorService);
+        if (vantiqClient != null) {
+            vantiqClient.close();
+        }
     }
     
     /**
      * A handler for dealing with publishes to Camel.
      */
-    private final Handler<ExtensionServiceMessage> publishHandler = new Handler<ExtensionServiceMessage>() {
+    private final Handler<ExtensionServiceMessage> publishHandler = new Handler<>() {
         @Override
         public void handleMessage(ExtensionServiceMessage message) {
             // When we get a message, process it in the background...
-            executorService.submit( () -> {
-                processMessage(message, InOnly);
-            });
+            executorService.submit( () -> processMessage(message, InOnly));
         }
     };
     
     /**
      * A handler for dealing with queries to Camel.
      */
-    private final Handler<ExtensionServiceMessage> queryHandler = new Handler<ExtensionServiceMessage>() {
+    private final Handler<ExtensionServiceMessage> queryHandler = new Handler<>() {
         @Override
         public void handleMessage(ExtensionServiceMessage message) {
             // When we get a message, process it in the background...
-            executorService.submit( () -> {
-                processMessage(message, InOut);
-            });
+            executorService.submit( () -> processMessage(message, InOut));
         }
     };
     
@@ -96,7 +99,25 @@ private void processMessage(Object msg, ExchangePattern pattern) {
         } else {
             ExtensionServiceMessage message = (ExtensionServiceMessage) msg;
             Object msgBody = message.getObject();
+            Map<String, Object> camelHdrs = null;
+            Object camelBody = null;
+            if (endpoint.isStructuredMessageHeader() && msgBody instanceof Map) {
+                camelHdrs = new HashMap<>();
+                Map<?,?> msgAsMap = (Map<?,?>) msgBody;
+                if (msgAsMap.get(STRUCTURED_MESSAGE_HEADERS_PROPERTY) instanceof Map) {
+                    Map<?,?> hdrMap = (Map<?,?>) msgAsMap.get(VantiqEndpoint.STRUCTURED_MESSAGE_HEADERS_PROPERTY);
+                    camelHdrs = hdrMap.entrySet()
+                                .stream()
+                                .filter(e -> e.getKey() instanceof String)
+                                .collect(Collectors.toMap(e -> (String) e.getKey(), Map.Entry::getValue));
 
+                }
+                if (msgAsMap.get(STRUCTURED_MESSAGE_MESSAGE_PROPERTY) != null) {
+                    camelBody = msgAsMap.get(STRUCTURED_MESSAGE_MESSAGE_PROPERTY);
+                }
+                msgBody = camelBody;
+                log.debug("Structured message -- hdrs: {}, message: {}", camelHdrs, camelBody);
+            }
             if (endpoint.isConsumerOutputJson()) {
                 // Convert to JSON output
                JsonNode jnode =  mapper.convertValue(msgBody, JsonNode.class);
@@ -106,10 +127,13 @@ private void processMessage(Object msg, ExchangePattern pattern) {
             // Create an exchange to move our message along.  In the publish case,
             // we have no interest in the result, so we'll allow it to be released when
             // camel operations are complete.  If/when we support queries, we will care about the
-            // result so we'll deal with those separately.
+            // result, so we'll deal with those separately.
             final Exchange exchange = createExchange(false);
             exchange.setPattern(pattern);
             exchange.getIn().setBody(msgBody);
+            if (camelHdrs != null) {
+                exchange.getIn().setHeaders(camelHdrs);
+            }
             if (pattern == InOut) {
                 // In this case, we need to save the reply address in the exchange so that we can reply appropriately
                 // This is the case when we are processing a Vantiq query.
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 d1c7180e..33ee1763 100644
--- a/camelComponent/src/main/java/io/vantiq/extsrc/camel/VantiqEndpoint.java
+++ b/camelComponent/src/main/java/io/vantiq/extsrc/camel/VantiqEndpoint.java
@@ -53,33 +53,49 @@
 @UriParams
 @Slf4j
 public class VantiqEndpoint extends DefaultEndpoint {
+    
+    public static final String STRUCTURED_MESSAGE_MESSAGE_PROPERTY = "message";
+    public static final String STRUCTURED_MESSAGE_HEADERS_PROPERTY = "headers";
+    
+    public static final String SOURCE_NAME_PARAM = "sourceName";
     @UriParam @Metadata(required = true)
     @Setter
     private String sourceName;
     
+    public static final String ACCESS_TOKEN_PARAM = "accessToken";
     @UriParam(label="security", secret = true) @Metadata(required = true)
     @Setter
     private String accessToken;
     
+    public static final String SEND_PINGS_PARAM = "sendPings";
     @UriParam(defaultValue = "false")
     @Getter
     @Setter
     private boolean sendPings;
     
+    public static final String NO_SSL_PARAM = "noSsl";
     @UriParam(defaultValue = "false")
     @Getter
     @Setter
     private boolean noSsl;
     
+    public static final String CONSUMER_OUTPUT_JSON_PARAM = "consumerOutputJson";
     @UriParam(defaultValue = "false")
     @Getter
     @Setter
     private boolean consumerOutputJson;
     
+    public static final String FAILED_MESSAGE_QUEUE_SIZE_PARAM = "failedMessageQueueSize";
     @UriParam(defaultValue = "25")
     @Setter
     private int failedMessageQueueSize;
     
+    public static final String STRUCTURED_MESSAGE_HEADER_PARAM = "structuredMessageHeader";
+    @UriParam(defaultValue = "false")
+    @Getter
+    @Setter
+    private boolean structuredMessageHeader;
+    
     @Setter
     @Getter
     private ExtensionWebSocketClient vantiqClient;
@@ -94,7 +110,7 @@ public class VantiqEndpoint extends DefaultEndpoint {
     @Getter
     private String endpointName;
     
-    InstanceConfigUtils utils = null;
+    InstanceConfigUtils utils;
 
     public VantiqEndpoint() {
         utils = new InstanceConfigUtils();
@@ -294,7 +310,7 @@ public String createEndpointUri() {
             this.setEndpointUri(baseUri);
             String origScheme = vantiqURI.getScheme();
             
-            StringBuffer epString = new StringBuffer(vantiqURI.toString());
+            StringBuilder epString = new StringBuilder(vantiqURI.toString());
             epString.append("?sourceName=").append(sourceName);
             this.sourceName = sourceName;
             epString.append("&accessToken=").append(accessToken);
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 e21a285c..17c0c2db 100644
--- a/camelComponent/src/main/java/io/vantiq/extsrc/camel/VantiqProducer.java
+++ b/camelComponent/src/main/java/io/vantiq/extsrc/camel/VantiqProducer.java
@@ -8,6 +8,9 @@
 
 package io.vantiq.extsrc.camel;
 
+import static io.vantiq.extsrc.camel.VantiqEndpoint.STRUCTURED_MESSAGE_HEADERS_PROPERTY;
+import static io.vantiq.extsrc.camel.VantiqEndpoint.STRUCTURED_MESSAGE_MESSAGE_PROPERTY;
+
 import com.fasterxml.jackson.core.type.TypeReference;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -19,8 +22,6 @@
 import org.apache.camel.converter.stream.InputStreamCache;
 import org.apache.camel.InvalidPayloadException;
 import org.apache.camel.support.DefaultProducer;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 import java.net.HttpURLConnection;
 import java.util.HashMap;
@@ -28,7 +29,7 @@
 
 @Slf4j
 public class VantiqProducer extends DefaultProducer {
-    private VantiqEndpoint endpoint;
+    private final VantiqEndpoint endpoint;
     ObjectMapper mapper = new ObjectMapper();
     
     public VantiqProducer(VantiqEndpoint endpoint) {
@@ -51,7 +52,7 @@ public void process(Exchange exchange) throws Exception {
                 log.debug("Camel converted body is {}, msg type: {}, msg: {}", vMsg, msg.getClass().getName(), msg);
             }
             if (vMsg == null) {
-                // Then Camel couldn't convert on it's own.  Let's see if we can help things along...
+                // Then Camel couldn't convert on its own.  Let's see if we can help things along...
                 if (msg instanceof String) {
                     // Then we must be fetching a JSON string.
                     String strMsg = (String) msg;
@@ -98,6 +99,17 @@ public void process(Exchange exchange) throws Exception {
         if (ra instanceof String) {
             responseAddr = (String) ra;
         }
+        
+        if (endpoint.isStructuredMessageHeader()) {
+            // Then, our caller is asking for message data formatted as header & message
+            Map<String, Object> fmtMsg = new HashMap<>();
+            // the headers are usually implemented as a CaseInsensitiveMap, so we'll have Java walk the list.
+            // We walk the list so that we get the original case to send on...
+            Map<String, Object> hdrs = new HashMap<>(exchange.getMessage().getHeaders());
+            fmtMsg.put(STRUCTURED_MESSAGE_HEADERS_PROPERTY, hdrs);
+            fmtMsg.put(STRUCTURED_MESSAGE_MESSAGE_PROPERTY, vMsg);
+            vMsg = fmtMsg;
+        }
         if (exchange.getPattern() == ExchangePattern.InOut) {
             if (exchange.getException() != null) {
                 Exception e = exchange.getException();
diff --git a/camelComponent/src/main/resources/types/io/vantiq/extsrc/camelcomp/message.json b/camelComponent/src/main/resources/types/io/vantiq/extsrc/camelcomp/message.json
new file mode 100644
index 00000000..1406b668
--- /dev/null
+++ b/camelComponent/src/main/resources/types/io/vantiq/extsrc/camelcomp/message.json
@@ -0,0 +1,21 @@
+{
+  "ars_relationships" : [ ],
+  "name" : "io.vantiq.extsrc.camelcomp.message",
+  "properties" : {
+    "headers" : {
+      "encrypted" : false,
+      "multi" : false,
+      "required" : false,
+      "type" : "Object",
+      "unique" : false
+    },
+    "message" : {
+      "encrypted" : false,
+      "multi" : false,
+      "required" : false,
+      "type" : "Object",
+      "unique" : false
+    }
+  },
+  "role" : "schema"
+}
\ No newline at end of file


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 c3204f15..95d4e9e0 100644
--- a/camelComponent/src/test/java/io/vantiq/extsrc/camel/VantiqComponentTest.java
+++ b/camelComponent/src/test/java/io/vantiq/extsrc/camel/VantiqComponentTest.java
@@ -8,6 +8,13 @@
 
 package io.vantiq.extsrc.camel;
 
+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.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;
+import static io.vantiq.extsrc.camel.VantiqEndpoint.STRUCTURED_MESSAGE_MESSAGE_PROPERTY;
+
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.node.ObjectNode;
@@ -44,33 +51,46 @@ public class VantiqComponentTest extends CamelTestSupport {
     private static final String TEST_MSG_KEY = "someRandomKey";
     
     private final String routeStartUri = "direct:start";
+    
+    private final String routeStartStructuredUri = "direct:structuredstart";
     private final String routeEndUri = "mock:direct:result";
     
     private final String exceptionEndpoint = "mock:direct:error";
     private final String vantiqEndpointUri = "vantiq://doesntmatter/" +
-            "?sourceName=" + testSourceName +
-            "&accessToken=" + accessToken;
+            "?" + SOURCE_NAME_PARAM + "=" + testSourceName +
+            "&" + ACCESS_TOKEN_PARAM + "=" + accessToken;
+    
+    private final String vantiqEndpointStructuredUri = "vantiq://doesntmatter/" +
+            "?" + SOURCE_NAME_PARAM + "=" + testSourceName +
+            "&" + ACCESS_TOKEN_PARAM + "=" + accessToken +
+            "&" + STRUCTURED_MESSAGE_HEADER_PARAM + "=true";
     
     private final String vantiqSenderUri = "vantiq://senderdoesntmatter/" +
-            "?sourceName=" + testSourceName +
-            "&accessToken=" + accessToken;
+            "?" + SOURCE_NAME_PARAM + "=" + testSourceName +
+            "&" + ACCESS_TOKEN_PARAM + "=" + accessToken;
     
     private final String vantiqQuerySenderUri = "vantiq://querierdoesntmatter/" +
             "?sourceName=" + testSourceName +
             "&accessToken=" + accessToken;
     
     private final String vantiqJsonSenderUri = "vantiq://jsonsenderdoesntmatter/" +
-            "?sourceName=" + testSourceName +
-            "&accessToken=" + accessToken +
-            "&consumerOutputJson=true";
+            "?" + SOURCE_NAME_PARAM + "=" + testSourceName +
+            "&" + ACCESS_TOKEN_PARAM + "=" + accessToken +
+            "&" + CONSUMER_OUTPUT_JSON_PARAM + "=true";
+    
+    public final String vantiqStructuredJsonSenderUri = "vantiq://structuredsenderdoesntmatter/" +
+            "?" + SOURCE_NAME_PARAM + "=" + testSourceName +
+            "&" + ACCESS_TOKEN_PARAM + "=" + accessToken +
+            "&" + CONSUMER_OUTPUT_JSON_PARAM + "=true" +
+            "&" + STRUCTURED_MESSAGE_HEADER_PARAM + "=true";
     
     private final String vantiqJsonQuerierUri = "vantiq://jsonquerierdoesntmatter/" +
-            "?sourceName=" + testSourceName +
-            "&accessToken=" + accessToken +
-            "&consumerOutputJson=true";
+            "?" + SOURCE_NAME_PARAM + "=" + testSourceName +
+            "&" + ACCESS_TOKEN_PARAM + "=" + accessToken +
+            "&" + CONSUMER_OUTPUT_JSON_PARAM + "=true";
     
     @Test
-    public void testVantiqSetup() throws Exception {
+    public void testVantiqSetup() {
         // First, grab our test environment.
         FauxVantiqComponent vc = (FauxVantiqComponent) context.getComponent("vantiq");
         assert vc != null;
@@ -89,7 +109,7 @@ public void testVantiqSetup() throws Exception {
     }
     
     @Test
-    public void testVantiqProducerJson() throws Exception {
+    public void testVantiqProducerJson() {
     
         // First, grab our test environment.
         FauxVantiqComponent vc = (FauxVantiqComponent) context.getComponent("vantiq");
@@ -98,7 +118,7 @@ public void testVantiqProducerJson() throws Exception {
         FauxVantiqEndpoint ep = (FauxVantiqEndpoint) context.getEndpoint(vantiqEndpointUri);
         FalseClient fc = ep.myClient;
         
-        ArrayList<String> msgs =new ArrayList<String>();
+        ArrayList<String> msgs =new ArrayList<>();
         int itemCount = 10;
         for (int i = 0; i < itemCount; i++) {
             msgs.add("{\"hi\": \"mom " + i + "\"}");
@@ -109,18 +129,9 @@ public void testVantiqProducerJson() throws Exception {
             
             assert item != null;
             sendBody(routeStartUri, item);
+            //noinspection rawtypes
             Map lastMsg = fc.getLastMessageAsMap();
-            assert lastMsg != null;
-            assert lastMsg.containsKey("op");
-            assert "notification".equals(lastMsg.get("op"));
-            assert lastMsg.containsKey("sourceName");
-            assert testSourceName.equals(lastMsg.get("sourceName"));
-            assert lastMsg.containsKey("object");
-            assert lastMsg.get("object") instanceof Map;
-            //noinspection unchecked
-            Map<String, Object> msg = (Map<String, Object>) lastMsg.get("object");
-            assert msg.containsKey("hi");
-            assert ((String) msg.get("hi")).contains("mom");
+            validateExtensionMsg(lastMsg, false, null, new String[] {"hi"}, "mom");
         }
     
         // Now, try a similar test sending maps
@@ -130,21 +141,13 @@ public void testVantiqProducerJson() throws Exception {
             mapList.add(mp);
         }
 
+        //noinspection rawtypes
         for (Map item: mapList) {
             sendBody(routeStartUri, item);
     
+            //noinspection rawtypes
             Map lastMsg = fc.getLastMessageAsMap();
-            assert lastMsg != null;
-            assert lastMsg.containsKey("op");
-            assert "notification".equals(lastMsg.get("op"));
-            assert lastMsg.containsKey("sourceName");
-            assert testSourceName.equals(lastMsg.get("sourceName"));
-            assert lastMsg.containsKey("object");
-            assert lastMsg.get("object") instanceof Map;
-            //noinspection unchecked
-            Map<String, Object> msg = (Map<String, Object>) lastMsg.get("object");
-            assert msg.containsKey("hi");
-            assert ((String) msg.get("hi")).contains("dad");
+            validateExtensionMsg(lastMsg, false, null, new String[] {"hi"}, "dad");
         }
     
         ObjectMapper mapper = new ObjectMapper();
@@ -155,27 +158,96 @@ public void testVantiqProducerJson() throws Exception {
                 node.put("hi", "aki " + counter.getAndIncrement());
                 sendBody(routeStartUri, node);
     
+                //noinspection rawtypes
                 Map lastMsg = fc.getLastMessageAsMap();
-                assert lastMsg != null;
-                assert lastMsg.containsKey("op");;
-                assert "notification".equals(lastMsg.get("op"));
-                assert lastMsg.containsKey("sourceName");
-                assert testSourceName.equals(lastMsg.get("sourceName"));
-                assert lastMsg.containsKey("object");
-                assert lastMsg.get("object") instanceof Map;
-                //noinspection unchecked
-                Map<String, Object> testMsg = (Map<String, Object>) lastMsg.get("object");
-                assert testMsg.containsKey("hi");
-                assert ((String) testMsg.get("hi")).contains("aki");
+                validateExtensionMsg(lastMsg, false, null, new String[] {"hi"}, "aki");
             } catch (JsonProcessingException e) {
                 throw new RuntimeException(e);
             }
         });
         
         sendBody(routeStartUri, "{\"bye\": \"mom\"}");
+        //noinspection rawtypes
+        Map lastMsg = fc.getLastMessageAsMap();
+        validateExtensionMsg(lastMsg, false, null, new String[] {"bye"}, "mom");
+    }
+    
+    @Test
+    public void testVantiqProducerStructuredJson() {
+        
+        // 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(vantiqEndpointStructuredUri);
+        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(routeStartStructuredUri, 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());
+            sendBody(routeStartStructuredUri, msg, hdrs);
+    
+            //noinspection rawtypes
+            Map lastMsg = fc.getLastMessageAsMap();
+            validateExtensionMsg(lastMsg, true, hdrs, 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(routeStartStructuredUri, node, hdrs);
     
+                //noinspection rawtypes
+                Map lastMsg = fc.getLastMessageAsMap();
+                validateExtensionMsg(lastMsg, true, hdrs, new String[] {"hi"}, "aki");
+            } catch (JsonProcessingException e) {
+                throw new RuntimeException(e);
+            }
+        });
+        
+        sendBody(routeStartStructuredUri, "{\"bye\": \"mom\"}");
+        
+        //noinspection rawtypes
         Map lastMsg = fc.getLastMessageAsMap();
-        assert lastMsg != null;
+        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");
         assert "notification".equals(lastMsg.get("op"));
         assert lastMsg.containsKey("sourceName");
@@ -184,8 +256,27 @@ public void testVantiqProducerJson() throws Exception {
         assert lastMsg.get("object") instanceof Map;
         //noinspection unchecked
         Map<String, Object> msg = (Map<String, Object>) lastMsg.get("object");
-        assert msg.containsKey("bye");
-        assert "mom".equals(msg.get("bye"));
+        if (isStructured) {
+            if (expHdrs != null) {
+                //noinspection unchecked
+                Map<String, Object> hdrs =
+                        (Map<String, Object>) msg.get(STRUCTURED_MESSAGE_HEADERS_PROPERTY);
+                assert hdrs != null;
+                assert expHdrs.size() == hdrs.size();
+                expHdrs.forEach( (key, value) -> {
+                    assert hdrs.containsKey(key);
+                    assert hdrs.get(key).equals(expHdrs.get(key));
+                });
+            }
+            assert msg.containsKey(STRUCTURED_MESSAGE_MESSAGE_PROPERTY);
+            assert msg.containsKey(STRUCTURED_MESSAGE_HEADERS_PROPERTY);
+            //noinspection unchecked
+            msg = (Map<String, Object>) msg.get(STRUCTURED_MESSAGE_MESSAGE_PROPERTY);
+        }
+        for (String key: msgKeys) {
+            assert msg.containsKey(key);
+            assert ((String) msg.get(key)).contains(msgPreamble);
+        }
     }
     
     @Test
@@ -256,9 +347,10 @@ public void testVantiqConsumer() throws Exception {
         
         assertEquals("Mocked service expected vs. actual", expectedMsgCount, mocked.getReceivedCounter());
         List<Exchange> exchanges = mocked.getReceivedExchanges();
-        Set<String> uniqueMsgs = new HashSet<String>();
+        Set<String> uniqueMsgs = new HashSet<>();
         exchanges.forEach(exchange -> {
                 assert exchange.getIn().getBody() instanceof Map;
+                //noinspection rawtypes
                 Map msg = exchange.getIn().getBody(Map.class);
                 assert msg.containsKey(TEST_MSG_KEY);
                 assert msg.get(TEST_MSG_KEY) instanceof String;
@@ -315,7 +407,7 @@ public void testVantiqConsumerOutputJson() throws Exception {
         assertEquals("Mocked service expected vs. actual", mapMsgCount + extraTestMsgs.size(),
                      mocked.getReceivedCounter());
         List<Exchange> exchanges = mocked.getReceivedExchanges();
-        Set<String> uniqueMsgs = new HashSet<String>();
+        Set<String> uniqueMsgs = new HashSet<>();
         AtomicInteger msgsReceived = new AtomicInteger();
         exchanges.forEach(exchange -> {
             Object exchangeBody = exchange.getIn().getBody();
@@ -324,7 +416,108 @@ public void testVantiqConsumerOutputJson() throws Exception {
             assertTrue("Vantiq Consumer Output wrong type: " + exchangeBody.getClass().getName(),
                 exchangeBody instanceof String);
             msgsReceived.addAndGet(1);
-            Object msg = null;
+            Object msg;
+            try {
+                msg = mapper.readValue((String) exchangeBody, Map.class);
+            } catch (JsonProcessingException e) {
+                // Maybe we asked for the wrong type
+                try {
+                    msg = mapper.readValue((String) exchangeBody, String.class);
+                } catch (JsonProcessingException ex) {
+                    try {
+                        msg = mapper.readValue((String) exchangeBody, List.class);
+                    } catch (JsonProcessingException 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);
+                uniqueMsgs.add(((String) ((Map<?,?>) msg).get(TEST_MSG_KEY)));
+            } else if (msg instanceof String) {
+                assertEquals(msg, extraTestMsgs.get(0));
+            } else {
+                assertEquals(msg, extraTestMsgs.get(1));
+            }
+        });
+        assert uniqueMsgs.size() == mapMsgCount;
+        assert msgsReceived.get() == mapMsgCount + extraTestMsgs.size();
+    }
+    
+    @Test
+    public void testVantiqConsumerOutputJsonStructured() 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(vantiqStructuredJsonSenderUri);
+        
+        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);
+                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 String);
+            msgsReceived.addAndGet(1);
+            Object msg;
             try {
                 msg = mapper.readValue((String) exchangeBody, Map.class);
             } catch (JsonProcessingException e) {
@@ -341,14 +534,27 @@ public void testVantiqConsumerOutputJson() throws Exception {
             }
             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);
-                uniqueMsgs.add(((String) ((Map) msg).get(TEST_MSG_KEY)));
+                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((String) msg, extraTestMsgs.get(0));
+                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");
             } else {
-                assertEquals((List) msg, extraTestMsgs.get(1));
+                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 uniqueMsgs.size() == mapMsgCount;
@@ -400,10 +606,10 @@ public void testVantiqConsumerQuery() throws Exception {
         
         assertEquals("Mocked service expected vs. actual", expectedMsgCount, mocked.getReceivedCounter());
         List<Exchange> exchanges = mocked.getReceivedExchanges();
-        Set<String> uniqueMsgs = new HashSet<String>();
+        Set<String> uniqueMsgs = new HashSet<>();
         exchanges.forEach(exchange -> {
             assert exchange.getIn().getBody() instanceof Map;
-            Map msg = exchange.getIn().getBody(Map.class);
+            Map<?,?> msg = exchange.getIn().getBody(Map.class);
             assert msg.containsKey(TEST_MSG_KEY);
             assert msg.get(TEST_MSG_KEY) instanceof String;
             assert ((String) msg.get(TEST_MSG_KEY)).startsWith(TEST_MSG_PREAMBLE);
@@ -423,7 +629,8 @@ public void testVantiqConsumerQuery() throws Exception {
         assert rsp.getStatus() == 200;
         Object o = rsp.getBody();
         assert o instanceof Map;
-        Map<String, Object> response = (Map) o;
+        //noinspection unchecked
+        Map<String, Object> response = (Map<String, Object>) o;
         log.debug("Found final response message: {}", response);
         assert response.containsKey("Response");
         String ra = rsp.getHeader(ExtensionServiceMessage.RESPONSE_ADDRESS_HEADER);
@@ -478,13 +685,13 @@ public void testVantiqConsumerOutputJsonQuery() throws Exception {
         
         assertEquals("Mocked service expected vs. actual", expectedMsgCount, mocked.getReceivedCounter());
         List<Exchange> exchanges = mocked.getReceivedExchanges();
-        Set<String> uniqueMsgs = new HashSet<String>();
+        Set<String> uniqueMsgs = new HashSet<>();
         exchanges.forEach(exchange -> {
             Object exchBody = exchange.getIn().getBody();
             assertNotNull("Exchange boddy is null", exchBody);
             assertTrue("Exchange body wrong type: " + exchBody.getClass().getName(),
                        exchange.getIn().getBody() instanceof String);
-            Map msg = null;
+            Map<?,?> msg;
             try {
                 msg = mapper.readValue((String) exchBody, Map.class);
             } catch (JsonProcessingException e) {
@@ -510,7 +717,8 @@ public void testVantiqConsumerOutputJsonQuery() throws Exception {
         assert rsp.getStatus() == 200;
         Object o = rsp.getBody();
         assert o instanceof Map;
-        Map<String, Object> response = (Map) o;
+        //noinspection unchecked
+        Map<String, Object> response = (Map<String, Object>) o;
         log.debug("Found final response message: {}", response);
         assert response.containsKey("Response");
         String ra = rsp.getHeader(ExtensionServiceMessage.RESPONSE_ADDRESS_HEADER);
@@ -518,7 +726,7 @@ public void testVantiqConsumerOutputJsonQuery() throws Exception {
     }
 
     @Override
-    protected RouteBuilder createRouteBuilder() throws Exception {
+    protected RouteBuilder createRouteBuilder() {
         return new RouteBuilder() {
             public void configure() {
                 
@@ -535,12 +743,18 @@ public void configure() {
                 from(routeStartUri)
                   .to(vantiqEndpointUri);
                 
+                from(routeStartStructuredUri)
+                        .to(vantiqEndpointStructuredUri);
+                
                 from(vantiqSenderUri)
                         .to(routeEndUri);
     
                 from(vantiqJsonSenderUri)
                         .to(routeEndUri);
-                
+    
+                from(vantiqStructuredJsonSenderUri)
+                        .to(routeEndUri);
+    
                 from(vantiqQuerySenderUri)
                         .setExchangePattern(ExchangePattern.InOut)
                         .to(routeEndUri)