Merge branch 'v2.0.2'

.travis.yml

@@ -24,8 +24,8 @@ cache:
   - "$HOME/.gradle/wrapper/"
 after_success:
 - "./gradlew test jacocoTestReport coveralls"
-- "./gradlew uploadArchives"
-- "test ${TRAVIS_PULL_REQUEST} != 'true' && test ${TRAVIS_BRANCH} = 'master' && ./gradlew closeAndReleaseRepository"
+- test ${TRAVIS_BRANCH} != 'master' && "./gradlew uploadArchives" # Disable master for now, it fails due to ip address issues
+#Disable for now, upload archives fails because of IP address changes - "test ${TRAVIS_PULL_REQUEST} != 'true' && test ${TRAVIS_BRANCH} = 'master' && ./gradlew closeAndReleaseRepository"
 env:
   global:
   - gnatsd_version=v1.2.0

.travis/deploying.md

@@ -5,7 +5,9 @@
 
 There are currently two steps to the deployment, travis + sonatype, with a few random extra bits of knowledge.
 
-Travis will either deploy a snapshot or a release based on the version in build.gradle. If you deploy a release build, you will need to manually go to sonatype to release it. Those builds are found in the staging area. (We need to try to automate this in the future.)
+Travis doesn't support sonatype deploy correctly. There is an issue where the various artifacts get split across multiple repositories. That code has been deleted from the travis file and manual releases are required.
+
+~~Travis will either deploy a snapshot or a release based on the version in build.gradle. If you deploy a release build, you will need to manually go to sonatype to release it. Those builds are found in the staging area. (We need to try to automate this in the future.)~~
 
 ## Important note about release repositories
 

CHANGELOG.md

@@ -1,6 +1,10 @@
 
 # Change Log
 
+## Version 2.0.0
+
+* [FIXED] In a cluster situation the library wasn't using each server's auth info if it was in the URI.
+
 ## Version 2.0.1
 
 * [CHANGED] Request now returns a CompletableFuture to allow more application async options

src/main/java/io/nats/client/Options.java

