add changes to release resources when the channel becomes inactive

plc4j/drivers/ab-eth/src/main/java/org/apache/plc4x/java/abeth/protocol/AbEthProtocolLogic.java

@@ -214,4 +214,9 @@ private PlcResponseCode decodeResponseCode(short status) {
         return PlcResponseCode.NOT_FOUND;
     }
 
+    @Override
+    public void channelInactive(ConversationContext<CIPEncapsulationPacket> context) {
+        tm.shutdown();
+    }
+
 }

plc4j/drivers/ads/src/main/java/org/apache/plc4x/java/ads/protocol/AdsProtocolLogic.java

@@ -1984,4 +1984,9 @@ protected byte[] getNullByteTerminatedArray(String value) {
         return nullTerminatedBytes;
     }
 
+    @Override
+    public void channelInactive(ConversationContext<AmsTCPPacket> context) {
+        tm.shutdown();
+    }
+
 }

plc4j/drivers/bacnet/src/main/java/org/apache/plc4x/java/bacnetip/protocol/BacNetIpProtocolLogic.java

@@ -288,4 +288,9 @@ private String toString(BacNetIpTag tag) {
         return tag.getDeviceIdentifier() + "/" + tag.getObjectType() + "/" + tag.getObjectInstance();
     }
 
+    @Override
+    public void channelInactive(ConversationContext<BVLC> context) {
+        // Nothing to do here ...
+    }
+
 }

plc4j/drivers/c-bus/src/main/java/org/apache/plc4x/java/cbus/protocol/CBusProtocolLogic.java

@@ -78,4 +78,9 @@ public CompletableFuture<PlcReadResponse> read(PlcReadRequest readRequest) {
     protected void decode(ConversationContext<CBusCommand> context, CBusCommand msg) throws Exception {
     }
 
+    @Override
+    public void channelInactive(ConversationContext<CBusCommand> context) {
+        tm.shutdown();
+    }
+
 }

plc4j/drivers/can/src/main/java/org/apache/plc4x/java/can/adapter/CANDriverAdapter.java

@@ -113,4 +113,9 @@ public void close(ConversationContext<C> context) {
         delegate.close(new ConversationContextWrapper<>(context, wireType, adapter, frameHandler, context.getAuthentication()));
     }
 
+    @Override
+    public void channelInactive(ConversationContext<C> context) {
+        delegate.close(new ConversationContextWrapper<>(context, wireType, adapter, frameHandler, context.getAuthentication()));
+    }
+
 }

plc4j/drivers/can/src/main/java/org/apache/plc4x/java/can/generic/protocol/GenericCANProtocolLogic.java

@@ -233,4 +233,9 @@ public PlcConsumerRegistration register(Consumer<PlcSubscriptionEvent> consumer,
     public void unregister(PlcConsumerRegistration registration) {
         consumers.remove(registration);
     }
+
+    @Override
+    public void channelInactive(ConversationContext<GenericFrame> context) {
+        tm.shutdown();
+    }
 }

plc4j/drivers/canopen/src/main/java/org/apache/plc4x/java/canopen/protocol/CANOpenProtocolLogic.java

@@ -453,4 +453,9 @@ public void onDisconnect(ConversationContext<CANOpenFrame> context) {
         }
     }
 
+    @Override
+    public void channelInactive(ConversationContext<CANOpenFrame> context) {
+        tm.shutdown();
+    }
+
 }

plc4j/drivers/eip/src/main/java/org/apache/plc4x/java/eip/base/protocol/EipProtocolLogic.java

@@ -1355,4 +1355,9 @@ private PlcResponseCode decodeResponseCode(int status) {
         }
     }
 
+    @Override
+    public void channelInactive(ConversationContext<EipPacket> context) {
+        tm.shutdown();
+    }
+
 }

plc4j/drivers/firmata/src/main/java/org/apache/plc4x/java/firmata/readwrite/protocol/FirmataProtocolLogic.java

@@ -320,4 +320,9 @@ protected BitSet getDigitalValues(int byteBlock, List<Byte> data) {
         return BitSet.valueOf(bitSetData);
     }
 
+    @Override
+    public void channelInactive(ConversationContext<FirmataMessage> context) {
+        connected.set(false);
+    }
+
 }

plc4j/drivers/iec-60870/src/main/java/org/apache/plc4x/java/iec608705104/readwrite/protocol/Iec608705104Protocol.java

@@ -235,4 +235,9 @@ protected void publishEvent(LocalDateTime timeStamp, Iec608705104Tag tag, PlcVal
         }
     }
 
