Merge pull request #65 from netifi-proteus/feature/useCreateUnresolved

Feature/use create unresolved

Author: flyinprogrammer

proteus-client/src/main/java/io/netifi/proteus/DefaultProteusBrokerService.java

@@ -145,7 +145,7 @@ public DefaultProteusBrokerService(
     this.addressSelector = addressSelector;
     this.clientTransportFactory = clientTransportFactory;
     this.poolSize = poolSize;
-    this.selectRefresh = poolSize / 2;
+    this.selectRefresh = poolSize == 1 ? 1 : poolSize / 2;
     this.keepalive = keepalive;
     this.tickPeriodSeconds = tickPeriodSeconds;
     this.ackTimeoutSeconds = ackTimeoutSeconds;
@@ -182,7 +182,10 @@ private void useDiscoveryStrategy() {
         discoveryStrategy
             .discoverNodes()
             .flatMapIterable(Function.identity())
-            .map(hostAndPort -> new InetSocketAddress(hostAndPort.getHost(), hostAndPort.getPort()))
+            .map(
+                hostAndPort ->
+                    InetSocketAddress.createUnresolved(
+                        hostAndPort.getHost(), hostAndPort.getPort()))
             .collectList()
             .doOnNext(
                 i -> {

proteus-client/src/main/java/io/netifi/proteus/Proteus.java

@@ -707,7 +707,7 @@ public static class Builder {
     private int missedAcks = DefaultBuilderConfig.getMissedAcks();
     private DiscoveryStrategy discoveryStrategy = null;
     private Function<Broker, InetSocketAddress> addressSelector =
-        BrokerAddressSelectors.BIND_ADDRESS; // Default
+        BrokerAddressSelectors.TCP_ADDRESS; // Default
 
     private Function<SocketAddress, ClientTransport> clientTransportFactory = null;
     private int poolSize = Runtime.getRuntime().availableProcessors() * 2;

proteus-client/src/main/java/io/netifi/proteus/rsocket/transport/BrokerAddressSelectors.java

@@ -20,9 +20,6 @@
 import java.util.function.Function;
 
 public class BrokerAddressSelectors {
-
-  public static Function<Broker, InetSocketAddress> BIND_ADDRESS =
-      broker -> InetSocketAddress.createUnresolved(broker.getIpAddress(), broker.getPort());
   public static Function<Broker, InetSocketAddress> TCP_ADDRESS =
       broker -> InetSocketAddress.createUnresolved(broker.getTcpAddress(), broker.getTcpPort());
   public static Function<Broker, InetSocketAddress> WEBSOCKET_ADDRESS =

proteus-client/src/test/java/io/netifi/proteus/rsocket/transport/WeightedClientTransportSupplierTest.java

@@ -38,7 +38,7 @@ public void testShouldDecrementActiveCountOnComplete() {
     Broker b = Broker.newBuilder().setTcpAddress("localhost").setTcpPort(8001).build();
     WeightedClientTransportSupplier supplier =
         new WeightedClientTransportSupplier(
-            b, BrokerAddressSelectors.BIND_ADDRESS, address -> transport);
+            b, BrokerAddressSelectors.TCP_ADDRESS, address -> transport);
 
     supplier.select();
     DuplexConnection block = supplier.get().connect().block();