~ ConnectionListener for NettyTcpClientTransport

Author: kevinherron

modbus-tcp/src/main/java/com/digitalpetri/modbus/client/NettyTcpClientTransport.java

@@ -19,8 +19,10 @@
 import io.netty.channel.SimpleChannelInboundHandler;
 import io.netty.channel.socket.SocketChannel;
 import io.netty.channel.socket.nio.NioSocketChannel;
+import java.util.List;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
+import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
 import org.slf4j.Logger;
@@ -36,6 +38,8 @@ public class NettyTcpClientTransport implements ModbusTcpClientTransport {
 
   private final AtomicReference<Consumer<ModbusTcpFrame>> frameReceiver = new AtomicReference<>();
 
+  private final List<ConnectionListener> connectionListeners = new CopyOnWriteArrayList<>();
+
   private final ChannelFsm channelFsm;
   private final ExecutionQueue executionQueue;
 
@@ -54,12 +58,23 @@ public NettyTcpClientTransport(NettyClientTransportConfig config) {
             .build()
     );
 
+    executionQueue = new ExecutionQueue(config.executor());
+
     channelFsm.addTransitionListener(
-        (from, to, via) ->
-            logger.debug("onStateTransition: {} -> {} via {}", from, to, via)
+        (from, to, via) -> {
+          logger.debug("onStateTransition: {} -> {} via {}", from, to, via);
+
+          executionQueue.submit(() -> handleStateTransition(from, to, via));
+        }
     );
+  }
 
-    executionQueue = new ExecutionQueue(config.executor());
+  private void handleStateTransition(State from, State to, Event via) {
+    if (from != State.Connected && to == State.Connected) {
+      connectionListeners.forEach(ConnectionListener::onConnection);
+    } else if (from == State.Connected && to != State.Connected) {
+      connectionListeners.forEach(ConnectionListener::onConnectionLost);
+    }
   }
 
   @Override
@@ -99,6 +114,24 @@ public boolean isConnected() {
     return channelFsm.getState() == State.Connected;
   }
 
+  /**
+   * Add a {@link ConnectionListener} to this transport.
+   *
+   * @param listener the listener to add.
+   */
+  public void addConnectionListener(ConnectionListener listener) {
+    connectionListeners.add(listener);
+  }
+
+  /**
+   * Remove a {@link ConnectionListener} from this transport.
+   *
+   * @param listener the listener to remove.
+   */
+  public void removeConnectionListener(ConnectionListener listener) {
+    connectionListeners.remove(listener);
+  }
+
   private class ModbusTcpFrameHandler extends SimpleChannelInboundHandler<ModbusTcpFrame> {
 
     @Override
@@ -179,4 +212,18 @@ public static NettyTcpClientTransport create(
     return new NettyTcpClientTransport(config);
   }
 
+  public interface ConnectionListener {
+
+    /**
+     * Callback invoked when the transport has connected.
+     */
+    void onConnection();
+
+    /**
+     * Callback invoked when the transport has disconnected.
+     */
+    void onConnectionLost();
+
+  }
+
 }