Added local bind address feature

Author: SgtSilvio

src/main/java/com/hivemq/client/internal/mqtt/MqttClientTransportConfigImpl.java

@@ -34,17 +34,20 @@ public class MqttClientTransportConfigImpl implements MqttClientTransportConfig
 
     public static final @NotNull MqttClientTransportConfigImpl DEFAULT = new MqttClientTransportConfigImpl(
             InetSocketAddress.createUnresolved(MqttClient.DEFAULT_SERVER_HOST, MqttClient.DEFAULT_SERVER_PORT), null,
-            null);
+            null, null);
 
     private final @NotNull InetSocketAddress serverAddress;
+    private final @Nullable InetSocketAddress localAddress;
     private final @Nullable MqttClientSslConfigImpl sslConfig;
     private final @Nullable MqttWebSocketConfigImpl webSocketConfig;
 
     MqttClientTransportConfigImpl(
-            final @NotNull InetSocketAddress serverAddress, final @Nullable MqttClientSslConfigImpl sslConfig,
+            final @NotNull InetSocketAddress serverAddress, final @Nullable InetSocketAddress localAddress,
+            final @Nullable MqttClientSslConfigImpl sslConfig,
             final @Nullable MqttWebSocketConfigImpl webSocketConfig) {
 
         this.serverAddress = serverAddress;
+        this.localAddress = localAddress;
         this.sslConfig = sslConfig;
         this.webSocketConfig = webSocketConfig;
     }
@@ -54,6 +57,15 @@ public class MqttClientTransportConfigImpl implements MqttClientTransportConfig
         return serverAddress;
     }
 
+    @Override
+    public @NotNull Optional<InetSocketAddress> getLocalAddress() {
+        return Optional.ofNullable(localAddress);
+    }
+
+    public @Nullable InetSocketAddress getRawLocalAddress() {
+        return localAddress;
+    }
+
     @Override
     public @NotNull Optional<MqttClientSslConfig> getSslConfig() {
         return Optional.ofNullable(sslConfig);

src/main/java/com/hivemq/client/internal/mqtt/MqttClientTransportConfigImplBuilder.java

@@ -38,6 +38,7 @@ public abstract class MqttClientTransportConfigImplBuilder<B extends MqttClientT
     private @Nullable InetSocketAddress serverAddress;
     private @NotNull Object serverHost = DEFAULT_SERVER_HOST; // String or InetAddress
     private int serverPort = -1;
+    private @Nullable InetSocketAddress localAddress;
     private @Nullable MqttClientSslConfigImpl sslConfig;
     private @Nullable MqttWebSocketConfigImpl webSocketConfig;
 
@@ -51,6 +52,7 @@ public abstract class MqttClientTransportConfigImplBuilder<B extends MqttClientT
         serverAddress = builder.serverAddress;
         serverHost = builder.serverHost;
         serverPort = builder.serverPort;
+        localAddress = builder.localAddress;
         sslConfig = builder.sslConfig;
         webSocketConfig = builder.webSocketConfig;
     }
@@ -78,8 +80,8 @@ void set(final @NotNull MqttClientTransportConfigImpl transportConfig) {
         return self();
     }
 