+    @Override
+    public void channelInactive(ConversationContext<APDU> context) {
+        tm.shutdown();
+    }
+
 }

plc4j/drivers/knxnetip/src/main/java/org/apache/plc4x/java/knxnetip/protocol/KnxNetIpProtocolLogic.java

@@ -640,4 +640,9 @@ protected static String toString(KnxGroupAddress groupAddress) {
         throw new PlcRuntimeException("Unsupported Group Address Type " + groupAddress.getClass().getName());
     }
 
+    @Override
+    public void channelInactive(ConversationContext<KnxNetIpMessage> context) {
+        tm.shutdown();
+    }
+
 }

plc4j/drivers/modbus/src/main/java/org/apache/plc4x/java/modbus/base/protocol/ModbusProtocolLogic.java

@@ -63,6 +63,11 @@ public void close(ConversationContext<T> context) {
         // Nothing to do here ...
     }
 
+    @Override
+    public void channelInactive(ConversationContext<T> context) {
+        // Nothing to do here ...
+    }
+
     @Override
     protected void decode(ConversationContext<T> context, T msg) throws Exception {
         super.decode(context, msg);
@@ -417,5 +422,4 @@ public static byte[] byteSwap(byte[] in) {
         }
         return out;
     }
-
 }

plc4j/drivers/modbus/src/main/java/org/apache/plc4x/java/modbus/tcp/protocol/ModbusTcpProtocolLogic.java

@@ -66,6 +66,11 @@ public void close(ConversationContext<ModbusTcpADU> context) {
         tm.shutdown();
     }
 
+    @Override
+    public void channelInactive(ConversationContext<ModbusTcpADU> context) {
+        tm.shutdown();
+    }
+
     @Override
     public CompletableFuture<PlcPingResponse> ping(PlcPingRequest pingRequest) {
         CompletableFuture<PlcPingResponse> future = new CompletableFuture<>();

plc4j/drivers/opcua/src/main/java/org/apache/plc4x/java/opcua/protocol/OpcuaProtocolLogic.java

@@ -901,4 +901,8 @@ private static <T> void bridge(RequestTransaction transaction, CompletableFuture
         }
     }
 
+    @Override
+    public void channelInactive(ConversationContext<OpcuaAPU> context) {
+        tm.shutdown();
+    }
 }

plc4j/drivers/open-protocol/src/main/java/org/apache/plc4x/java/openprotocol/protocol/OpenProtocolProtocolLogic.java

@@ -80,4 +80,8 @@ public PlcConsumerRegistration register(Consumer<PlcSubscriptionEvent> consumer,
     public void unregister(PlcConsumerRegistration plcConsumerRegistration) {
     }
 
+    @Override
+    public void channelInactive(ConversationContext<OpenProtocolMessage> context) {
+    }
+
 }

plc4j/drivers/plc4x/src/main/java/org/apache/plc4x/java/plc4x/protocol/Plc4xProtocolLogic.java

@@ -200,4 +200,9 @@ protected void decode(ConversationContext<Plc4xMessage> context, Plc4xMessage ms
         super.decode(context, msg);
     }
 
+    @Override
+    public void channelInactive(ConversationContext<Plc4xMessage> context) {
+        tm.shutdown();
+    }
+
 }

plc4j/drivers/profinet-ng/src/main/java/org/apache/plc4x/java/profinet/protocol/ProfinetProtocolLogic.java

@@ -905,4 +905,9 @@ protected int getDataTypeLengthInBytes(PlcValueType dataType) {
         throw new PlcRuntimeException("Length undefined");
     }
 
+    @Override
+    public void channelInactive(ConversationContext<Ethernet_Frame> context) {
+        context.getChannel().close();
+    }
+
 }

plc4j/drivers/profinet/src/main/java/org/apache/plc4x/java/profinet/protocol/ProfinetProtocolLogic.java

@@ -271,4 +271,9 @@ public CompletableFuture<PlcSubscriptionResponse> subscribe(PlcSubscriptionReque
     protected void decode(ConversationContext<Ethernet_Frame> context, Ethernet_Frame msg) throws Exception {
         super.decode(context, msg);
     }
+
+    @Override
+    public void channelInactive(ConversationContext<Ethernet_Frame> context) {
+        // TODO:- Do something here
+    }
 }

plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/readwrite/protocol/S7ProtocolLogic.java

@@ -2494,4 +2494,8 @@ public int getCurLength() {
 
     }
 
-}
+    @Override
+    public void channelInactive(ConversationContext<TPKTPacket> context) {
+        tm.shutdown();
+    }
+}

plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7light/readwrite/protocol/S7ProtocolLogic.java

@@ -88,6 +88,11 @@ public void close(ConversationContext<TPKTPacket> context) {
         tm.shutdown();
     }
 
+    @Override
+    public void channelInactive(ConversationContext<TPKTPacket> context) {
+        tm.shutdown();
+    }
+
     @Override
     public void onConnect(ConversationContext<TPKTPacket> context) {
         // Only the TCP transport supports login.
@@ -1099,5 +1104,4 @@ public int getCurLength() {
         }
 
     }
-
-}
+}

plc4j/spi/src/main/java/org/apache/plc4x/java/spi/Plc4xNettyWrapper.java

@@ -136,10 +136,19 @@ public ExpectRequestContext<T> expectRequest(Class<T> clazz, Duration timeout) {
 
     @Override
     public void close(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
+        logger.trace("close.. context: {}", ctx.name());
         super.close(ctx, promise);
         timeoutManager.stop();
     }
 
+    @Override
+    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+        logger.trace("channelInactive.. context: {}", ctx.name());
+        super.channelInactive(ctx);
+        this.protocolBase.channelInactive(new DefaultConversationContext<>(this::registerHandler, ctx, authentication, passive));
+        timeoutManager.stop();
+    }
+
     @Override
     protected void encode(ChannelHandlerContext channelHandlerContext, Object msg, List<Object> list) throws Exception {
         logger.debug("Forwarding request to plc {}", msg);

plc4j/spi/src/main/java/org/apache/plc4x/java/spi/Plc4xProtocolBase.java

@@ -101,4 +101,6 @@ public CompletableFuture<PlcBrowseResponse> browseWithInterceptor(PlcBrowseReque
 
     public abstract void close(ConversationContext<T> context);
 
+    public abstract void channelInactive(ConversationContext<T> context);
+
 }

plc4j/spi/src/test/java/org/apache/plc4x/java/spi/connection/DefaultNettyPlcConnectionTest.java

@@ -41,13 +41,14 @@ class DefaultNettyPlcConnectionTest {
     final GateKeeper connect = new GateKeeper("connect");
     final GateKeeper disconnect = new GateKeeper("disconnect");
     final GateKeeper close = new GateKeeper("close");
+    final GateKeeper channelInactive = new GateKeeper("channelInactive");
 
     @Test
     void checkInitializationSequence() throws Exception {
         ChannelFactory channelFactory = new TestChannelFactory();
 
         ProtocolStackConfigurer<Message> stackConfigurer = (configuration, pipeline, authentication, passive, listeners) -> {
-            TestProtocolBase base = new TestProtocolBase(discovery, connect, disconnect, close);
+            TestProtocolBase base = new TestProtocolBase(discovery, connect, disconnect, close, channelInactive);
             Plc4xNettyWrapper<Message> context = new Plc4xNettyWrapper<>(new NettyHashTimerTimeoutManager(), pipeline, passive, base, authentication, Message.class);
             pipeline.addLast(context);
             return base;
@@ -153,12 +154,14 @@ static class TestProtocolBase extends Plc4xProtocolBase<Message> {
         private final GateKeeper connect;
         private final GateKeeper close;
         private final GateKeeper disconnect;
+        private final GateKeeper channelInactive;
 
-        public TestProtocolBase(GateKeeper discover, GateKeeper connect, GateKeeper disconnect, GateKeeper close) {
+        public TestProtocolBase(GateKeeper discover, GateKeeper connect, GateKeeper disconnect, GateKeeper close, GateKeeper channelInactive) {
             this.discover = discover;
             this.connect = connect;
             this.close = close;
             this.disconnect = disconnect;
+            this.channelInactive =channelInactive;
         }
 
         @Override
@@ -210,6 +213,13 @@ private void awaitIn(GateKeeper signal) {
                 throw new RuntimeException(e);
             }
         }
-    }
 
+        @Override
+        public void channelInactive(ConversationContext<Message> context) {
+            logger.info("On ChannelInactive");
+            channelInactive.permitEntry();
+            awaitIn(channelInactive);
+            channelInactive.reportExit();
+        }
+    }
 }

plc4j/spi/src/test/java/org/apache/plc4x/java/spi/connection/TestProtocol.java

@@ -57,4 +57,9 @@ public void close(ConversationContext<TestMessage> context) {
         context.getChannel().close();
     }
 
+    @Override
+    public void channelInactive(ConversationContext<TestMessage> context) {
+        context.getChannel().close();
+    }
+
 }