You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@plc4x.apache.org by sr...@apache.org on 2018/06/13 11:23:09 UTC
[incubator-plc4x] branch master updated (aaad9ce -> e69ef91)
This is an automated email from the ASF dual-hosted git repository.
sruehl pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-plc4x.git.
from aaad9ce smaller cleanups
new 83c69c9 added single message rate limiter to throttle outgoing messages to one on serial connection
new c14ecc1 added a ADSException to correlate ads messages using the invokeId
new e69ef91 implement retry mechanism on ADS/AMS using the fixed rate scheduler.
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../java/ads/api/serial/AmsSerialResetFrame.java | 13 +++
.../ads/connection/AdsSerialPlcConnection.java | 2 +
.../java/ads/protocol/Payload2SerialProtocol.java | 34 +++++-
.../plc4x/java/ads/protocol/Plc4x2AdsProtocol.java | 16 ++-
.../java/ads/protocol/exception/AdsException.java | 57 ++++++++++
.../protocol/util/SingleMessageRateLimiter.java | 120 +++++++++++++++++++++
.../ads/connection/AdsSerialPlcConnectionTest.java | 16 +++
7 files changed, 252 insertions(+), 6 deletions(-)
create mode 100644 plc4j/protocols/ads/src/main/java/org/apache/plc4x/java/ads/protocol/exception/AdsException.java
create mode 100644 plc4j/protocols/ads/src/main/java/org/apache/plc4x/java/ads/protocol/util/SingleMessageRateLimiter.java
--
To stop receiving notification emails like this one, please contact
sruehl@apache.org.
[incubator-plc4x] 01/03: added single message rate limiter to
throttle outgoing messages to one on serial connection
Posted by 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
commit 83c69c9a9a2becb2c14568c937b3de3d010ed630
Author: Sebastian Rühl <sr...@apache.org>
AuthorDate: Wed Jun 13 12:44:15 2018 +0200
added single message rate limiter to throttle outgoing messages to one
on serial connection
---
.../ads/connection/AdsSerialPlcConnection.java | 2 +
.../protocol/util/SingleMessageRateLimiter.java | 120 +++++++++++++++++++++
.../ads/connection/AdsSerialPlcConnectionTest.java | 16 +++
3 files changed, 138 insertions(+)
diff --git a/plc4j/protocols/ads/src/main/java/org/apache/plc4x/java/ads/connection/AdsSerialPlcConnection.java b/plc4j/protocols/ads/src/main/java/org/apache/plc4x/java/ads/connection/AdsSerialPlcConnection.java
index 848a743..82ca6a2 100644
--- a/plc4j/protocols/ads/src/main/java/org/apache/plc4x/java/ads/connection/AdsSerialPlcConnection.java
+++ b/plc4j/protocols/ads/src/main/java/org/apache/plc4x/java/ads/connection/AdsSerialPlcConnection.java
@@ -27,6 +27,7 @@ import org.apache.plc4x.java.ads.api.generic.types.AmsPort;
import org.apache.plc4x.java.ads.protocol.Ads2PayloadProtocol;
import org.apache.plc4x.java.ads.protocol.Payload2SerialProtocol;
import org.apache.plc4x.java.ads.protocol.Plc4x2AdsProtocol;
+import org.apache.plc4x.java.ads.protocol.util.SingleMessageRateLimiter;
import org.apache.plc4x.java.base.connection.SerialChannelFactory;
import java.util.concurrent.CompletableFuture;
@@ -57,6 +58,7 @@ public class AdsSerialPlcConnection extends AdsAbstractPlcConnection {
// Build the protocol stack for communicating with the ads protocol.
ChannelPipeline pipeline = channel.pipeline();
pipeline.addLast(new Payload2SerialProtocol());
+ pipeline.addLast(new SingleMessageRateLimiter());
pipeline.addLast(new Ads2PayloadProtocol());
pipeline.addLast(new Plc4x2AdsProtocol(targetAmsNetId, targetAmsPort, sourceAmsNetId, sourceAmsPort, addressMapping));
}
diff --git a/plc4j/protocols/ads/src/main/java/org/apache/plc4x/java/ads/protocol/util/SingleMessageRateLimiter.java b/plc4j/protocols/ads/src/main/java/org/apache/plc4x/java/ads/protocol/util/SingleMessageRateLimiter.java
new file mode 100644
index 0000000..afab15d
--- /dev/null
+++ b/plc4j/protocols/ads/src/main/java/org/apache/plc4x/java/ads/protocol/util/SingleMessageRateLimiter.java
@@ -0,0 +1,120 @@
+/*
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+ */
+package org.apache.plc4x.java.ads.protocol.util;
+
+import io.netty.channel.ChannelDuplexHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelPromise;
+import io.netty.util.concurrent.ScheduledFuture;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.net.SocketAddress;
+import java.util.ArrayDeque;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * Inspired by {@link ChannelTrafficShapingHandler} this limiter ensures only one message is sent at a time.
+ */
+public class SingleMessageRateLimiter extends ChannelDuplexHandler {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(SingleMessageRateLimiter.class);
+
+ private final ArrayDeque<ToSend> messagesQueue = new ArrayDeque<>();
+
+ private AtomicBoolean messageOnTheWay = new AtomicBoolean(false);
+
+ private ScheduledFuture<?> sender;
+
+ @Override
+ public void bind(ChannelHandlerContext ctx, SocketAddress localAddress, ChannelPromise promise) throws Exception {
+ LOGGER.debug("bind({}, {}, {})", ctx, localAddress, promise);
+ super.bind(ctx, localAddress, promise);
+ }
+
+ @Override
+ public void deregister(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
+ LOGGER.debug("deregister({}, {})", ctx, promise);
+ super.deregister(ctx, promise);
+ }
+
+ @Override
+ public void connect(ChannelHandlerContext ctx, SocketAddress remoteAddress, SocketAddress localAddress, ChannelPromise promise) throws Exception {
+ LOGGER.debug("connect({}, {}, {}, {})", ctx, remoteAddress, localAddress, promise);
+ sender = ctx.executor().scheduleAtFixedRate(() -> {
+ LOGGER.trace("Woke up and doing work messageOnTheWay:{}, messageQueue:{}", messageOnTheWay, messagesQueue);
+ if (!messagesQueue.isEmpty() && messageOnTheWay.compareAndSet(false, true)) {
+ ToSend pop = messagesQueue.pop();
+ LOGGER.debug("Sending {}", pop);
+ pop.channelHandlerContext.writeAndFlush(pop.toSend, pop.promise);
+ LOGGER.debug("Send {}", pop);
+ }
+ }, 0, 10, TimeUnit.MILLISECONDS);
+ super.connect(ctx, remoteAddress, localAddress, promise);
+ }
+
+ @Override
+ public void disconnect(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
+ LOGGER.debug("disconnect({}, {}, {}, {})", ctx, promise);
+ sender.cancel(true);
+ super.disconnect(ctx, promise);
+ }
+
+ @Override
+ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
+ LOGGER.trace("(<--OUT): {}, {}, {}", ctx, msg, promise);
+ messagesQueue.add(new ToSend(ctx, msg, promise));
+ }
+
+ @Override
+ public void read(ChannelHandlerContext ctx) throws Exception {
+ LOGGER.trace("(-->In): {}", ctx);
+ messageOnTheWay.set(false);
+ super.read(ctx);
+ }
+
+ @Override
+ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
+ LOGGER.trace("(-->ERR): {}", ctx, cause);
+ messageOnTheWay.set(false);
+ super.exceptionCaught(ctx, cause);
+ }
+
+ private static final class ToSend {
+ final ChannelHandlerContext channelHandlerContext;
+ final Object toSend;
+ final ChannelPromise promise;
+
+ private ToSend(ChannelHandlerContext channelHandlerContext, Object toSend, ChannelPromise promise) {
+ this.channelHandlerContext = channelHandlerContext;
+ this.toSend = toSend;
+ this.promise = promise;
+ }
+
+ @Override
+ public String toString() {
+ return "ToSend{" +
+ "channelHandlerContext=" + channelHandlerContext +
+ ", toSend=" + toSend +
+ ", promise=" + promise +
+ '}';
+ }
+ }
+}
diff --git a/plc4j/protocols/ads/src/test/java/org/apache/plc4x/java/ads/connection/AdsSerialPlcConnectionTest.java b/plc4j/protocols/ads/src/test/java/org/apache/plc4x/java/ads/connection/AdsSerialPlcConnectionTest.java
index 5f6b16a..703c205 100644
--- a/plc4j/protocols/ads/src/test/java/org/apache/plc4x/java/ads/connection/AdsSerialPlcConnectionTest.java
+++ b/plc4j/protocols/ads/src/test/java/org/apache/plc4x/java/ads/connection/AdsSerialPlcConnectionTest.java
@@ -20,7 +20,9 @@ package org.apache.plc4x.java.ads.connection;
import io.netty.buffer.ByteBuf;
import io.netty.channel.embedded.EmbeddedChannel;
+import io.netty.channel.jsc.JSerialCommDeviceAddress;
import org.apache.commons.lang3.reflect.FieldUtils;
+import org.apache.commons.lang3.reflect.MethodUtils;
import org.apache.plc4x.java.ads.api.generic.types.AmsNetId;
import org.apache.plc4x.java.ads.api.generic.types.AmsPort;
import org.apache.plc4x.java.ads.api.serial.AmsSerialAcknowledgeFrame;
@@ -39,6 +41,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.lang.reflect.Field;
+import java.lang.reflect.InvocationTargetException;
import java.util.Arrays;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
@@ -113,6 +116,7 @@ public class AdsSerialPlcConnectionTest {
SerialChannelFactory serialChannelFactory = (SerialChannelFactory) channelFactoryField.get(SUT);
SerialChannelFactory serialChannelFactorySpied = spy(serialChannelFactory);
EmbeddedChannel embeddedChannel = new EmbeddedChannel(SUT.getChannelHandler(null));
+ embeddedChannel.connect(new JSerialCommDeviceAddress("/dev/tty0"));
doReturn(embeddedChannel).when(serialChannelFactorySpied).createChannel(any());
channelFactoryField.set(SUT, serialChannelFactorySpied);
SUT.connect();
@@ -135,12 +139,14 @@ public class AdsSerialPlcConnectionTest {
@Override
public void run() {
while (true) {
+ LOGGER.trace("in state {}. CurrentInvokeId: {}", state, currentInvokeId);
switch (state) {
// Receiving state
case RECEIVE_REQUEST: {
LOGGER.info("Waiting for normal message");
ByteBuf outputBuffer;
while ((outputBuffer = embeddedChannel.readOutbound()) == null) {
+ LOGGER.trace("No buffer available yet");
if (!trySleep()) {
return;
}
@@ -173,6 +179,11 @@ public class AdsSerialPlcConnectionTest {
ReceiverAddress.of((byte) 0x0),
FragmentNumber.of((byte) 0)
).getByteBuf();
+ try {
+ MethodUtils.invokeMethod(byteBuf, true,"setRefCnt", 2);
+ } catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException e) {
+ throw new RuntimeException(e);
+ }
embeddedChannel.writeOneInbound(byteBuf);
LOGGER.info("Acked Message");
state = SimulatorState.SEND_RESPONSE;
@@ -198,6 +209,11 @@ public class AdsSerialPlcConnectionTest {
}
)
).getByteBuf();
+ try {
+ MethodUtils.invokeMethod(byteBuf, true,"setRefCnt", 2);
+ } catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException e) {
+ throw new RuntimeException(e);
+ }
embeddedChannel.writeOneInbound(byteBuf);
LOGGER.info("Wrote Inbound");
state = SimulatorState.WAIT_FOR_ACK;
--
To stop receiving notification emails like this one, please contact
sruehl@apache.org.
[incubator-plc4x] 02/03: added a ADSException to correlate ads
messages using the invokeId
Posted by 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
commit c14ecc167b0be565454cfd21a03763c1ad960ec6
Author: Sebastian Rühl <sr...@apache.org>
AuthorDate: Wed Jun 13 13:22:00 2018 +0200
added a ADSException to correlate ads messages using the invokeId
---
.../plc4x/java/ads/protocol/Plc4x2AdsProtocol.java | 16 +++++-
.../java/ads/protocol/exception/AdsException.java | 57 ++++++++++++++++++++++
2 files changed, 72 insertions(+), 1 deletion(-)
diff --git a/plc4j/protocols/ads/src/main/java/org/apache/plc4x/java/ads/protocol/Plc4x2AdsProtocol.java b/plc4j/protocols/ads/src/main/java/org/apache/plc4x/java/ads/protocol/Plc4x2AdsProtocol.java
index 7e2c19e..d1485bc 100644
--- a/plc4j/protocols/ads/src/main/java/org/apache/plc4x/java/ads/protocol/Plc4x2AdsProtocol.java
+++ b/plc4j/protocols/ads/src/main/java/org/apache/plc4x/java/ads/protocol/Plc4x2AdsProtocol.java
@@ -28,6 +28,7 @@ import org.apache.plc4x.java.ads.api.generic.types.AmsPort;
import org.apache.plc4x.java.ads.api.generic.types.Invoke;
import org.apache.plc4x.java.ads.model.AdsAddress;
import org.apache.plc4x.java.ads.model.SymbolicAdsAddress;
+import org.apache.plc4x.java.ads.protocol.exception.AdsException;
import org.apache.plc4x.java.api.exceptions.PlcException;
import org.apache.plc4x.java.api.exceptions.PlcIoException;
import org.apache.plc4x.java.api.exceptions.PlcProtocolException;
@@ -101,7 +102,20 @@ public class Plc4x2AdsProtocol extends MessageToMessageCodec<AmsPacket, PlcReque
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
- if ((cause instanceof IOException) && (cause.getMessage().contains("Connection reset by peer") ||
+ LOGGER.trace("(-->ERR): {}", ctx, cause);
+ if (cause instanceof AdsException) {
+ Invoke invokeId = ((AdsException) cause).getInvokeId();
+ if (invokeId != null) {
+ PlcRequestContainer<PlcRequest, PlcResponse> remove = requests.remove(invokeId.getAsLong());
+ if (remove != null) {
+ remove.getResponseFuture().completeExceptionally(new PlcIoException(cause));
+ } else {
+ LOGGER.warn("Unrelated exception received {}", invokeId, cause);
+ }
+ } else {
+ super.exceptionCaught(ctx, cause);
+ }
+ } else if ((cause instanceof IOException) && (cause.getMessage().contains("Connection reset by peer") ||
cause.getMessage().contains("Operation timed out"))) {
String reason = cause.getMessage().contains("Connection reset by peer") ?
"Connection terminated unexpectedly" : "Remote host not responding";
diff --git a/plc4j/protocols/ads/src/main/java/org/apache/plc4x/java/ads/protocol/exception/AdsException.java b/plc4j/protocols/ads/src/main/java/org/apache/plc4x/java/ads/protocol/exception/AdsException.java
new file mode 100644
index 0000000..4d21693
--- /dev/null
+++ b/plc4j/protocols/ads/src/main/java/org/apache/plc4x/java/ads/protocol/exception/AdsException.java
@@ -0,0 +1,57 @@
+/*
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+ */
+package org.apache.plc4x.java.ads.protocol.exception;
+
+import org.apache.plc4x.java.ads.api.generic.types.Invoke;
+import org.apache.plc4x.java.api.exceptions.PlcProtocolException;
+
+public class AdsException extends PlcProtocolException {
+ private final Invoke invokeId;
+
+ public AdsException(Invoke invokeId, String message) {
+ super(message);
+ this.invokeId = invokeId;
+ }
+
+ public AdsException(Invoke invokeId, String message, Throwable cause) {
+ super(message, cause);
+ this.invokeId = invokeId;
+ }
+
+ public AdsException(Invoke invokeId, Throwable cause) {
+ super(cause);
+ this.invokeId = invokeId;
+ }
+
+ public AdsException(Invoke invokeId, String message, Throwable cause, boolean enableSuppression, boolean writableStackTrace) {
+ super(message, cause, enableSuppression, writableStackTrace);
+ this.invokeId = invokeId;
+ }
+
+ public Invoke getInvokeId() {
+ return invokeId;
+ }
+
+ @Override
+ public String toString() {
+ return "AdsException{" +
+ "invokeId=" + invokeId +
+ "} " + super.toString();
+ }
+}
--
To stop receiving notification emails like this one, please contact
sruehl@apache.org.
[incubator-plc4x] 03/03: implement retry mechanism on ADS/AMS using
the fixed rate scheduler.
Posted by 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
commit e69ef91abcea75d1648a15a122c46ddb6c107e02
Author: Sebastian Rühl <sr...@apache.org>
AuthorDate: Wed Jun 13 13:22:58 2018 +0200
implement retry mechanism on ADS/AMS using the fixed rate scheduler.
---
.../java/ads/api/serial/AmsSerialResetFrame.java | 13 +++++++++
.../java/ads/protocol/Payload2SerialProtocol.java | 34 ++++++++++++++++++----
2 files changed, 42 insertions(+), 5 deletions(-)
diff --git a/plc4j/protocols/ads/src/main/java/org/apache/plc4x/java/ads/api/serial/AmsSerialResetFrame.java b/plc4j/protocols/ads/src/main/java/org/apache/plc4x/java/ads/api/serial/AmsSerialResetFrame.java
index e815d68..4c8ab65 100644
--- a/plc4j/protocols/ads/src/main/java/org/apache/plc4x/java/ads/api/serial/AmsSerialResetFrame.java
+++ b/plc4j/protocols/ads/src/main/java/org/apache/plc4x/java/ads/api/serial/AmsSerialResetFrame.java
@@ -81,6 +81,15 @@ public class AmsSerialResetFrame implements ByteReadable {
this.crc = CRC.of(DigestUtil.calculateCrc16(magicCookie, transmitterAddress, receiverAddress, fragmentNumber, userDataLength));
}
+ private AmsSerialResetFrame(FragmentNumber fragmentNumber) {
+ this.magicCookie = MagicCookie.of(ID);
+ this.transmitterAddress = TransmitterAddress.RS232_COMM_ADDRESS;
+ this.receiverAddress = ReceiverAddress.RS232_COMM_ADDRESS;
+ this.fragmentNumber = FragmentNumber.of((byte) 0);
+ this.userDataLength = UserDataLength.of((byte) 0);
+ this.crc = CRC.of(DigestUtil.calculateCrc16(magicCookie, transmitterAddress, receiverAddress, fragmentNumber, userDataLength));
+ }
+
public static AmsSerialResetFrame of(MagicCookie magicCookie, TransmitterAddress transmitterAddress, ReceiverAddress receiverAddress, FragmentNumber fragmentNumber, UserDataLength userDataLength, CRC crc) {
return new AmsSerialResetFrame(magicCookie, transmitterAddress, receiverAddress, fragmentNumber, userDataLength, crc);
}
@@ -89,6 +98,10 @@ public class AmsSerialResetFrame implements ByteReadable {
return new AmsSerialResetFrame(transmitterAddress, receiverAddress, fragmentNumber);
}
+ public static AmsSerialResetFrame of(FragmentNumber fragmentNumber) {
+ return new AmsSerialResetFrame(fragmentNumber);
+ }
+
@Override
public ByteBuf getByteBuf() {
return buildByteBuff(magicCookie, transmitterAddress, receiverAddress, fragmentNumber, userDataLength, crc);
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 95193d0..3d7d90a 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.ChannelHandlerContext;
import io.netty.handler.codec.MessageToMessageCodec;
import io.netty.util.ReferenceCountUtil;
+import io.netty.util.concurrent.ScheduledFuture;
+import org.apache.commons.lang3.mutable.MutableInt;
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;
@@ -35,7 +37,9 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.List;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
public class Payload2SerialProtocol extends MessageToMessageCodec<ByteBuf, ByteBuf> {
@@ -43,20 +47,37 @@ public class Payload2SerialProtocol extends MessageToMessageCodec<ByteBuf, ByteB
private final AtomicInteger fragmentCounter = new AtomicInteger(0);
+ private AtomicReference<ScheduledFuture<?>> currentRetryer = new AtomicReference<>();
+
@Override
protected void encode(ChannelHandlerContext channelHandlerContext, ByteBuf amsPacket, List<Object> out) throws PlcProtocolPayloadTooBigException {
LOGGER.trace("(<--OUT): {}, {}, {}", channelHandlerContext, amsPacket, out);
- int fragmentNumber = fragmentCounter.getAndIncrement();
- if (fragmentNumber > 255) {
- fragmentNumber = 0;
- fragmentCounter.set(fragmentNumber);
- }
+ int fragmentNumber = fragmentCounter.getAndUpdate(value -> value > 255 ? 0 : ++value);
LOGGER.debug("Using fragmentNumber {} for {}", fragmentNumber, amsPacket);
UserData userData = UserData.of(amsPacket);
if (userData.getCalculatedLength() > 255) {
throw new PlcProtocolPayloadTooBigException("ADS/AMS", 255, (int) userData.getCalculatedLength(), amsPacket);
}
AmsSerialFrame amsSerialFrame = AmsSerialFrame.of(FragmentNumber.of((byte) fragmentNumber), userData);
+
+ MutableInt retryCount = new MutableInt(0);
+ ScheduledFuture<?> oldRetryer = currentRetryer.get();
+ if (oldRetryer != null) {
+ oldRetryer.cancel(false);
+ }
+ currentRetryer.set(channelHandlerContext.executor().scheduleAtFixedRate(() -> {
+ LOGGER.trace("Retrying {} the {} time", amsSerialFrame, retryCount);
+ int currentTry = retryCount.incrementAndGet();
+ if (currentTry > 10) {
+ // TODO: we might need to throw an exception to potentially cancel upstream waiting
+ channelHandlerContext.writeAndFlush(AmsSerialResetFrame.of(FragmentNumber.of((byte) fragmentNumber)));
+ PlcRuntimeException plcRuntimeException = new PlcRuntimeException("Retry exhausted after " + retryCount + " times");
+ channelHandlerContext.fireExceptionCaught(plcRuntimeException);
+ throw plcRuntimeException;
+ } else {
+ channelHandlerContext.writeAndFlush(amsSerialFrame);
+ }
+ }, 100, 100, TimeUnit.MILLISECONDS));
out.add(amsSerialFrame.getByteBuf());
}
@@ -92,6 +113,9 @@ public class Payload2SerialProtocol extends MessageToMessageCodec<ByteBuf, ByteB
}
CRC crc = CRC.of(byteBuf);
+ // we don't need to retransmit
+ currentRetryer.get().cancel(false);
+
Runnable postAction = null;
switch (magicCookie.getAsInt()) {
case AmsSerialFrame.ID:
--
To stop receiving notification emails like this one, please contact
sruehl@apache.org.