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

Issue:
Apache Camel Conn:  server.config with vantiq address scrambles things
When using the server.config style address with targetServer containing 'vantiq', the connector seems to replace the vantiq part with https (overzealous replacement from vantiq://node...) (that's my current guess).

Via Mark:

```
$ ./camelConnector

[ttps://internal.vantiq.com/...] bSocketListener#EventHubSource INFO  WebSocket open

[ttps://internal.vantiq.com/...] bSocketListener#EventHubSource ERROR Map of the received message: {status=400, contentType=application/json, body=[{code=io.vantiq.sourcemgr.source.noConnectionLocal, message=The CONNECT_EXTENSION operation on data source mejfm_CamelTesting:EventHubSource did not complete successfully: EXTENSION source mejfm_CamelTesting:EventHubSource connectExtension operation attempted, but a connection already exists., params=[mejfm_CamelTesting:EventHubSource, EXTENSION source mejfm_CamelTesting:EventHubSource connectExtension operation attempted, but a connection already exists.]}]}

[ttps://internal.vantiq.com/...] bSocketListener#EventHubSource WARN  Error occurred attempting to connect to source.

[                      Thread-0] CamelCore                      ERROR Failed to connect to all sources.

[                      Thread-0] CamelCore                      ERROR Failed to connect within 10 seconds

[ttps://internal.vantiq.com/...] bSocketListener#EventHubSource INFO  WebSocket open

[ttps://internal.vantiq.com/...] bSocketListener#EventHubSource INFO  Successful connection to EventHubSource

[ttps://internal.vantiq.com/...] CamelHandleConfiguration       DEBUG Fetching contents of document eventhub.yml from path /docs/eventhub.yml.

[ttps://internal.vantiq.com/...] CamelRunner                    DEBUG Calling Run()

[ttps://internal.vantiq.com/...] CamelRunner                    DEBUG AwaitStarted()

[                      Thread-1] MainSupport                    INFO  Apache Camel (Main) 3.18.3 is starting

[                      Thread-1] CamelRunner                    DEBUG >>>Using context: camel-1, stopped: true

[                      Thread-1] CamelRunner                    DEBUG Loading route (specificationType yaml): - route:

   id: "EventHub Sink"

   from:

      uri: "vantiq://server.config"

      steps:

      - choice:

          when:

          - simple: "${header[partition-id]}"

            steps:

            - set-header:

                name: CamelAzureEventHubsPartitionId

                simple: "${header[partition-id]}"

          - simple: "${header[ce-partition-id]}"

            steps:

            - set-header:

                name: CamelAzureEventHubsPartitionId

                simple: "${header[ce-partition-id]}"

      - to:

          uri: "azure-eventhubs://mmunro-test/mmunrohub"

          parameters:

            sharedAccessName: "mmunroHubSAS"

            sharedAccessKey: "byxOS4q8zIn3wTRdGtWxbZaXpCohpxSnU+AEhAZML5c="

[                      Thread-1] CamelDiscovery                 DEBUG Discovering Camel (version 3.18.3) components using context: camel-2

[                      Thread-1] EnumeratingComponentResolver   DEBUG Resolver asked to resolve component: vantiq

[                      Thread-1] PlaceholderEndpoint            DEBUG PlaceholderEndpoint for vantiq://server.config called

[                      Thread-1] EnumeratingComponentResolver   DEBUG Resolver asked to resolve component: azure-eventhubs

[                      Thread-1] PlaceholderEndpoint            DEBUG PlaceholderEndpoint for azure-eventhubs://mmunro-test/mmunrohub?sharedAccessKey=byxOS4q8zIn3wTRdGtWxbZaXpCohpxSnU+AEhAZML5c%3D&sharedAccessName=mmunroHubSAS called

[                      Thread-1] AbstractCamelContext           INFO  Apache Camel 3.18.3 (camel-2) is starting

[                      Thread-1] AbstractCamelContext           INFO  Using 2 instances of same component class: io.vantiq.extsrc.camelconn.discover.PlaceholderComponent with names: azure-eventhubs, vantiq

[                      Thread-1] AbstractCamelContext           INFO  Routes startup (started:1)

[                      Thread-1] AbstractCamelContext           INFO      Started EventHub Sink (vantiq://server.config)

[                      Thread-1] AbstractCamelContext           INFO  Apache Camel 3.18.3 (camel-2) started in 99ms (build:1ms init:86ms start:12ms)

[                      Thread-1] CamelDiscovery                 DEBUG Component discovery complete:

[                      Thread-1] CamelDiscovery                 DEBUG     ---> azure-eventhubs

[                      Thread-1] CamelDiscovery                 DEBUG System components (ignored):

[                      Thread-1] CamelDiscovery                 DEBUG DataFormat discovery complete:

[                      Thread-1] CamelDiscovery                 DEBUG System dataformats (ignored):

[                      Thread-1] AbstractCamelContext           INFO  Apache Camel 3.18.3 (camel-2) is shutting down (timeout:45s)

[                      Thread-1] AbstractCamelContext           INFO  Routes stopped (stopped:1)

[                      Thread-1] AbstractCamelContext           INFO      Stopped EventHub Sink (vantiq://server.config)

[                      Thread-1] AbstractCamelContext           INFO  Apache Camel 3.18.3 (camel-2) shutdown in 10ms (uptime:0s)

:: loading settings :: url = jar:file:/C:/Data/ExtensionSource/13032023/camelConnector/camelConnector/lib/ivy-2.5.1.jar!/org/apache/ivy/core/settings/ivysettings.xml

[                      Thread-1] CamelResolver                  INFO  Resolving required libraries -- Component Resolution

[                      Thread-1] CamelResolver                  DEBUG CamelResolver EventHub, repos: [], cache: C:\Data\ExtensionSource\13032023\camelConnector\camelConnector\bin\componentCache, destination: C:\Data\ExtensionSource\13032023\camelConnector\camelConnector\bin\componentLib -- Making 64 artifacts available to Component Resolution

[                      Thread-1] VantiqEndpoint                 INFO  VantiqEndpoint creating consumer for uri: vantiq://internal.vantiq.com with sourceName: EventHubSource, accessToken: present

[                      Thread-1] AbstractCamelContext           INFO  Apache Camel 3.18.3 (camel-1) is starting

[                      Thread-1] EventHubClientBuilder          INFO  {"az.sdk.message":"Emitting a single connection.","connectionId":"MF_5ca9d2_1678708389286"}

[                      Thread-1] EventHubConnectionProcessor    INFO  {"az.sdk.message":"Setting next AMQP channel.","entityPath":"mmunrohub"}

[https://internal.https.com/...] bSocketListener#EventHubSource ERROR Failure occurred in listener

java.net.ProtocolException: Expected HTTP 101 response but was '200 OK'

        at okhttp3.internal.ws.RealWebSocket.checkUpgradeSuccess$okhttp(RealWebSocket.kt:224) [okhttp-4.9.3.jar:?]

        at okhttp3.internal.ws.RealWebSocket$connect$1.onResponse(RealWebSocket.kt:170) [okhttp-4.9.3.jar:?]

        at okhttp3.internal.connection.RealCall$AsyncCall.run(RealCall.kt:519) [okhttp-4.9.3.jar:?]

        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) [?:?]

        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) [?:?]

        at java.lang.Thread.run(Thread.java:833) [?:?]

[                      Thread-1] VantiqEndpoint                 ERROR Failed to connect to Vantiq server: vantiq://internal.vantiq.com (https://internal.https.com/).

[                      Thread-1] WebSocketClient#EventHubSource INFO  Websocket closed.

[                      Thread-1] EventHubConnectionProcessor    INFO  {"az.sdk.message":"Upstream connection publisher was completed. Terminating processor.","entityPath":"mmunrohub"}

[                      Thread-1] ReactorConnection              INFO  {"az.sdk.message":"Disposing of ReactorConnection.","connectionId":"MF_5ca9d2_1678708389286","isTransient":false,"isInitiatedByClient":true,"shutdownMessage":"Disposed by client."}

[                      Thread-1] EventHubConnectionProcessor    INFO  {"az.sdk.message":"Channel is disposed.","entityPath":"mmunrohub"}

[                      Thread-1] AbstractCamelContext           ERROR Error starting CamelContext (camel-1) due to exception thrown: org.apache.camel.CamelException: Failed to connect to Vantiq server: vantiq://internal.vantiq.com (https://internal.https.com/).

org.apache.camel.RuntimeCamelException: org.apache.camel.CamelException: Failed to connect to Vantiq server: vantiq://internal.vantiq.com (https://internal.https.com/).$ ./camelConnector

[ttps://internal.vantiq.com/...] bSocketListener#EventHubSource INFO  WebSocket open

[ttps://internal.vantiq.com/...] bSocketListener#EventHubSource ERROR Map of the received message: {status=400, contentType=application/json, body=[{code=io.vantiq.sourcemgr.source.noConnectionLocal, message=The CONNECT_EXTENSION operation on data source mejfm_CamelTesting:EventHubSource did not complete successfully: EXTENSION source mejfm_CamelTesting:EventHubSource connectExtension operation attempted, but a connection already exists., params=[mejfm_CamelTesting:EventHubSource, EXTENSION source mejfm_CamelTesting:EventHubSource connectExtension operation attempted, but a connection already exists.]}]}

[ttps://internal.vantiq.com/...] bSocketListener#EventHubSource WARN  Error occurred attempting to connect to source.

[                      Thread-0] CamelCore                      ERROR Failed to connect to all sources.

[                      Thread-0] CamelCore                      ERROR Failed to connect within 10 seconds

[ttps://internal.vantiq.com/...] bSocketListener#EventHubSource INFO  WebSocket open

[ttps://internal.vantiq.com/...] bSocketListener#EventHubSource INFO  Successful connection to EventHubSource

[ttps://internal.vantiq.com/...] CamelHandleConfiguration       DEBUG Fetching contents of document eventhub.yml from path /docs/eventhub.yml.

[ttps://internal.vantiq.com/...] CamelRunner                    DEBUG Calling Run()

[ttps://internal.vantiq.com/...] CamelRunner                    DEBUG AwaitStarted()

[                      Thread-1] MainSupport                    INFO  Apache Camel (Main) 3.18.3 is starting

[                      Thread-1] CamelRunner                    DEBUG >>>Using context: camel-1, stopped: true

[                      Thread-1] CamelRunner                    DEBUG Loading route (specificationType yaml): - route:

   id: "EventHub Sink"

   from:

      uri: "vantiq://server.config"

      steps:

      - choice:

          when:

          - simple: "${header[partition-id]}"

            steps:

            - set-header:

                name: CamelAzureEventHubsPartitionId

                simple: "${header[partition-id]}"

          - simple: "${header[ce-partition-id]}"

            steps:

            - set-header:

                name: CamelAzureEventHubsPartitionId

                simple: "${header[ce-partition-id]}"

      - to:

          uri: "azure-eventhubs://mmunro-test/mmunrohub"

          parameters:

            sharedAccessName: "mmunroHubSAS"

            sharedAccessKey: "byxOS4q8zIn3wTRdGtWxbZaXpCohpxSnU+AEhAZML5c="

[                      Thread-1] CamelDiscovery                 DEBUG Discovering Camel (version 3.18.3) components using context: camel-2

[                      Thread-1] EnumeratingComponentResolver   DEBUG Resolver asked to resolve component: vantiq

[                      Thread-1] PlaceholderEndpoint            DEBUG PlaceholderEndpoint for vantiq://server.config called

[                      Thread-1] EnumeratingComponentResolver   DEBUG Resolver asked to resolve component: azure-eventhubs

[                      Thread-1] PlaceholderEndpoint            DEBUG PlaceholderEndpoint for azure-eventhubs://mmunro-test/mmunrohub?sharedAccessKey=byxOS4q8zIn3wTRdGtWxbZaXpCohpxSnU+AEhAZML5c%3D&sharedAccessName=mmunroHubSAS called

[                      Thread-1] AbstractCamelContext           INFO  Apache Camel 3.18.3 (camel-2) is starting

[                      Thread-1] AbstractCamelContext           INFO  Using 2 instances of same component class: io.vantiq.extsrc.camelconn.discover.PlaceholderComponent with names: azure-eventhubs, vantiq

[                      Thread-1] AbstractCamelContext           INFO  Routes startup (started:1)

[                      Thread-1] AbstractCamelContext           INFO      Started EventHub Sink (vantiq://server.config)

[                      Thread-1] AbstractCamelContext           INFO  Apache Camel 3.18.3 (camel-2) started in 99ms (build:1ms init:86ms start:12ms)

[                      Thread-1] CamelDiscovery                 DEBUG Component discovery complete:

[                      Thread-1] CamelDiscovery                 DEBUG     ---> azure-eventhubs

[                      Thread-1] CamelDiscovery                 DEBUG System components (ignored):

[                      Thread-1] CamelDiscovery                 DEBUG DataFormat discovery complete:

[                      Thread-1] CamelDiscovery                 DEBUG System dataformats (ignored):

[                      Thread-1] AbstractCamelContext           INFO  Apache Camel 3.18.3 (camel-2) is shutting down (timeout:45s)

[                      Thread-1] AbstractCamelContext           INFO  Routes stopped (stopped:1)

[                      Thread-1] AbstractCamelContext           INFO      Stopped EventHub Sink (vantiq://server.config)

[                      Thread-1] AbstractCamelContext           INFO  Apache Camel 3.18.3 (camel-2) shutdown in 10ms (uptime:0s)

:: loading settings :: url = jar:file:/C:/Data/ExtensionSource/13032023/camelConnector/camelConnector/lib/ivy-2.5.1.jar!/org/apache/ivy/core/settings/ivysettings.xml

[                      Thread-1] CamelResolver                  INFO  Resolving required libraries -- Component Resolution

[                      Thread-1] CamelResolver                  DEBUG CamelResolver EventHub, repos: [], cache: C:\Data\ExtensionSource\13032023\camelConnector\camelConnector\bin\componentCache, destination: C:\Data\ExtensionSource\13032023\camelConnector\camelConnector\bin\componentLib -- Making 64 artifacts available to Component Resolution

[                      Thread-1] VantiqEndpoint                 INFO  VantiqEndpoint creating consumer for uri: vantiq://internal.vantiq.com with sourceName: EventHubSource, accessToken: present

[                      Thread-1] AbstractCamelContext           INFO  Apache Camel 3.18.3 (camel-1) is starting

[                      Thread-1] EventHubClientBuilder          INFO  {"az.sdk.message":"Emitting a single connection.","connectionId":"MF_5ca9d2_1678708389286"}

[                      Thread-1] EventHubConnectionProcessor    INFO  {"az.sdk.message":"Setting next AMQP channel.","entityPath":"mmunrohub"}

[https://internal.https.com/...] bSocketListener#EventHubSource ERROR Failure occurred in listener

java.net.ProtocolException: Expected HTTP 101 response but was '200 OK'

        at okhttp3.internal.ws.RealWebSocket.checkUpgradeSuccess$okhttp(RealWebSocket.kt:224) [okhttp-4.9.3.jar:?]

        at okhttp3.internal.ws.RealWebSocket$connect$1.onResponse(RealWebSocket.kt:170) [okhttp-4.9.3.jar:?]

        at okhttp3.internal.connection.RealCall$AsyncCall.run(RealCall.kt:519) [okhttp-4.9.3.jar:?]

        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) [?:?]

        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) [?:?]

        at java.lang.Thread.run(Thread.java:833) [?:?]

[                      Thread-1] VantiqEndpoint                 ERROR Failed to connect to Vantiq server: vantiq://internal.vantiq.com (https://internal.https.com/).

[                      Thread-1] WebSocketClient#EventHubSource INFO  Websocket closed.

[                      Thread-1] EventHubConnectionProcessor    INFO  {"az.sdk.message":"Upstream connection publisher was completed. Terminating processor.","entityPath":"mmunrohub"}

[                      Thread-1] ReactorConnection              INFO  {"az.sdk.message":"Disposing of ReactorConnection.","connectionId":"MF_5ca9d2_1678708389286","isTransient":false,"isInitiatedByClient":true,"shutdownMessage":"Disposed by client."}

[                      Thread-1] EventHubConnectionProcessor    INFO  {"az.sdk.message":"Channel is disposed.","entityPath":"mmunrohub"}

[                      Thread-1] AbstractCamelContext           ERROR Error starting CamelContext (camel-1) due to exception thrown: org.apache.camel.CamelException: Failed to connect to Vantiq server: vantiq://internal.vantiq.com (https://internal.https.com/).

org.apache.camel.RuntimeCamelException: org.apache.camel.CamelException: Failed to connect to Vantiq server: vantiq://internal.vantiq.com (https://internal.https.com/).
```


Discussion / hints:

Fix patch:
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 c2b473c7..1f2c1fa6 100644
--- a/camelComponent/src/main/java/io/vantiq/extsrc/camel/VantiqEndpoint.java
+++ b/camelComponent/src/main/java/io/vantiq/extsrc/camel/VantiqEndpoint.java
@@ -81,7 +81,7 @@ public class VantiqEndpoint extends DefaultEndpoint {
     
     private String correctedVantiqUrl = null;
     private boolean started = false;
-    private String startStopLock = "startStopLock";
+    private final String startStopLock = "startStopLock";
     private boolean runningInConnector = false;
     
     @Getter
@@ -187,13 +187,8 @@ public void startup() throws CamelException {
             try {
                 log.debug("Attempting to connect to URL: {} from {}", getEndpointBaseUri(), getEndpointUri());
                 String vtq = getEndpointBaseUri();
-                String protocol = "https";
-                if (noSsl) {
-                    // This is primarily for use in development.  But supported...
-                    protocol = "http";
-                }
-                correctedVantiqUrl = vtq.replace("vantiq", protocol);
-                log.trace("Fixed-up Vantiq URL: {}", correctedVantiqUrl);
+               
+                correctedVantiqUrl = adjustVantiqTarget(vtq, noSsl);
                 utils.provideServerConfig(correctedVantiqUrl, accessToken, sourceName, sendPings, null);
                 buildEndpointName();
         
@@ -308,6 +303,41 @@ public String createEndpointUri() {
         }
     }
     
+    /**
+     * Adjust Vantiq URL.
+     *
+     * If using a standard vantiq:// connection, set the adjusted URL to a more reasonable protocol.  If the URL does
+     * not have a scheme of "vantiq", then leave it alone.
+     *
+     * @param baseUrl String the baseURL to adjusted
+     * @param noSsl boolean indicating whether we've been told to skip the SSL handling
+     *                          (i.e.. use http rather than https)
+     * @return
+     */
+    public static String adjustVantiqTarget(String baseUrl, boolean noSsl) throws CamelException {
+        String correctedUrl = null;
+        String protocol = "https";
+        if (noSsl) {
+            // This is used in development or edge servers.
+            protocol = "http";
+        }
+        try {
+            URI vtqUri = new URI(baseUrl);
+            if (vtqUri.getScheme().equalsIgnoreCase("vantiq")) {
+                vtqUri = new URI(protocol, null, vtqUri.getHost(), vtqUri.getPort(), vtqUri.getPath(), null,
+                                 null);
+                correctedUrl = vtqUri.toASCIIString();
+            } else {
+                correctedUrl = baseUrl;
+            }
+            log.trace("Adjusted Vantiq URL: {} (from {})", correctedUrl, baseUrl);
+        } catch (URISyntaxException uriSE) {
+            throw new CamelException("Unable to connect to provided URI: " + baseUrl +
+                                             " (adjusted: " + correctedUrl + ")", uriSE);
+        }
+        return correctedUrl;
+    }
+    
     @Override
     public void doStop() {
         synchronized (startStopLock) {


Test patch:
diff --git a/camelComponent/src/test/java/io/vantiq/extsrc/camel/ClientRegistryTest.java b/camelComponent/src/test/java/io/vantiq/extsrc/camel/ClientRegistryTest.java
index b055d1f1..90d8409a 100644
--- a/camelComponent/src/test/java/io/vantiq/extsrc/camel/ClientRegistryTest.java
+++ b/camelComponent/src/test/java/io/vantiq/extsrc/camel/ClientRegistryTest.java
@@ -1,11 +1,13 @@
 package io.vantiq.extsrc.camel;
 
+import static org.assertj.core.api.Fail.fail;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 
 import io.vantiq.extjsdk.ExtensionWebSocketClient;
 import io.vantiq.extsrc.camel.utils.ClientRegistry;
+import org.apache.camel.CamelException;
 import org.junit.After;
 import org.junit.Test;
 
@@ -100,6 +102,44 @@ public void testServerUrlWSpaces() {
         assertEquals("Known client count", 1, ClientRegistry.getKnownClientCount());
     }
     
+    @Test
+    public void testUriAdjustment() {
+        String newUrl = null;
+        try {
+            newUrl = VantiqEndpoint.adjustVantiqTarget("vantiq://somewhere", false);
+            assert newUrl.equals("https://somewhere");
+    
+            newUrl = VantiqEndpoint.adjustVantiqTarget("vantiq://somewhere", false);
+            assert newUrl.equals("https://somewhere");
+    
+            newUrl = VantiqEndpoint.adjustVantiqTarget("https://internal.vantiq.com", false);
+            assert newUrl.equals("https://internal.vantiq.com");
+    
+            newUrl = VantiqEndpoint.adjustVantiqTarget("https://internal.vantiq.com", true);
+            assert newUrl.equals("https://internal.vantiq.com");
+    
+            newUrl = VantiqEndpoint.adjustVantiqTarget("vantiq://internal.vantiq.com", false);
+            assert newUrl.equals("https://internal.vantiq.com");
+    
+            newUrl = VantiqEndpoint.adjustVantiqTarget("vantiq://internal.vantiq.com", true);
+            assert newUrl.equals("http://internal.vantiq.com");
+    
+            newUrl = VantiqEndpoint.adjustVantiqTarget("https://vantiqnode:8080", false);
+            assert newUrl.equals("https://vantiqnode:8080");
+    
+            newUrl = VantiqEndpoint.adjustVantiqTarget("https://vantiqnode:8080", true);
+            assert newUrl.equals("https://vantiqnode:8080");
+    
+            newUrl = VantiqEndpoint.adjustVantiqTarget("vantiq://vantiqnode:8080", false);
+            assert newUrl.equals("https://vantiqnode:8080");
+    
+            newUrl = VantiqEndpoint.adjustVantiqTarget("vantiq://vantiqnode:8080", true);
+            assert newUrl.equals("http://vantiqnode:8080");
+        } catch (CamelException ce) {
+            fail("Could not complete adjustment for URI: " + newUrl, ce);
+        }
+    }
+    
     public void buildAndVerify(String srcName, String target, boolean shouldBeCreated) {
         // Lambda's can only reference "effectively final" values, so we'll wrap in an AtomicReference
         AtomicReference<ExtensionWebSocketClient> client = new AtomicReference<>();
diff --git a/camelComponent/src/test/java/io/vantiq/extsrc/camel/VantiqLiveComponentTest.java b/camelComponent/src/test/java/io/vantiq/extsrc/camel/VantiqLiveComponentTest.java
index 8d981c1e..7f39e9e5 100644
--- a/camelComponent/src/test/java/io/vantiq/extsrc/camel/VantiqLiveComponentTest.java
+++ b/camelComponent/src/test/java/io/vantiq/extsrc/camel/VantiqLiveComponentTest.java
@@ -434,7 +434,6 @@ public static void createSource() {
         sourceDef.put("name", testSourceName);
         sourceDef.put("type", SRC_IMPL_TYPE);
         sourceDef.put("active", "true");
-        sourceDef.put("direction", "BOTH");
         sourceDef.put("config", new LinkedHashMap<String, Object>());
     
         VantiqResponse insertResponse = vantiq.insert("system.sources", sourceDef);
@@ -454,7 +453,6 @@ public static void createQuerySource() {
         sourceDef.put("name", testQuerySourceName);
         sourceDef.put("type", SRC_IMPL_TYPE);
         sourceDef.put("active", "true");
-        sourceDef.put("direction", "BOTH");
         sourceDef.put("config", new LinkedHashMap<String, Object>());
         
         VantiqResponse insertResponse = vantiq.insert("system.sources", sourceDef);