This is an automated email from the ASF dual-hosted git repository.
mxsm pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/eventmesh.git
The following commit(s) were added to refs/heads/master by this push:
new 1f0bf0c77 [ISSUE #4701]Fix use tcp protocol client send message, it
throw a DecoderException (#4702)
1f0bf0c77 is described below
commit 1f0bf0c7722fe220f01b0fd01feaa2a8a41e818f
Author: mxsm <[email protected]>
AuthorDate: Wed Jan 10 09:07:53 2024 +0800
[ISSUE #4701]Fix use tcp protocol client send message, it throw a
DecoderException (#4702)
* [ISSUE #4701]fix use tcp protocol client send message, it throw a
DecoderException
* optmize code
* refactor with LengthFieldBasedFrameDecoder
* fix code style
* optimize code
* fix log print
* optimize code and add some comments
---
.../eventmesh/common/protocol/tcp/codec/Codec.java | 85 +++++++++++++---------
.../common/protocol/tcp/codec/CodecTest.java | 14 ++--
.../eventmesh/client/tcp/common/TcpClient.java | 2 +-
3 files changed, 59 insertions(+), 42 deletions(-)
diff --git
a/eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/tcp/codec/Codec.java
b/eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/tcp/codec/Codec.java
index cded7d67e..2417021b6 100644
---
a/eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/tcp/codec/Codec.java
+++
b/eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/tcp/codec/Codec.java
@@ -31,14 +31,11 @@ import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.StringUtils;
import java.util.Arrays;
-import java.util.List;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
-import io.netty.handler.codec.ByteToMessageCodec;
+import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.MessageToByteEncoder;
-import io.netty.handler.codec.ReplayingDecoder;
-
import com.fasterxml.jackson.core.JsonProcessingException;
import com.google.common.base.Preconditions;
@@ -46,25 +43,16 @@ import com.google.common.base.Preconditions;
import lombok.extern.slf4j.Slf4j;
@Slf4j
-public class Codec extends ByteToMessageCodec<Package> {
+public class Codec {
private static final int FRAME_MAX_LENGTH = 1024 * 1024 * 4;
private static final byte[] CONSTANT_MAGIC_FLAG =
serializeBytes("EventMesh");
private static final byte[] VERSION = serializeBytes("0000");
- private Encoder encoder = new Encoder();
- private Decoder decoder = new Decoder();
+ private static final int PREFIX_LENGTH = CONSTANT_MAGIC_FLAG.length +
VERSION.length; //13
- @Override
- protected void encode(ChannelHandlerContext ctx, Package pkg, ByteBuf out)
throws Exception {
- encoder.encode(ctx, pkg, out);
- }
-
- @Override
- protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object>
out) throws Exception {
- decoder.decode(ctx, in, out);
- }
+ private static final int PACKAGE_BYTES_FIELD_LENGTH = 4;
public static class Encoder extends MessageToByteEncoder<Package> {
@@ -89,7 +77,7 @@ public class Codec extends ByteToMessageCodec<Package> {
int headerLength = ArrayUtils.getLength(headerData);
int bodyLength = ArrayUtils.getLength(bodyData);
- final int length = CONSTANT_MAGIC_FLAG.length + VERSION.length +
headerLength + bodyLength;
+ final int length = PREFIX_LENGTH + headerLength + bodyLength;
if (length > FRAME_MAX_LENGTH) {
throw new IllegalArgumentException("message size is exceed
limit!");
@@ -116,31 +104,62 @@ public class Codec extends ByteToMessageCodec<Package> {
}
}
- public static class Decoder extends ReplayingDecoder<Package> {
+ public static class Decoder extends LengthFieldBasedFrameDecoder {
+
+ public Decoder() {
+ /**
+ * lengthAdjustment value = -9 explain:
+ * Header + Body, Format:
+ * <pre>
+ *
┌───────────────┬─────────────┬──────────────────┬──────────────────┬──────────────────┬─────────────────┐
+ * │ MAGIC_FLAG │ VERSION │ package length │ Header
length │ Header │ body │
+ * │ (9bytes) │ (4bytes) │ (4bytes) │
(4bytes) │ (header bytes) │ (body bytes) │
+ *
└───────────────┴─────────────┴──────────────────┴──────────────────┴──────────────────┴─────────────────┘
+ * </pre>
+ * package length = MAGIC_FLAG + VERSION + Header length + Body
length,Currently,
+ * adding MAGIC_FLAG + VERSION + package length field (4 bytes)
actually adds 17 bytes.
+ * However, the value of the package length field is only reduced
by the four bytes of
+ * the package length field itself and the four bytes of the
header length field.
+ * Therefore, the compensation value to be added to the length
field value is -9,
+ * which means subtracting the extra 9 bytes.
+ * Refer to the encoding in the {@link Encoder}
+ */
+ super(FRAME_MAX_LENGTH, PREFIX_LENGTH, PACKAGE_BYTES_FIELD_LENGTH,
-9, 0);
+ }
@Override
- public void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object>
out) throws Exception {
+ protected Object decode(ChannelHandlerContext ctx, ByteBuf in) throws
Exception {
+
+ ByteBuf target = null;
+
try {
- if (null == in) {
- return;
+ target = (ByteBuf) super.decode(ctx, in);
+ if (null == target) {
+ return null;
}
-
- byte[] flagBytes = parseFlag(in);
- byte[] versionBytes = parseVersion(in);
+ byte[] flagBytes = parseFlag(target);
+ byte[] versionBytes = parseVersion(target);
validateFlag(flagBytes, versionBytes, ctx);
- final int length = in.readInt();
- final int headerLength = in.readInt();
- final int bodyLength = length - CONSTANT_MAGIC_FLAG.length -
VERSION.length - headerLength;
- Header header = parseHeader(in, headerLength);
- Object body = parseBody(in, header, bodyLength);
+ final int length = target.readInt();
+ final int headerLength = target.readInt();
+ final int bodyLength = length - PREFIX_LENGTH - headerLength;
+ Header header = parseHeader(target, headerLength);
+ Object body = parseBody(target, header, bodyLength);
Package pkg = new Package(header, body);
- out.add(pkg);
- } catch (Exception e) {
- log.error("decode error| received data: {}.",
deserializeBytes(in.array()), e);
- throw e;
+ return pkg;
+
+ } catch (Exception ex) {
+ log.error("decode error", ex);
+ ctx.channel().close();
+ } finally {
+ if (target != null) {
+ target.release();
+ }
}
+
+ return null;
}
private byte[] parseFlag(ByteBuf in) {
diff --git
a/eventmesh-common/src/test/java/org/apache/eventmesh/common/protocol/tcp/codec/CodecTest.java
b/eventmesh-common/src/test/java/org/apache/eventmesh/common/protocol/tcp/codec/CodecTest.java
index 29724e9b6..19907ffd2 100644
---
a/eventmesh-common/src/test/java/org/apache/eventmesh/common/protocol/tcp/codec/CodecTest.java
+++
b/eventmesh-common/src/test/java/org/apache/eventmesh/common/protocol/tcp/codec/CodecTest.java
@@ -20,8 +20,8 @@ package org.apache.eventmesh.common.protocol.tcp.codec;
import org.apache.eventmesh.common.protocol.tcp.Command;
import org.apache.eventmesh.common.protocol.tcp.Header;
import org.apache.eventmesh.common.protocol.tcp.Package;
-
-import java.util.ArrayList;
+import org.apache.eventmesh.common.protocol.tcp.codec.Codec.Decoder;
+import org.apache.eventmesh.common.protocol.tcp.codec.Codec.Encoder;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -37,14 +37,12 @@ public class CodecTest {
header.setCmd(Command.HELLO_REQUEST);
Package testP = new Package(header);
testP.setBody(new Object());
- Codec.Encoder ce = new Codec.Encoder();
+ Encoder ce = new Codec.Encoder();
ByteBuf buf = PooledByteBufAllocator.DEFAULT.buffer();
ce.encode(null, testP, buf);
- Codec.Decoder cd = new Codec.Decoder();
- ArrayList<Object> result = new ArrayList<>();
- cd.decode(null, buf, result);
- Assertions.assertNotNull(result.get(0));
- Assertions.assertEquals(testP.getHeader(), ((Package)
result.get(0)).getHeader());
+ Decoder cd = new Codec.Decoder();
+ final Package decode = (Package) cd.decode(null, buf);
+ Assertions.assertEquals(testP.getHeader(), decode.getHeader());
}
}
diff --git
a/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/tcp/common/TcpClient.java
b/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/tcp/common/TcpClient.java
index fed4af541..74a88945d 100644
---
a/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/tcp/common/TcpClient.java
+++
b/eventmesh-sdks/eventmesh-sdk-java/src/main/java/org/apache/eventmesh/client/tcp/common/TcpClient.java
@@ -110,7 +110,7 @@ public abstract class TcpClient implements Closeable {
@Override
public void initChannel(SocketChannel ch) {
- ch.pipeline().addLast(new Codec())
+ ch.pipeline().addLast(new Codec.Encoder(), new Codec.Decoder())
.addLast(handler, newExceptionHandler());
}
});
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
