From 8bd0dc785290a34df0c9b3a8bf42a76d364561c2 Mon Sep 17 00:00:00 2001
From: andsel <selva.andre@gmail.com>
Date: Sat, 23 Feb 2019 19:51:35 +0100
Subject: [PATCH] Added get*Port() methods on Server to read local bind port
 (when it's passed to '0'), issue #453

---
 .../io/moquette/broker/NewNettyAcceptor.java  | 45 +++++++++++++++++--
 .../main/java/io/moquette/broker/Server.java  |  8 ++++
 2 files changed, 50 insertions(+), 3 deletions(-)

diff --git a/broker/src/main/java/io/moquette/broker/NewNettyAcceptor.java b/broker/src/main/java/io/moquette/broker/NewNettyAcceptor.java
index 353cfea2..c28d96fb 100644
--- a/broker/src/main/java/io/moquette/broker/NewNettyAcceptor.java
+++ b/broker/src/main/java/io/moquette/broker/NewNettyAcceptor.java
@@ -46,7 +46,11 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import javax.net.ssl.SSLEngine;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
 import java.util.concurrent.TimeUnit;
 
@@ -56,6 +60,8 @@ import static io.netty.channel.ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE;
 class NewNettyAcceptor {
 
     private static final String MQTT_SUBPROTOCOL_CSV_LIST = "mqtt, mqttv3.1, mqttv3.1.1";
+    public static final String PLAIN_MQTT_PROTO = "TCP MQTT";
+    public static final String SSL_MQTT_PROTO = "SSL MQTT";
 
     static class WebSocketFrameToByteBufDecoder extends MessageToMessageDecoder<BinaryWebSocketFrame> {
 
@@ -89,10 +95,33 @@ class NewNettyAcceptor {
         abstract void init(SocketChannel channel) throws Exception;
     }
 
+
+    private class LocalPortReaderFutureListener implements ChannelFutureListener {
+        private String transportName;
+
+        LocalPortReaderFutureListener(String transportName) {
+            this.transportName = transportName;
+        }
+
+        @Override
+        public void operationComplete(ChannelFuture future) throws Exception {
+            if (future.isSuccess()) {
+                final SocketAddress localAddress = future.channel().localAddress();
+                if (localAddress instanceof InetSocketAddress) {
+                    InetSocketAddress inetAddress = (InetSocketAddress) localAddress;
+                    LOG.debug("bound {} port: {}", transportName, inetAddress.getPort());
+                    int port = inetAddress.getPort();
+                    ports.put(transportName, port);
+                }
+            }
+        }
+    }
+
     private static final Logger LOG = LoggerFactory.getLogger(NewNettyAcceptor.class);
 
     private EventLoopGroup bossGroup;
     private EventLoopGroup workerGroup;
+    private final Map<String, Integer> ports = new HashMap<>();
     private BytesMetricsCollector bytesMetricsCollector = new BytesMetricsCollector();
     private MessageMetricsCollector metricsCollector = new MessageMetricsCollector();
     private Optional<? extends ChannelInboundHandler> metrics;
@@ -187,12 +216,22 @@ class NewNettyAcceptor {
             // Bind and start to accept incoming connections.
             ChannelFuture f = b.bind(host, port);
             LOG.info("Server bound to host={}, port={}, protocol={}", host, port, protocol);
-            f.sync().addListener(FIRE_EXCEPTION_ON_FAILURE);
+            f.sync()
+                .addListener(new LocalPortReaderFutureListener(protocol))
+                .addListener(FIRE_EXCEPTION_ON_FAILURE);
         } catch (InterruptedException ex) {
             LOG.error("An interruptedException was caught while initializing integration. Protocol={}", protocol, ex);
         }
     }
 
+    public int getPort() {
+        return ports.computeIfAbsent(PLAIN_MQTT_PROTO, i -> 0);
+    }
+
+    public int getSslPort() {
+        return ports.computeIfAbsent(SSL_MQTT_PROTO, i -> 0);
+    }
+
     private void initializePlainTCPTransport(NewNettyMQTTHandler handler, IConfig props) {
         LOG.debug("Configuring TCP MQTT transport");
         final MoquetteIdleTimeoutHandler timeoutHandler = new MoquetteIdleTimeoutHandler();
@@ -204,7 +243,7 @@ class NewNettyAcceptor {
             return;
         }
         int port = Integer.parseInt(tcpPortProp);
-        initFactory(host, port, "TCP MQTT", new PipelineInitializer() {
+        initFactory(host, port, PLAIN_MQTT_PROTO, new PipelineInitializer() {
 
             @Override
             void init(SocketChannel channel) {
@@ -281,7 +320,7 @@ class NewNettyAcceptor {
         String host = props.getProperty(BrokerConstants.HOST_PROPERTY_NAME);
         String sNeedsClientAuth = props.getProperty(BrokerConstants.NEED_CLIENT_AUTH, "false");
         final boolean needsClientAuth = Boolean.valueOf(sNeedsClientAuth);
-        initFactory(host, sslPort, "SSL MQTT", new PipelineInitializer() {
+        initFactory(host, sslPort, SSL_MQTT_PROTO, new PipelineInitializer() {
 
             @Override
             void init(SocketChannel channel) throws Exception {
diff --git a/broker/src/main/java/io/moquette/broker/Server.java b/broker/src/main/java/io/moquette/broker/Server.java
index 166de91f..2d2ecee8 100644
--- a/broker/src/main/java/io/moquette/broker/Server.java
+++ b/broker/src/main/java/io/moquette/broker/Server.java
@@ -340,6 +340,14 @@ public class Server {
         LOG.info("Moquette integration has been stopped.");
     }
 
+    public int getPort() {
+        return acceptor.getPort();
+    }
+
+    public int getSslPort() {
+        return acceptor.getSslPort();
+    }
+
     /**
      * SPI method used by Broker embedded applications to get list of subscribers. Returns null if
      * the broker is not started.