[incubator-plc4x] branch master updated: implement proper serial communication using locks for now

[sr...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

sruehl pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-plc4x.git


The following commit(s) were added to refs/heads/master by this push:
     new 3a77612  implement proper serial communication using locks for now
3a77612 is described below

commit 3a77612411709ccc5c61f9e3b0a754e293709f3f
Author: Sebastian Rühl <sr...@apache.org>
AuthorDate: Mon May 14 14:53:26 2018 +0200

    implement proper serial communication using locks for now
---
 .../java/ads/protocol/Payload2SerialProtocol.java  | 52 +++++++++++++++++++---
 1 file changed, 47 insertions(+), 5 deletions(-)

diff --git a/plc4j/protocols/ads/src/main/java/org/apache/plc4x/java/ads/protocol/Payload2SerialProtocol.java b/plc4j/protocols/ads/src/main/java/org/apache/plc4x/java/ads/protocol/Payload2SerialProtocol.java
index 4ef0fab..2e4cc37 100644
--- a/plc4j/protocols/ads/src/main/java/org/apache/plc4x/java/ads/protocol/Payload2SerialProtocol.java
+++ b/plc4j/protocols/ads/src/main/java/org/apache/plc4x/java/ads/protocol/Payload2SerialProtocol.java
@@ -23,6 +23,8 @@ import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelFutureListener;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.handler.codec.MessageToMessageCodec;
+import io.netty.util.ReferenceCountUtil;
+import io.netty.util.concurrent.ScheduledFuture;
 import org.apache.plc4x.java.ads.api.serial.AmsSerialAcknowledgeFrame;
 import org.apache.plc4x.java.ads.api.serial.AmsSerialFrame;
 import org.apache.plc4x.java.ads.api.serial.AmsSerialResetFrame;
@@ -33,16 +35,52 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
 
 public class Payload2SerialProtocol extends MessageToMessageCodec<ByteBuf, ByteBuf> {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(Payload2TcpProtocol.class);
 
+    private final AtomicInteger fragmentCounter = new AtomicInteger(0);
+
+    private final AtomicBoolean frameOnTheWay = new AtomicBoolean(false);
+
+    private volatile ScheduledFuture<ChannelFuture> retryHandler;
+
+    private final Lock lock = new ReentrantLock(true);
+
     @Override
     protected void encode(ChannelHandlerContext channelHandlerContext, ByteBuf amsPacket, List<Object> out) throws Exception {
-        out.add(AmsSerialFrame.of(FragmentNumber.of((byte) 0), UserData.of(amsPacket)).getByteBuf());
-        // TODO: we need to remember the fragment and maybe even need to spilt up the package
-        // TODO: if we exceed 255 byte
+        while (frameOnTheWay.get() || !lock.tryLock()) {
+            // In this case we might not send it yet.
+            TimeUnit.MILLISECONDS.sleep(10);
+        }
+        int fragmentNumber = fragmentCounter.getAndIncrement();
+        if (fragmentNumber > 255) {
+            fragmentNumber = 0;
+            fragmentCounter.set(fragmentNumber);
+        }
+        try {
+            // TODO: we need to remember the fragment and maybe even need to spilt up the package
+            // TODO: if we exceed 255 byte
+            AmsSerialFrame amsSerialFrame = AmsSerialFrame.of(FragmentNumber.of((byte) fragmentNumber), UserData.of(amsPacket));
+            out.add(amsSerialFrame.getByteBuf());
+            retryHandler = channelHandlerContext.executor().schedule(() -> {
+                try {
+                    TimeUnit.SECONDS.sleep(2);
+                    channelHandlerContext.channel().writeAndFlush(amsSerialFrame.getByteBuf());
+                } catch (InterruptedException ignore) {
+                }
+                return channelHandlerContext.voidPromise();
+            }, 0, TimeUnit.MILLISECONDS);
+            frameOnTheWay.set(true);
+        } finally {
+            lock.unlock();
+        }
     }
 
     @Override
@@ -70,19 +108,23 @@ public class Payload2SerialProtocol extends MessageToMessageCodec<ByteBuf, ByteB
                 // TODO: check if this is the right way to ack a package.
                 ChannelFuture channelFuture = channelHandlerContext.writeAndFlush(AmsSerialAcknowledgeFrame.of(transmitterAddress, receiverAddress, fragmentNumber));
                 channelFuture.addListener((ChannelFutureListener) future -> {
-                    // TODO: we might wait for the ack-frame to be transmitted before we forward the package
+                    frameOnTheWay.set(false);
                 });
-                // TODO: check if this await is right as this might be against the concept of netty
+                // waiting for the ack-frame to be transmitted before we forward the package
                 channelFuture.await();
                 out.add(userData.getByteBuf());
                 break;
             case AmsSerialAcknowledgeFrame.ID:
                 AmsSerialAcknowledgeFrame amsSerialAcknowledgeFrame = AmsSerialAcknowledgeFrame.of(magicCookie, transmitterAddress, receiverAddress, fragmentNumber, userDataLength, crc);
                 LOGGER.debug("Ams Serial ACK Frame received {}", amsSerialAcknowledgeFrame);
+                retryHandler.cancel(true);
+                ReferenceCountUtil.release(byteBuf);
                 break;
             case AmsSerialResetFrame.ID:
+                // TODO: how to react to a reset
                 AmsSerialResetFrame amsSerialResetFrame = AmsSerialResetFrame.of(magicCookie, transmitterAddress, receiverAddress, fragmentNumber, userDataLength, crc);
                 LOGGER.debug("Ams Serial Reset Frame received {}", amsSerialResetFrame);
+                ReferenceCountUtil.release(byteBuf);
                 break;
         }
         CRC calculatedCrc = CRC.of(DigestUtil.calculateCrc16(magicCookie, transmitterAddress, receiverAddress, fragmentNumber, userDataLength, userData));

-- 
To stop receiving notification emails like this one, please contact
sruehl@apache.org.