Make URI-to-ClientParameters conversion public

src/main/java/com/rabbitmq/stream/impl/Client.java

@@ -119,6 +119,7 @@
 import java.net.ConnectException;
 import java.net.InetSocketAddress;
 import java.net.SocketAddress;
+import java.net.URI;
 import java.nio.charset.Charset;
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
@@ -2467,6 +2468,11 @@ static ClientParameters cp() {
     return new ClientParameters();
   }
 
+  public static ClientParameters maybeSetUpClientParametersFromUris(
+      List<URI> uris, ClientParameters clientParametersPrototype) {
+    return Utils.maybeSetUpClientParametersFromUris(uris, clientParametersPrototype);
+  }
+
   public static class ClientParameters {
 
     private final Map<String, String> clientProperties = new ConcurrentHashMap<>();

src/main/java/com/rabbitmq/stream/impl/StreamEnvironment.java

@@ -16,6 +16,7 @@
 
 import static com.rabbitmq.stream.impl.AsyncRetry.asyncRetry;
 import static com.rabbitmq.stream.impl.Client.DEFAULT_RPC_TIMEOUT;
+import static com.rabbitmq.stream.impl.Client.maybeSetUpClientParametersFromUris;
 import static com.rabbitmq.stream.impl.ThreadUtils.threadFactory;
 import static com.rabbitmq.stream.impl.Utils.AVAILABLE_PROCESSORS;
 import static com.rabbitmq.stream.impl.Utils.DEFAULT_ADDRESS_RESOLVER;
@@ -65,9 +66,7 @@
 import io.netty.channel.EventLoopGroup;
 import io.netty.handler.ssl.SslContext;
 import io.netty.handler.ssl.SslContextBuilder;
-import java.io.IOException;
 import java.net.URI;
-import java.net.URLDecoder;
 import java.time.Duration;
 import java.time.LocalDateTime;
 import java.util.Collections;
@@ -476,57 +475,6 @@ private Locator locator(int i) {
     return this.locators.get(i);
   }
 
-  private static String uriDecode(String s) {
-    try {
-      // URLDecode decodes '+' to a space, as for
-      // form encoding. So protect plus signs.
-      return URLDecoder.decode(s.replace("+", "%2B"), "US-ASCII");
-    } catch (IOException e) {
-      throw new IllegalArgumentException(e);
-    }
-  }
-
-  Client.ClientParameters maybeSetUpClientParametersFromUris(
-      List<URI> uris, Client.ClientParameters clientParametersPrototype) {
-    if (uris.isEmpty()) {
-      return clientParametersPrototype;
-    } else {
-      URI uri = uris.get(0);
-      clientParametersPrototype = clientParametersPrototype.duplicate();
-      String host = uri.getHost();
-      if (host != null) {
-        clientParametersPrototype.host(host);
-      }
-
-      int port = uri.getPort();
-      if (port != -1) {
-        clientParametersPrototype.port(port);
-      }
-
-      String userInfo = uri.getRawUserInfo();
-      if (userInfo != null) {
-        String[] userPassword = userInfo.split(":");
-        if (userPassword.length > 2) {
-          throw new IllegalArgumentException("Bad user info in URI " + userInfo);
-        }
-
-        clientParametersPrototype.username(uriDecode(userPassword[0]));
-        if (userPassword.length == 2) {
-          clientParametersPrototype.password(uriDecode(userPassword[1]));
-        }
-      }
-
-      String path = uri.getRawPath();
-      if (path != null && path.length() > 0) {
-        if (path.indexOf('/', 1) != -1) {
-          throw new IllegalArgumentException("Multiple segments in path of URI: " + path);
-        }
-        clientParametersPrototype.virtualHost(uriDecode(uri.getPath().substring(1)));
-      }
-      return clientParametersPrototype;
-    }
-  }
-
   public ByteBufAllocator byteBufAllocator() {
     return byteBufAllocator;
   }

src/main/java/com/rabbitmq/stream/impl/Utils.java

@@ -15,6 +15,7 @@
 package com.rabbitmq.stream.impl;
 
 import static java.lang.String.format;
+import static java.nio.charset.StandardCharsets.US_ASCII;
 import static java.util.Map.copyOf;
 
 import com.rabbitmq.stream.Address;
@@ -33,6 +34,8 @@
 import io.netty.channel.MultiThreadIoEventLoopGroup;
 import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.channel.nio.NioIoHandler;
+import java.net.URI;
+import java.net.URLDecoder;
 import java.net.UnknownHostException;
 import java.security.cert.X509Certificate;
 import java.time.Duration;
@@ -534,6 +537,51 @@ static boolean is3_11_OrMore(String brokerVersion) {
     return versionCompare(currentVersion(brokerVersion), "3.11.0") >= 0;
   }
 
+  static Client.ClientParameters maybeSetUpClientParametersFromUris(
+      List<URI> uris, Client.ClientParameters clientParametersPrototype) {
+    if (!uris.isEmpty()) {
+      URI uri = uris.get(0);
+      clientParametersPrototype = clientParametersPrototype.duplicate();
+      String host = uri.getHost();
+      if (host != null) {
+        clientParametersPrototype.host(host);
+      }
+
+      int port = uri.getPort();
+      if (port != -1) {
+        clientParametersPrototype.port(port);
+      }
+
+      String userInfo = uri.getRawUserInfo();
+      if (userInfo != null) {
+        String[] userPassword = userInfo.split(":");
+        if (userPassword.length > 2) {
+          throw new IllegalArgumentException("Bad user info in URI " + userInfo);
+        }
+
+        clientParametersPrototype.username(uriDecode(userPassword[0]));
+        if (userPassword.length == 2) {
+          clientParametersPrototype.password(uriDecode(userPassword[1]));
+        }
+      }
+
+      String path = uri.getRawPath();
+      if (path != null && path.length() > 0) {
+        if (path.indexOf('/', 1) != -1) {
+          throw new IllegalArgumentException("Multiple segments in path of URI: " + path);
+        }
+        clientParametersPrototype.virtualHost(uriDecode(uri.getPath().substring(1)));
+      }
+    }
+    return clientParametersPrototype;
+  }
+
+  private static String uriDecode(String s) {
+    // URLDecode decodes '+' to a space, as for
+    // form encoding. So protect plus signs.
+    return URLDecoder.decode(s.replace("+", "%2B"), US_ASCII);
+  }
+
   static StreamException convertCodeToException(
       short responseCode, String stream, Supplier<String> fallbackMessage) {
     if (responseCode == Constants.RESPONSE_CODE_STREAM_DOES_NOT_EXIST) {

src/test/java/com/rabbitmq/stream/impl/UtilsTest.java

@@ -20,7 +20,9 @@
 import static com.rabbitmq.stream.impl.Utils.defaultConnectionNamingStrategy;
 import static com.rabbitmq.stream.impl.Utils.formatConstant;
 import static com.rabbitmq.stream.impl.Utils.offsetBefore;
+import static java.net.URI.create;
 import static java.util.Collections.emptyList;
+import static java.util.List.of;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.mock;
@@ -34,7 +36,10 @@
 import com.rabbitmq.stream.impl.Utils.ClientFactory;
 import com.rabbitmq.stream.impl.Utils.ClientFactoryContext;
 import com.rabbitmq.stream.impl.Utils.ConditionalClientFactory;
+import com.rabbitmq.stream.sasl.DefaultUsernamePasswordCredentialsProvider;
+import com.rabbitmq.stream.sasl.UsernamePasswordCredentialsProvider;
 import java.time.Duration;
+import java.util.UUID;
 import java.util.function.BiPredicate;
 import java.util.function.Function;
 import java.util.stream.IntStream;
@@ -126,4 +131,35 @@ void testOffsetBefore() {
   void is_3_11_OrMore(String input, boolean expected) {
     assertThat(Utils.is3_11_OrMore(input)).isEqualTo(expected);
   }
+
+  @Test
+  void maybeSetUpClientParametersFromUrisShouldKeepCredentialsIfNotInUri() {
+    String u = UUID.randomUUID().toString();
+    String p = UUID.randomUUID().toString();
+    UsernamePasswordCredentialsProvider provider =
+        new DefaultUsernamePasswordCredentialsProvider(u, p);
+    ClientParameters cp = new ClientParameters().credentialsProvider(provider);
+    cp =
+        Utils.maybeSetUpClientParametersFromUris(of(create("rabbitmq-stream://host1:5552/vh")), cp);
+
+    provider = (UsernamePasswordCredentialsProvider) cp.credentialsProvider();
+    assertThat(provider.getUsername()).isEqualTo(u);
+    assertThat(provider.getPassword()).isEqualTo(p);
+  }
+
+  @Test
+  void maybeSetUpClientParametersFromUrisShouldOverrideCredentialsIfInUri() {
+    String u = UUID.randomUUID().toString();
+    String p = UUID.randomUUID().toString();
+    UsernamePasswordCredentialsProvider provider =
+        new DefaultUsernamePasswordCredentialsProvider(u, p);
+    ClientParameters cp = new ClientParameters().credentialsProvider(provider);
+    cp =
+        Utils.maybeSetUpClientParametersFromUris(
+            of(create("rabbitmq-stream://foo:bar@host1:5552/vh")), cp);
+
+    provider = (UsernamePasswordCredentialsProvider) cp.credentialsProvider();
+    assertThat(provider.getUsername()).isNotEqualTo(u).isEqualTo("foo");
+    assertThat(provider.getPassword()).isNotEqualTo(p).isEqualTo("bar");
+  }
 }