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 2989941cc [ISSUE #4677]netty codec enhancement (#4678)
2989941cc is described below
commit 2989941cc34b0d1bacb6750c911014bf345cb94d
Author: Karson <[email protected]>
AuthorDate: Fri Dec 22 09:34:57 2023 +0800
[ISSUE #4677]netty codec enhancement (#4678)
---
.../eventmesh/common/protocol/tcp/codec/Codec.java | 17 ++++++++++++++++-
.../apache/eventmesh/client/tcp/common/TcpClient.java | 2 +-
2 files changed, 17 insertions(+), 2 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 6c9a2fadc..cded7d67e 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
@@ -35,22 +35,37 @@ 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.MessageToByteEncoder;
import io.netty.handler.codec.ReplayingDecoder;
+
import com.fasterxml.jackson.core.JsonProcessingException;
import com.google.common.base.Preconditions;
import lombok.extern.slf4j.Slf4j;
@Slf4j
-public class Codec {
+public class Codec extends ByteToMessageCodec<Package> {
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();
+
+ @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);
+ }
+
public static class Encoder extends MessageToByteEncoder<Package> {
@Override
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 c42d9af27..a8b164c6d 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.Encoder(), new Codec.Decoder())
+ ch.pipeline().addLast(new Codec())
.addLast(handler, newExceptionHandler());
}
});
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