-    private void setServerHost(final @NotNull Object serverHost) {
-        this.serverHost = serverHost;
+    private void setServerHost(final @NotNull Object host) {
+        serverHost = host;
         if (serverAddress != null) {
             serverPort = serverAddress.getPort();
             serverAddress = null;
@@ -100,6 +102,69 @@ private void setServerHost(final @NotNull Object serverHost) {
         return self();
     }
 
+    public @NotNull B localAddress(final @Nullable InetSocketAddress address) {
+        if (address == null) {
+            localAddress = null;
+        } else {
+            localAddress = checkLocalAddress(address);
+        }
+        return self();
+    }
+
+    public @NotNull B localAddress(final @Nullable String address) {
+        if (address == null) {
+            removeLocalAddress();
+        } else {
+            localAddress = checkLocalAddress(new InetSocketAddress(address, getLocalPort()));
+        }
+        return self();
+    }
+
+    public @NotNull B localAddress(final @Nullable InetAddress address) {
+        if (address == null) {
+            removeLocalAddress();
+        } else {
+            localAddress = new InetSocketAddress(address, getLocalPort());
+        }
+        return self();
+    }
+
+    private @NotNull InetSocketAddress checkLocalAddress(final @NotNull InetSocketAddress address) {
+        if (address.isUnresolved()) {
+            throw new IllegalArgumentException("Local bind address must not be unresolved.");
+        }
+        return address;
+    }
+
+    private void removeLocalAddress() {
+        if ((localAddress != null) && (localAddress.getAddress() != null)) {
+            if (localAddress.getPort() == 0) {
+                localAddress = null;
+            } else {
+                localAddress = new InetSocketAddress(localAddress.getPort());
+            }
+        }
+    }
+
+    private int getLocalPort() {
+        return (localAddress == null) ? 0 : localAddress.getPort();
+    }
+
+    public @NotNull B localPort(final int port) {
+        if (port == 0) {
+            if ((localAddress != null) && (localAddress.getPort() != 0)) {
+                if (localAddress.getAddress() == null) {
+                    localAddress = null;
+                } else {
+                    localAddress = new InetSocketAddress(localAddress.getAddress(), 0);
+                }
+            }
+        } else {
+            localAddress = new InetSocketAddress((localAddress == null) ? null : localAddress.getAddress(), port);
+        }
+        return self();
+    }
+
     public @NotNull B sslWithDefaultConfig() {
         this.sslConfig = MqttClientSslConfigImpl.DEFAULT;
         return self();
@@ -156,7 +221,7 @@ private int getServerPort() {
     }
 
     @NotNull MqttClientTransportConfigImpl buildTransportConfig() {
-        return new MqttClientTransportConfigImpl(getServerAddress(), sslConfig, webSocketConfig);
+        return new MqttClientTransportConfigImpl(getServerAddress(), localAddress, sslConfig, webSocketConfig);
     }
 
     public static class Default extends MqttClientTransportConfigImplBuilder<Default>

src/main/java/com/hivemq/client/internal/mqtt/handler/connect/MqttConnAckSingle.java

@@ -84,18 +84,24 @@ private static void connect(
                     .build()
                     .bootstrap();
 
-            bootstrap.group(eventLoop).connect(flow.getTransportConfig().getServerAddress()).addListener(future -> {
-                final Throwable cause = future.cause();
-                if (cause != null) {
-                    final ConnectionFailedException e = new ConnectionFailedException(cause);
-                    if (eventLoop.inEventLoop()) {
-                        reconnect(clientConfig, MqttDisconnectSource.CLIENT, e, connect, flow, eventLoop);
-                    } else {
-                        eventLoop.execute(() -> reconnect(clientConfig, MqttDisconnectSource.CLIENT, e, connect, flow,
-                                eventLoop));
-                    }
-                }
-            });
+            final MqttClientTransportConfigImpl transportConfig = flow.getTransportConfig();
+
+            bootstrap.group(eventLoop)
+                    .localAddress(transportConfig.getRawLocalAddress())
+                    .connect(transportConfig.getServerAddress())
+                    .addListener(future -> {
+                        final Throwable cause = future.cause();
+                        if (cause != null) {
+                            final ConnectionFailedException e = new ConnectionFailedException(cause);
+                            if (eventLoop.inEventLoop()) {
+                                reconnect(clientConfig, MqttDisconnectSource.CLIENT, e, connect, flow, eventLoop);
+                            } else {
+                                eventLoop.execute(
+                                        () -> reconnect(clientConfig, MqttDisconnectSource.CLIENT, e, connect, flow,
+                                                eventLoop));
+                            }
+                        }
+                    });
         }
     }
 

src/main/java/com/hivemq/client/mqtt/MqttClientTransportConfig.java

@@ -47,6 +47,12 @@ public interface MqttClientTransportConfig {
      */
     @NotNull InetSocketAddress getServerAddress();
 
+    /**
+     * @return the optional local bind address.
+     * @since 1.2
+     */
+    @NotNull Optional<InetSocketAddress> getLocalAddress();
+
     /**
      * @return the optional secure transport configuration.
      */

src/main/java/com/hivemq/client/mqtt/MqttClientTransportConfigBuilderBase.java

@@ -65,6 +65,46 @@ public interface MqttClientTransportConfigBuilderBase<B extends MqttClientTransp
      */
     @NotNull B serverPort(int port);
 
+    /**
+     * Sets the optional {@link MqttClientTransportConfig#getLocalAddress() local bind address}.
+     * <p>
+     * The address must be resolved.
+     *
+     * @param address the local bind address.
+     * @return the builder.
+     * @since 1.2
+     */
+    @NotNull B localAddress(@Nullable InetSocketAddress address);
+
+    /**
+     * Sets the optional local bind address.
+     * <p>
+     * The address must be resolvable.
+     *
+     * @param address the local bind address.
+     * @return the builder.
+     * @since 1.2
+     */
+    @NotNull B localAddress(@Nullable String address);
+
+    /**
+     * Sets the optional local bind address.
+     *
+     * @param address the local bind address
+     * @return the builder.
+     * @since 1.2
+     */
+    @NotNull B localAddress(@Nullable InetAddress address);
+
+    /**
+     * Sets the optional local bind port.
+     *
+     * @param port the local bind port.
+     * @return the builder.
+     * @since 1.2
+     */
+    @NotNull B localPort(int port);
+
     /**
      * Sets the {@link MqttClientTransportConfig#getSslConfig() secure transport configuration} to the default
      * configuration.