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

Issue:
CamelComponent: structured messages from (at least) google pubsub fail
Reason is that we're using the object mapper to copy the map and it doesn't understand the types used in some of the google-specific headers.  When using structured message, we try & send all these values to Vantiq, and things go awry from there.


Discussion / hints:

Fix patch:
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 22300d6a..94aceef7 100644
--- a/camelComponent/src/main/java/io/vantiq/extsrc/camel/VantiqProducer.java
+++ b/camelComponent/src/main/java/io/vantiq/extsrc/camel/VantiqProducer.java
@@ -47,7 +47,7 @@ public VantiqProducer(VantiqEndpoint endpoint) {
         JavaTimeModule mod = new JavaTimeModule();
         mapper =
                 new ObjectMapper().registerModule(mod)
-                                  .configure(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS, false);;
+                                  .configure(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS, false);
     }
     
     @SuppressWarnings("unchecked")
@@ -74,7 +74,7 @@ public void process(Exchange exchange) throws Exception {
                     byte[] ba = (byte[]) msg;
                     // Then we must be fetching a JSON string.
                     if (checkUTF8(ba)) {
-                        String strMsg = (String) new String((byte[]) msg, StandardCharsets.UTF_8);
+                        String strMsg = new String((byte[]) msg, StandardCharsets.UTF_8);
                         try {
                             vMsg = mapper.readValue(strMsg, new TypeReference<>() {});
                         } catch (Exception e) {
@@ -133,16 +133,31 @@ public void process(Exchange exchange) throws Exception {
             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 =
-                    mapper.convertValue(exchange.getMessage().getHeaders(), new TypeReference<>() {});
+            // Note that we can't always use the Jackson objectMapper here because headers sometimes use Java types it
+            // doesn't understand (e.g., things from google come with Protobuf timestamps or soemthing like that.
+            // Simple .toString() handles it, so we'll copy the map ourselves.
+            Map<String, Object> hdrs;
+            try {
+                hdrs = mapper.convertValue(exchange.getMessage().getHeaders(),
+                                           new TypeReference<>() {});
+            } catch (Exception e) {
+                Map<String, Object> h = new HashMap<>();
+                exchange.getMessage().getHeaders().forEach((k, v) -> {
+                    if (v instanceof Integer || v instanceof Map) {
+                        h.put(k, v);
+                    } else {
+                        h.put(k, v.toString());
+                    }
+                });
+                hdrs = h;
+            }
             fmtMsg.put(STRUCTURED_MESSAGE_HEADERS_PROPERTY, hdrs);
             Map<String, Object> m = mapper.convertValue(vMsg, new TypeReference<>() {});
             fmtMsg.put(STRUCTURED_MESSAGE_MESSAGE_PROPERTY, m);
             vMsg = fmtMsg;
         } else {
             // run the map thru the converted to serialize any embedded dates.
-            Map<String, Object> m = mapper.convertValue(vMsg, new TypeReference<>() {});
-            vMsg = m;
+            vMsg = mapper.convertValue(vMsg, new TypeReference<>() {});
         }
         if (exchange.getPattern() == ExchangePattern.InOut) {
             if (exchange.getException() != null) {


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 b942ded7..90ac57e1 100644
--- a/camelComponent/src/test/java/io/vantiq/extsrc/camel/VantiqComponentTest.java
+++ b/camelComponent/src/test/java/io/vantiq/extsrc/camel/VantiqComponentTest.java
@@ -285,6 +285,11 @@ void validateExtensionMsg(Map<?,?> lastMsg, Boolean isStructured, Map<String, Ob
                 assert expHdrs.size() == hdrs.size();
                 expHdrs.forEach( (key, value) -> {
                     assert hdrs.containsKey(key);
+                    assert hdrs.get(key) != null;
+                    assert expHdrs.get(key) != null;
+                    log.debug("For key {}, comparing value {} ){} with expected {} ({}).",
+                              key, hdrs.get(key), hdrs.get(key).getClass().getName(),
+                              expHdrs.get(key), expHdrs.get(key).getClass().getName());
                     assert hdrs.get(key).equals(expHdrs.get(key));
                 });
             }