Modbus Coil/DI array fix, support for Word Swap

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

@@ -1985,4 +1985,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,10 @@ 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/optimizer/ModbusOptimizer.java

@@ -51,6 +51,7 @@
 import org.apache.plc4x.java.spi.messages.utils.PlcTagItem;
 import org.apache.plc4x.java.spi.optimizer.SingleTagOptimizer;
 import org.apache.plc4x.java.spi.values.PlcBOOL;
+import org.apache.plc4x.java.spi.values.PlcList;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -208,11 +209,25 @@ protected PlcReadResponse processReadResponses(PlcReadRequest readRequest, Map<P
 
                             // Calculate the byte that contains the response for this Coil
                             byte[] responseData = response.getResponseData();
-                            int bitPosition = modbusTag.getAddress() - response.startingAddress;
-                            int bytePosition = bitPosition / 8;
-                            int bitPositionInByte = bitPosition % 8;
-                            boolean isBitSet = (responseData[bytePosition] & (1 << bitPositionInByte)) != 0;
-                            values.put(tagName, new DefaultPlcResponseItem<>(PlcResponseCode.OK, new PlcBOOL(isBitSet)));
+
+                            // Check if we're dealing with an array of Coils or Discrete Inputs
+                            if (modbusTag.getNumberOfElements() > 1) {
+                                PlcList bitValues = new PlcList();
+                                for (int i = 0; i < modbusTag.getNumberOfElements(); i++) {
+                                    int bitPosition = modbusTag.getAddress() - response.startingAddress + i;
+                                    int bytePosition = bitPosition / 8;
+                                    int bitPositionInByte = bitPosition % 8;
+                                    boolean isBitSet = (responseData[bytePosition] & (1 << bitPositionInByte)) != 0;
+                                    bitValues.add(new PlcBOOL(isBitSet));
+                                }
+                                values.put(tagName, new DefaultPlcResponseItem<>(PlcResponseCode.OK, bitValues));
+                            } else {
+                                int bitPosition = modbusTag.getAddress() - response.startingAddress;
+                                int bytePosition = bitPosition / 8;
+                                int bitPositionInByte = bitPosition % 8;
+                                boolean isBitSet = (responseData[bytePosition] & (1 << bitPositionInByte)) != 0;
+                                values.put(tagName, new DefaultPlcResponseItem<>(PlcResponseCode.OK, new PlcBOOL(isBitSet)));
+                            }
                             break;
                         }
                     }
@@ -436,6 +451,32 @@ private ReadBuffer getReadBuffer(byte[] data, ModbusByteOrder byteOrder) {
                 byte[] reordered = ModbusProtocolLogic.byteSwap(data);
                 return new ReadBufferByteBased(reordered, ByteOrder.LITTLE_ENDIAN);
             }
+            case BIG_ENDIAN_WORD_SWAP: {
+                // [3, 4, 1, 2]
+                // [3, 4, 1, 2, 7, 8, 5, 6]
+                byte[] reordered = ModbusProtocolLogic.wordSwap(data);
+                return new ReadBufferByteBased(reordered, ByteOrder.BIG_ENDIAN);
+            }
+            case LITTLE_ENDIAN_WORD_SWAP: {
+                // [2, 1, 4, 3]
+                // [6, 5, 8, 7, 2, 1, 4, 3]
+                byte[] reordered = ModbusProtocolLogic.wordSwap(data);
+                return new ReadBufferByteBased(reordered, ByteOrder.LITTLE_ENDIAN);
+            }
+            case BIG_ENDIAN_WORD_SWAP_BYTE_SWAP: {
+                // [4, 3, 2, 1]
+                // [4, 3, 2, 1, 8, 7, 6, 5]
+                byte[] reordered = ModbusProtocolLogic.byteSwap(data);
+                reordered = ModbusProtocolLogic.wordSwap(reordered);
+                return new ReadBufferByteBased(reordered, ByteOrder.BIG_ENDIAN);
+            }
+            case LITTLE_ENDIAN_WORD_SWAP_BYTE_SWAP: {
+                // [1, 2, 3, 4]
+                // [5, 6, 7, 8, 1, 2, 3, 4]
+                byte[] reordered = ModbusProtocolLogic.byteSwap(data);
+                reordered = ModbusProtocolLogic.wordSwap(reordered);
+                return new ReadBufferByteBased(reordered, ByteOrder.LITTLE_ENDIAN);
+            }
             default:
                 // 16909060
                 // [1, 2, 3, 4]

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);
@@ -418,4 +423,18 @@ public static byte[] byteSwap(byte[] in) {
         return out;
     }
 
+    public static byte[] wordSwap(byte[] in) {
+        if (in.length % 2 != 0) {
+            throw new PlcRuntimeException("Input byte array length must be a multiple of 2 for word swapping.");
+        }
+        byte[] out = new byte[in.length];
+        for (int i = 0; i < in.length; i += 4) {
+            out[i] = in[i + 2];
+            out[i + 1] = in[i + 3];
+            out[i + 2] = in[i];
+            out[i + 3] = in[i + 1];
+        }
+        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/modbus/src/main/java/org/apache/plc4x/java/modbus/types/ModbusByteOrder.java

@@ -30,5 +30,17 @@ public enum ModbusByteOrder {
     BIG_ENDIAN_BYTE_SWAP,
     // [3, 4, 1, 2]
     // [7, 8, 5, 6, 3, 4, 1, 2]
-    LITTLE_ENDIAN_BYTE_SWAP
+    LITTLE_ENDIAN_BYTE_SWAP,
+    // [3, 4, 1, 2]
+    // [3, 4, 1, 2, 7, 8, 5, 6]
+    BIG_ENDIAN_WORD_SWAP,
+    // [2, 1, 4, 3]
+    // [6, 5, 8, 7, 2, 1, 4, 3]
+    LITTLE_ENDIAN_WORD_SWAP,
+    // [4, 3, 2, 1]
+    // [4, 3, 2, 1, 8, 7, 6, 5]
+    BIG_ENDIAN_WORD_SWAP_BYTE_SWAP,
+    // [1, 2, 3, 4]
+    // [5, 6, 7, 8, 1, 2, 3, 4]
+    LITTLE_ENDIAN_WORD_SWAP_BYTE_SWAP,
 }

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

@@ -900,6 +900,11 @@ private GuidValue toGuidValue(String identifier) {
         return new GuidValue(0L, 0, 0, data4, data5);
     }
 
+    @Override
+    public void channelInactive(ConversationContext<OpcuaAPU> context) {
+        tm.shutdown();
+    }
+
     private static <T> void bridge(RequestTransaction transaction, CompletableFuture<T> future, T response, Throwable error) {
         if (error != null) {
             future.completeExceptionally(error);

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

@@ -2497,4 +2497,9 @@ 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.

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
@@ -200,6 +203,14 @@ public void close(ConversationContext<Message> context) {
             close.reportExit();
         }
 
+        @Override
+        public void channelInactive(ConversationContext<Message> context) {
+            logger.trace("On ChannelInactive");
+            channelInactive.permitEntry();
+            awaitIn(channelInactive);
+            channelInactive.reportExit();
+        }
+
         private void awaitIn(GateKeeper signal) {
             try {
                 if (!signal.awaitIn()) {