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

Issue:
Camel Component: Allow Vantiq Camel Consumer to produce JSON in lieu of Java Map
Vastly simplifies some other ongoing work.


Discussion / hints:

Fix patch:
diff --git a/camelComponent/README.md b/camelComponent/README.md
index 0265603e..b238c9a7 100644
--- a/camelComponent/README.md
+++ b/camelComponent/README.md
@@ -81,7 +81,11 @@ your Source Implementation) as the Source Type. No configuration information is
 ## Messages from the Source
 
 Messages that are sent to the component will arrive in a Camel Exchange as a Java Map, where the keys in the map
-correspond to the property names in the object sent from Vantiq.  Messages sent to Vantiq from the component will
+correspond to the property names in the object sent from Vantiq. If desired, you change this behavior
+by adding the `consumerOutputJson` component endpoint option (see below) with a value of `true`.
+When so specified, messages sent to a Vantiq consumer will arrive in a Camel Exchange as a JSON string.
+
+Messages sent to Vantiq from the component will
 arrive as Vail objects, where the property names correspond to the Map keys.
 
 ## Exchanges to Vantiq
@@ -96,7 +100,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>]
+    vantiq://host[:port]?sourceName=<source name>&accessToken=<access token>[&sendPings=<boolean>][&failedMessageQueueSize=<int>][&consumerOutputJson=<boolean>]
 ```
 
 ### Component Endpoint Options
@@ -109,6 +113,7 @@ The endpoint options apply to producer and consumer endpoints.  The following ar
 * **accessToken** is the access token used for the connection
 * **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.
 
 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 1d7519e9..f4274740 100644
--- a/camelComponent/src/main/java/io/vantiq/extsrc/camel/VantiqConsumer.java
+++ b/camelComponent/src/main/java/io/vantiq/extsrc/camel/VantiqConsumer.java
@@ -11,6 +11,8 @@
 import static org.apache.camel.ExchangePattern.InOnly;
 import static org.apache.camel.ExchangePattern.InOut;
 
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import io.vantiq.extjsdk.ExtensionServiceMessage;
 import io.vantiq.extjsdk.ExtensionWebSocketClient;
 import io.vantiq.extjsdk.Handler;
@@ -30,6 +32,8 @@ public class VantiqConsumer extends DefaultConsumer {
 
     private ExecutorService executorService;
     
+    ObjectMapper mapper = new ObjectMapper();
+    
     public VantiqConsumer(VantiqEndpoint endpoint, Processor processor) {
         super(endpoint, processor);
         this.endpoint = endpoint;
@@ -92,6 +96,12 @@ private void processMessage(Object msg, ExchangePattern pattern) {
         } else {
             ExtensionServiceMessage message = (ExtensionServiceMessage) msg;
             Object msgBody = message.getObject();
+
+            if (endpoint.isConsumerOutputJson()) {
+                // Convert to JSON output
+               JsonNode jnode =  mapper.convertValue(msgBody, JsonNode.class);
+               msgBody = jnode.toPrettyString();
+            }
     
             // 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
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 0d0a538f..d1c7180e 100644
--- a/camelComponent/src/main/java/io/vantiq/extsrc/camel/VantiqEndpoint.java
+++ b/camelComponent/src/main/java/io/vantiq/extsrc/camel/VantiqEndpoint.java
@@ -71,6 +71,11 @@ public class VantiqEndpoint extends DefaultEndpoint {
     @Setter
     private boolean noSsl;
     
+    @UriParam(defaultValue = "false")
+    @Getter
+    @Setter
+    private boolean consumerOutputJson;
+    
     @UriParam(defaultValue = "25")
     @Setter
     private int failedMessageQueueSize;


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 60d37996..c3204f15 100644
--- a/camelComponent/src/test/java/io/vantiq/extsrc/camel/VantiqComponentTest.java
+++ b/camelComponent/src/test/java/io/vantiq/extsrc/camel/VantiqComponentTest.java
@@ -16,6 +16,15 @@
 import io.vantiq.extjsdk.FalseWebSocket;
 import io.vantiq.extjsdk.Response;
 import io.vantiq.extjsdk.TestListener;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 import okhttp3.WebSocket;
 import okio.ByteString;
 import org.apache.camel.Exchange;
@@ -27,16 +36,6 @@
 import org.apache.camel.test.junit4.CamelTestSupport;
 import org.junit.Test;
 
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.UUID;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
-
 public class VantiqComponentTest extends CamelTestSupport {
     
     private static final String testSourceName = "camelSource";
@@ -60,6 +59,16 @@ public class VantiqComponentTest extends CamelTestSupport {
             "?sourceName=" + testSourceName +
             "&accessToken=" + accessToken;
     
+    private final String vantiqJsonSenderUri = "vantiq://jsonsenderdoesntmatter/" +
+            "?sourceName=" + testSourceName +
+            "&accessToken=" + accessToken +
+            "&consumerOutputJson=true";
+    
+    private final String vantiqJsonQuerierUri = "vantiq://jsonquerierdoesntmatter/" +
+            "?sourceName=" + testSourceName +
+            "&accessToken=" + accessToken +
+            "&consumerOutputJson=true";
+    
     @Test
     public void testVantiqSetup() throws Exception {
         // First, grab our test environment.
@@ -108,6 +117,7 @@ public void testVantiqProducerJson() throws Exception {
             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");
@@ -131,6 +141,7 @@ public void testVantiqProducerJson() throws Exception {
             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");
@@ -152,6 +163,7 @@ public void testVantiqProducerJson() throws Exception {
                 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");
@@ -170,6 +182,7 @@ public void testVantiqProducerJson() throws Exception {
         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("bye");
         assert "mom".equals(msg.get("bye"));
@@ -255,6 +268,93 @@ public void testVantiqConsumer() throws Exception {
         assert uniqueMsgs.size() == expectedMsgCount;
     }
     
+    @Test
+    public void testVantiqConsumerOutputJson() 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(vantiqJsonSenderUri);
+        
+        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);
+                ep.object = msg;
+            } else {
+                ep.object = extraTestMsgs.get(i - mapMsgCount);
+            }
+            
+            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<String>();
+        AtomicInteger msgsReceived = new AtomicInteger();
+        exchanges.forEach(exchange -> {
+            Object exchangeBody = exchange.getIn().getBody();
+            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 = null;
+            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((String) msg, extraTestMsgs.get(0));
+            } else {
+                assertEquals((List) msg, extraTestMsgs.get(1));
+            }
+        });
+        assert uniqueMsgs.size() == mapMsgCount;
+        assert msgsReceived.get() == mapMsgCount + extraTestMsgs.size();
+    }
+    
     @Test
     public void testVantiqConsumerQuery() throws Exception {
         
@@ -329,6 +429,93 @@ public void testVantiqConsumerQuery() throws Exception {
         String ra = rsp.getHeader(ExtensionServiceMessage.RESPONSE_ADDRESS_HEADER);
         assert responseAddresses.contains(ra);
     }
+    
+    @Test
+    public void testVantiqConsumerOutputJsonQuery() 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(vantiqJsonQuerierUri);
+    
+        assertNotNull("Endpoint's client is null", endp.myClient.getListener());
+        assertTrue("Endpoint's client is not a TestListener: " +
+                              endp.myClient.getListener().getClass().getName(),
+                      endp.myClient.getListener() instanceof TestListener);
+        TestListener tl = (TestListener) endp.myClient.getListener();
+        
+        int expectedMsgCount = 10;
+        MockEndpoint mocked = getMockEndpoint(routeEndUri);
+        mocked.expectedMinimumMessageCount(expectedMsgCount);
+        VantiqEndpoint ve = getMandatoryEndpoint(vantiqEndpointUri, VantiqEndpoint.class);
+        assert ve != null;
+        ObjectMapper mapper = new ObjectMapper();
+        WebSocket ws = new FalseWebSocket();
+        Set<String> responseAddresses = new HashSet<>();
+        for (int i = 0; i < expectedMsgCount; i++) {
+            ExtensionServiceMessage ep = new ExtensionServiceMessage(vantiqEndpointUri);
+            ep.op = ExtensionServiceMessage.OP_QUERY;
+            ep.resourceName = "SOURCES";
+            ep.resourceId = testSourceName;
+            Map<String, Object> hdrs = new HashMap<>();
+            String respAddr = UUID.randomUUID().toString();
+            hdrs.put(ExtensionServiceMessage.ORIGIN_ADDRESS_HEADER, respAddr);
+            responseAddresses.add(respAddr);
+            ep.messageHeaders = hdrs;
+            HashMap<String, Object> msg = new HashMap<>();
+            msg.put(TEST_MSG_KEY, TEST_MSG_PREAMBLE + i);
+            ep.object = msg;
+            
+            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", expectedMsgCount, mocked.getReceivedCounter());
+        List<Exchange> exchanges = mocked.getReceivedExchanges();
+        Set<String> uniqueMsgs = new HashSet<String>();
+        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;
+            try {
+                msg = mapper.readValue((String) exchBody, Map.class);
+            } catch (JsonProcessingException e) {
+                throw new RuntimeException(e);
+            }
+            assertNotNull("Deserialized JSON message is null", msg);
+            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);
+            uniqueMsgs.add(((String) msg.get(TEST_MSG_KEY)));
+            
+            Map<String, Object> props = exchange.getProperties();
+            assert props.containsKey(ExtensionServiceMessage.RESPONSE_ADDRESS_HEADER);
+        });
+        assert uniqueMsgs.size() == expectedMsgCount;
+        
+        // Note that we need to fetch the endpoints by URI since there are more than one of them.
+        FauxVantiqEndpoint ep = (FauxVantiqEndpoint) context.getEndpoint(vantiqJsonQuerierUri);
+        FalseClient fc = ep.myClient;
+        Response rsp = fc.getLastMessageAsResponse();
+        assert rsp != null;
+        assert rsp.getBody() != null;
+        assert rsp.getStatus() == 200;
+        Object o = rsp.getBody();
+        assert o instanceof Map;
+        Map<String, Object> response = (Map) o;
+        log.debug("Found final response message: {}", response);
+        assert response.containsKey("Response");
+        String ra = rsp.getHeader(ExtensionServiceMessage.RESPONSE_ADDRESS_HEADER);
+        assert responseAddresses.contains(ra);
+    }
 
     @Override
     protected RouteBuilder createRouteBuilder() throws Exception {
@@ -351,11 +538,20 @@ public void configure() {
                 from(vantiqSenderUri)
                         .to(routeEndUri);
     
+                from(vantiqJsonSenderUri)
+                        .to(routeEndUri);
+                
                 from(vantiqQuerySenderUri)
                         .setExchangePattern(ExchangePattern.InOut)
                         .to(routeEndUri)
                             .setBody(constant("{ \"Response\": \"Message\"}"))
                         .to(vantiqQuerySenderUri);
+    
+                from(vantiqJsonQuerierUri)
+                        .setExchangePattern(ExchangePattern.InOut)
+                        .to(routeEndUri)
+                        .setBody(constant("{ \"Response\": \"Message\"}"))
+                        .to(vantiqJsonQuerierUri);
             }
         };
     }