@@ -855,6 +855,9 @@ public Builder reconnectBufferSize(long size) {
         /**
          * Set the username and password for basic authentication.
          * 
+         * If the user and password are set in the server URL, they will override these values. However, in a clustering situation,
+         * these values can be used as a fallback.
+         * 
          * @param userName a non-empty user name
          * @param password the password, in plain text
          * @return the Builder for chaining
@@ -868,6 +871,8 @@ public Builder userInfo(String userName, String password) {
         /**
          * Set the token for token-based authentication.
          * 
+         * If a token is provided in a server URI it overrides this value.
+         * 
          * @param token The token
          * @return the Builder for chaining
          */
@@ -941,21 +946,6 @@ public Options build() throws IllegalStateException {
                 server(DEFAULT_URL);
             } else if (servers.size() == 1) { // Allow some URI based configs
                 URI serverURI = servers.get(0);
-
-                if (this.username==null && this.password==null && this.token == null) {
-                    String userInfo = serverURI.getUserInfo();
-
-                    if (userInfo != null) {
-                        String[] info = userInfo.split(":");
-
-                        if (info.length == 2) {
-                            this.username = info[0];
-                            this.password = info[1];
-                        } else {
-                            this.token = userInfo;
-                        }
-                    }
-                }
 
                 if ("tls".equals(serverURI.getScheme()) && this.sslContext == null)
                 {
@@ -1188,10 +1178,13 @@ public boolean isOldRequestStyle() {
     /**
      * Create the options string sent with a connect message.
      * 
+     * If includeAuth is true the auth information is included:
+     * If the server URIs have auth info it is used. Otherwise the userInfo is used.
+     * 
      * @param includeAuth tells the options to build a connection string that includes auth information
      * @return the options String, basically JSON
      */
-    public String buildProtocolConnectOptionsString(boolean includeAuth) {
+    public String buildProtocolConnectOptionsString(String serverURI, boolean includeAuth) {
         StringBuilder connectString = new StringBuilder();
         connectString.append("{");
 
@@ -1210,15 +1203,44 @@ public String buildProtocolConnectOptionsString(boolean includeAuth) {
         appendOption(connectString, Options.OPTION_ECHO, String.valueOf(!this.isNoEcho()), false, true);
 
         if (includeAuth) {
-            if (this.username != null) {
+            String uriUser = null;
+            String uriPass = null;
+            String uriToken = null;
+
+            // Values from URI override options
+            try {
+                URI uri = new URI(serverURI);
+                String userInfo = uri.getUserInfo();
+
+                if (userInfo != null) {
+                    String[] info = userInfo.split(":");
+
+                    if (info.length == 2) {
+                        uriUser = info[0];
+                        uriPass = info[1];
+                    } else {
+                        uriToken = userInfo;
+                    }
+                }
+            } catch(URISyntaxException e) {
+                uriUser = uriToken = uriPass = null;
+            }
+
+            if (uriUser != null) {
+                appendOption(connectString, Options.OPTION_USER, uriUser, true, true);
+            } else if (this.username != null) {
                 appendOption(connectString, Options.OPTION_USER, this.username, true, true);
             }
 
-            if (this.password != null) {
+            if (uriPass != null) {
+                appendOption(connectString, Options.OPTION_PASSWORD, uriPass, true, true);
+            } else if (this.password != null) {
                 appendOption(connectString, Options.OPTION_PASSWORD, this.password, true, true);
             }
 
-            if (this.token != null) {
+            if (uriToken != null) {
+                appendOption(connectString, Options.OPTION_AUTH_TOKEN, uriToken, true, true);
+            } else if (this.token != null) {
                 appendOption(connectString, Options.OPTION_AUTH_TOKEN, this.token, true, true);
             }
         }

src/main/java/io/nats/client/impl/NatsConnection.java

@@ -293,7 +293,7 @@ void tryToConnect(String serverURI) {
             this.reader.start(this.dataPortFuture);
             this.writer.start(this.dataPortFuture);
 
-            this.sendConnect();
+            this.sendConnect(serverURI);
             Future<Boolean> pongFuture = sendPing();
             pongFuture.get(connectTimeout.toNanos(), TimeUnit.NANOSECONDS);
 
@@ -884,12 +884,12 @@ public void flush(Duration timeout) throws TimeoutException, InterruptedExceptio
         }
     }
 
-    void sendConnect() {
+    void sendConnect(String serverURI) {
         NatsServerInfo info = this.serverInfo.get();
         StringBuilder connectString = new StringBuilder();
         connectString.append(NatsConnection.OP_CONNECT);
         connectString.append(" ");
-        String connectOptions = this.options.buildProtocolConnectOptionsString(info.isAuthRequired());
+        String connectOptions = this.options.buildProtocolConnectOptionsString(serverURI, info.isAuthRequired());
         connectString.append(connectOptions);
         NatsMessage msg = new NatsMessage(connectString.toString());
         queueOutgoing(msg);

src/test/java/io/nats/client/AuthTests.java

@@ -13,12 +13,17 @@
 
 package io.nats.client;
 
+import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 
 import java.io.IOException;
+import java.time.Duration;
+import java.util.concurrent.TimeUnit;
 
 import org.junit.Test;
 
+import io.nats.client.ConnectionListener.Events;
+
 public class AuthTests {
     @Test
     public void testUserPass() throws Exception {
@@ -84,6 +89,128 @@ public void testUserPassInURL() throws Exception {
         }
     }
 
+    @Test
+    public void testUserPassInURLClusteredWithDifferentUser() throws Exception {
+        String[] customArgs1 = {"--user","stephen","--pass","password"};
+        String[] customArgs2 = {"--user","alberto","--pass","casadecampo"};
+        TestHandler handler = new TestHandler();
+        try (NatsTestServer ts1 = new NatsTestServer(customArgs1, false);
+                NatsTestServer ts2 = new NatsTestServer(customArgs2, false)) {
+            // See config file for user/pass
+            Options options = new Options.Builder().
+                        server("nats://stephen:password@localhost:"+ts1.getPort()).
+                        server("nats://alberto:casadecampo@localhost:"+ts2.getPort()).
+                        maxReconnects(4).
+                        noRandomize().
+                        connectionListener(handler).
+                        pingInterval(Duration.ofMillis(100)).
+                        build();
+            Connection nc = Nats.connect(options);
+            assertTrue("Connected Status", Connection.Status.CONNECTED == nc.getStatus());
+            assertEquals(nc.getConnectedUrl(), "nats://stephen:password@localhost:"+ts1.getPort());
+
+            handler.prepForStatusChange(Events.RESUBSCRIBED);
+            ts1.close();
+            handler.waitForStatusChange(2, TimeUnit.SECONDS);
+
+            assertTrue("Connected Status", Connection.Status.CONNECTED == nc.getStatus());
+            assertEquals(nc.getConnectedUrl(), "nats://alberto:casadecampo@localhost:"+ts2.getPort());
+            nc.close();
+        }
+    }
+
+    @Test
+    public void testUserPassInURLWithFallback() throws Exception {
+        String[] customArgs1 = {"--user","stephen","--pass","password"};
+        String[] customArgs2 = {"--user","alberto","--pass","casadecampo"};
+        TestHandler handler = new TestHandler();
+        try (NatsTestServer ts1 = new NatsTestServer(customArgs1, false);
+                NatsTestServer ts2 = new NatsTestServer(customArgs2, false)) {
+            // See config file for user/pass
+            Options options = new Options.Builder().
+                        server("nats://stephen:password@localhost:"+ts1.getPort()).
+                        server("nats://localhost:"+ts2.getPort()).
+                        userInfo("alberto", "casadecampo").
+                        maxReconnects(4).
+                        noRandomize().
+                        connectionListener(handler).
+                        pingInterval(Duration.ofMillis(100)).
+                        build();
+            Connection nc = Nats.connect(options);
+            assertTrue("Connected Status", Connection.Status.CONNECTED == nc.getStatus());
+            assertEquals(nc.getConnectedUrl(), "nats://stephen:password@localhost:"+ts1.getPort());
+
+            handler.prepForStatusChange(Events.RESUBSCRIBED);
+            ts1.close();
+            handler.waitForStatusChange(2, TimeUnit.SECONDS);
+
+            assertTrue("Connected Status", Connection.Status.CONNECTED == nc.getStatus());
+            assertEquals(nc.getConnectedUrl(), "nats://localhost:"+ts2.getPort());
+            nc.close();
+        }
+    }
+
+    @Test
+    public void testTokenInURLClusteredWithDifferentUser() throws Exception {
+        String[] customArgs1 = {"--auth","token_one"};
+        String[] customArgs2 = {"--auth","token_two"};
+        TestHandler handler = new TestHandler();
+        try (NatsTestServer ts1 = new NatsTestServer(customArgs1, false);
+                NatsTestServer ts2 = new NatsTestServer(customArgs2, false)) {
+            // See config file for user/pass
+            Options options = new Options.Builder().
+                        server("nats://token_one@localhost:"+ts1.getPort()).
+                        server("nats://token_two@localhost:"+ts2.getPort()).
+                        maxReconnects(4).
+                        noRandomize().
+                        connectionListener(handler).
+                        pingInterval(Duration.ofMillis(100)).
+                        build();
+            Connection nc = Nats.connect(options);
+            assertTrue("Connected Status", Connection.Status.CONNECTED == nc.getStatus());
+            assertEquals(nc.getConnectedUrl(), "nats://token_one@localhost:"+ts1.getPort());
+
+            handler.prepForStatusChange(Events.RESUBSCRIBED);
+            ts1.close();
+            handler.waitForStatusChange(2, TimeUnit.SECONDS);
+
+            assertTrue("Connected Status", Connection.Status.CONNECTED == nc.getStatus());
+            assertEquals(nc.getConnectedUrl(), "nats://token_two@localhost:"+ts2.getPort());
+            nc.close();
+        }
+    }
+
+    @Test
+    public void testTokenInURLWithFallback() throws Exception {
+        String[] customArgs1 = {"--auth","token_one"};
+        String[] customArgs2 = {"--auth","token_two"};
+        TestHandler handler = new TestHandler();
+        try (NatsTestServer ts1 = new NatsTestServer(customArgs1, false);
+                NatsTestServer ts2 = new NatsTestServer(customArgs2, false)) {
+            // See config file for user/pass
+            Options options = new Options.Builder().
+                        server("nats://token_one@localhost:"+ts1.getPort()).
+                        server("nats://localhost:"+ts2.getPort()).
+                        token("token_two").
+                        maxReconnects(4).
+                        noRandomize().
+                        connectionListener(handler).
+                        pingInterval(Duration.ofMillis(100)).
+                        build();
+            Connection nc = Nats.connect(options);
+            assertTrue("Connected Status", Connection.Status.CONNECTED == nc.getStatus());
+            assertEquals(nc.getConnectedUrl(), "nats://token_one@localhost:"+ts1.getPort());
+
+            handler.prepForStatusChange(Events.RESUBSCRIBED);
+            ts1.close();
+            handler.waitForStatusChange(2, TimeUnit.SECONDS);
+
+            assertTrue("Connected Status", Connection.Status.CONNECTED == nc.getStatus());
+            assertEquals(nc.getConnectedUrl(), "nats://localhost:"+ts2.getPort());
+            nc.close();
+        }
+    }
+
     @Test
     public void testToken() throws Exception {
         String[] customArgs = {"--auth","derek"};

src/test/java/io/nats/client/OptionsTests.java

@@ -286,7 +286,7 @@ public void testDefaultConnectOptions() {
         Options o = new Options.Builder().build();
         String expected = "{\"lang\":\"java\",\"version\":\"" + Nats.CLIENT_VERSION + "\""
                 + ",\"protocol\":1,\"verbose\":false,\"pedantic\":false,\"tls_required\":false,\"echo\":true}";
-        assertEquals("default connect options", expected, o.buildProtocolConnectOptionsString(false));
+        assertEquals("default connect options", expected, o.buildProtocolConnectOptionsString("nats://localhost:4222", false));
     }
 
     @Test
@@ -295,7 +295,7 @@ public void testConnectOptionsWithNameAndContext() throws Exception {
         Options o = new Options.Builder().sslContext(ctx).connectionName("c1").build();
         String expected = "{\"lang\":\"java\",\"version\":\"" + Nats.CLIENT_VERSION + "\",\"name\":\"c1\""
                 + ",\"protocol\":1,\"verbose\":false,\"pedantic\":false,\"tls_required\":true,\"echo\":true}";
-        assertEquals("default connect options", expected, o.buildProtocolConnectOptionsString(false));
+        assertEquals("default connect options", expected, o.buildProtocolConnectOptionsString("nats://localhost:4222", false));
     }
 
     @Test
@@ -306,8 +306,8 @@ public void testAuthConnectOptions() {
         String expectedWithAuth = "{\"lang\":\"java\",\"version\":\"" + Nats.CLIENT_VERSION + "\""
                 + ",\"protocol\":1,\"verbose\":false,\"pedantic\":false,\"tls_required\":false,\"echo\":true"
                 + ",\"user\":\"hello\",\"pass\":\"world\"}";
-        assertEquals("no auth connect options", expectedNoAuth, o.buildProtocolConnectOptionsString(false));
-        assertEquals("auth connect options", expectedWithAuth, o.buildProtocolConnectOptionsString(true));
+        assertEquals("no auth connect options", expectedNoAuth, o.buildProtocolConnectOptionsString("nats://localhost:4222", false));
+        assertEquals("auth connect options", expectedWithAuth, o.buildProtocolConnectOptionsString("nats://localhost:4222", true));
     }
 
     @Test
@@ -333,20 +333,24 @@ public void testPropertyDataPortType() {
 
     @Test
     public void testUserPassInURL() {
-        Options o = new Options.Builder().server("nats://derek:password@localhost:2222").build();
+        String serverURI = "nats://derek:password@localhost:2222";
+        Options o = new Options.Builder().server(serverURI).build();
 
-        assertNull(o.getToken());
-        assertEquals("user from url", "derek", o.getUsername());
-        assertEquals("password from url", "password", o.getPassword());
+        String connectString = o.buildProtocolConnectOptionsString(serverURI, true);
+        assertTrue(connectString.contains("\"user\":\"derek\""));
+        assertTrue(connectString.contains("\"pass\":\"password\""));
+        assertFalse(connectString.contains("\"token\":"));
     }
 
     @Test
     public void testTokenInURL() {
-        Options o = new Options.Builder().server("nats://alberto@localhost:2222").build();
+        String serverURI = "nats://alberto@localhost:2222";
+        Options o = new Options.Builder().server(serverURI).build();
 
-        assertNull(o.getUsername());
-        assertNull(o.getPassword());
-        assertEquals("token from url", "alberto", o.getToken());
+        String connectString = o.buildProtocolConnectOptionsString(serverURI, true);
+        assertTrue(connectString.contains("\"auth_token\":\"alberto\""));
+        assertFalse(connectString.contains("\"user\":"));
+        assertFalse(connectString.contains("\"pass\":"));
     }
 
     @Test(expected=IllegalArgumentException.class)