You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@eventmesh.apache.org by mi...@apache.org on 2022/09/24 11:47:34 UTC
[incubator-eventmesh] branch protocol-amqp updated: Implement the decode base class
This is an automated email from the ASF dual-hosted git repository.
mikexue pushed a commit to branch protocol-amqp
in repository https://gitbox.apache.org/repos/asf/incubator-eventmesh.git
The following commit(s) were added to refs/heads/protocol-amqp by this push:
new 0d2f0ee5 Implement the decode base class
new 4e168334 Merge pull request #1408 from wangshaojie4039/protocol-amqp
0d2f0ee5 is described below
commit 0d2f0ee5f029eca7c407701701df31b4f3a4757b
Author: wangshaojie <wa...@cmss.chinamobile.com>
AuthorDate: Sat Sep 24 19:19:20 2022 +0800
Implement the decode base class
---
.../amqp/remoting/codec/AbstractBatchDecoder.java | 420 +++++++++++++++++++++
1 file changed, 420 insertions(+)
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/amqp/remoting/codec/AbstractBatchDecoder.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/amqp/remoting/codec/AbstractBatchDecoder.java
new file mode 100644
index 00000000..c42de46a
--- /dev/null
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/amqp/remoting/codec/AbstractBatchDecoder.java
@@ -0,0 +1,420 @@
+/*
+ * 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.eventmesh.runtime.core.protocol.amqp.remoting.codec;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.CompositeByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.handler.codec.ByteToMessageDecoder;
+import io.netty.handler.codec.DecoderException;
+import io.netty.util.internal.RecyclableArrayList;
+import io.netty.util.internal.StringUtil;
+
+
+public abstract class AbstractBatchDecoder extends ChannelInboundHandlerAdapter {
+ /**
+ * Cumulate {@link ByteBuf}s by merge them into one {@link ByteBuf}'s, using memory copies.
+ */
+ public static final Cumulator MERGE_CUMULATOR = new Cumulator() {
+ @Override
+ public ByteBuf cumulate(ByteBufAllocator alloc,
+ ByteBuf cumulation,
+ ByteBuf in) {
+ ByteBuf buffer;
+ if (cumulation.writerIndex() > cumulation
+ .maxCapacity()
+ - in.readableBytes()
+ || cumulation.refCnt() > 1) {
+ // Expand cumulation (by replace it) when either there is not more room in the buffer
+ // or if the refCnt is greater then 1 which may happen when the user use slice().retain() or
+ // duplicate().retain().
+ //
+ // See:
+ // - https://github.com/netty/netty/issues/2327
+ // - https://github.com/netty/netty/issues/1764
+ buffer = expandCumulation(alloc,
+ cumulation,
+ in.readableBytes());
+ } else {
+ buffer = cumulation;
+ }
+ buffer.writeBytes(in);
+ in.release();
+ return buffer;
+ }
+ };
+
+ /**
+ * Cumulate {@link ByteBuf}s by add them to a {@link CompositeByteBuf} and so do no memory copy whenever possible.
+ * Be aware that {@link CompositeByteBuf} use a more complex indexing implementation so depending on your use-case
+ * and the decoder implementation this may be slower then just use the {@link #MERGE_CUMULATOR}.
+ */
+ public static final Cumulator COMPOSITE_CUMULATOR = new Cumulator() {
+ @Override
+ public ByteBuf cumulate(ByteBufAllocator alloc,
+ ByteBuf cumulation,
+ ByteBuf in) {
+ ByteBuf buffer;
+ if (cumulation.refCnt() > 1) {
+ // Expand cumulation (by replace it) when the refCnt is greater then 1 which may happen when the user
+ // use slice().retain() or duplicate().retain().
+ //
+ // See:
+ // - https://github.com/netty/netty/issues/2327
+ // - https://github.com/netty/netty/issues/1764
+ buffer = expandCumulation(alloc,
+ cumulation,
+ in.readableBytes());
+ buffer.writeBytes(in);
+ in.release();
+ } else {
+ CompositeByteBuf composite;
+ if (cumulation instanceof CompositeByteBuf) {
+ composite = (CompositeByteBuf) cumulation;
+ } else {
+ int readable = cumulation
+ .readableBytes();
+ composite = alloc
+ .compositeBuffer();
+ composite.addComponent(
+ cumulation).writerIndex(
+ readable);
+ }
+ composite
+ .addComponent(in)
+ .writerIndex(
+ composite.writerIndex()
+ + in.readableBytes());
+ buffer = composite;
+ }
+ return buffer;
+ }
+ };
+
+ ByteBuf cumulation;
+ private Cumulator cumulator = MERGE_CUMULATOR;
+ private boolean singleDecode;
+ private boolean decodeWasNull;
+ private boolean first;
+ private int discardAfterReads = 16;
+ private int numReads;
+
+ /**
+ * If set then only one message is decoded on each {@link #channelRead(ChannelHandlerContext, Object)} call. This
+ * may be useful if you need to do some protocol upgrade and want to make sure nothing is mixed up.
+ * <p>
+ * Default is {@code false} as this has performance impacts.
+ */
+ public void setSingleDecode(boolean singleDecode) {
+ this.singleDecode = singleDecode;
+ }
+
+ /**
+ * If {@code true} then only one message is decoded on each {@link #channelRead(ChannelHandlerContext, Object)}
+ * call.
+ * <p>
+ * Default is {@code false} as this has performance impacts.
+ */
+ public boolean isSingleDecode() {
+ return singleDecode;
+ }
+
+ /**
+ * Set the {@link Cumulator} to use for cumulate the received {@link ByteBuf}s.
+ */
+ public void setCumulator(Cumulator cumulator) {
+ if (cumulator == null) {
+ throw new NullPointerException("cumulator");
+ }
+ this.cumulator = cumulator;
+ }
+
+ /**
+ * Set the number of reads after which {@link ByteBuf#discardSomeReadBytes()} are called and so free up memory. The
+ * default is {@code 16}.
+ */
+ public void setDiscardAfterReads(int discardAfterReads) {
+ if (discardAfterReads <= 0) {
+ throw new IllegalArgumentException("discardAfterReads must be > 0");
+ }
+ this.discardAfterReads = discardAfterReads;
+ }
+
+ /**
+ * Returns the actual number of readable bytes in the internal cumulative buffer of this decoder. You usually do not
+ * need to rely on this value to write a decoder. Use it only when you must use it at your own risk. This method is
+ * a shortcut to {@link #internalBuffer() internalBuffer().readableBytes()}.
+ */
+ protected int actualReadableBytes() {
+ return internalBuffer().readableBytes();
+ }
+
+ /**
+ * Returns the internal cumulative buffer of this decoder. You usually do not need to access the internal buffer
+ * directly to write a decoder. Use it only when you must use it at your own risk.
+ */
+ protected ByteBuf internalBuffer() {
+ if (cumulation != null) {
+ return cumulation;
+ } else {
+ return Unpooled.EMPTY_BUFFER;
+ }
+ }
+
+ @Override
+ public final void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
+ ByteBuf buf = internalBuffer();
+ int readable = buf.readableBytes();
+ if (readable > 0) {
+ ByteBuf bytes = buf.readBytes(readable);
+ buf.release();
+ ctx.fireChannelRead(bytes);
+ } else {
+ buf.release();
+ }
+ cumulation = null;
+ numReads = 0;
+ ctx.fireChannelReadComplete();
+ handlerRemoved0(ctx);
+ }
+
+ /**
+ * Gets called after the {@link ByteToMessageDecoder} was removed from the actual context and it doesn't handle
+ * events anymore.
+ */
+ protected void handlerRemoved0(ChannelHandlerContext ctx) throws Exception {
+ }
+
+ /**
+ * This method has been modified to check the size of decoded msgs, which is represented by the local variable
+ * {@code RecyclableArrayList out}. If has decoded more than one msg, then construct an array list to submit all
+ * decoded msgs to the pipeline.
+ *
+ * @param ctx channel handler context
+ * @param msg data
+ */
+ @Override
+ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
+ if (msg instanceof ByteBuf) {
+ RecyclableArrayList out = RecyclableArrayList.newInstance();
+ try {
+ ByteBuf data = (ByteBuf) msg;
+ first = cumulation == null;
+ if (first) {
+ cumulation = data;
+ } else {
+ cumulation = cumulator.cumulate(ctx.alloc(), cumulation, data);
+ }
+ callDecode(ctx, cumulation, out);
+ } catch (DecoderException e) {
+ throw e;
+ } catch (Throwable t) {
+ throw new DecoderException(t);
+ } finally {
+ if (cumulation != null && !cumulation.isReadable()) {
+ numReads = 0;
+ cumulation.release();
+ cumulation = null;
+ } else if (++numReads >= discardAfterReads) {
+ // We did enough reads already try to discard some bytes so we not risk to see a OOME.
+ // See https://github.com/netty/netty/issues/4275
+ numReads = 0;
+ discardSomeReadBytes();
+ }
+
+ int size = out.size();
+ if (size == 0) {
+ decodeWasNull = true;
+ } else if (size == 1) {
+ ctx.fireChannelRead(out.get(0));
+ } else {
+ ArrayList<Object> ret = new ArrayList<Object>(size);
+ for (int i = 0; i < size; i++) {
+ ret.add(out.get(i));
+ }
+ ctx.fireChannelRead(ret);
+ }
+
+ out.recycle();
+ }
+ } else {
+ ctx.fireChannelRead(msg);
+ }
+ }
+
+ @Override
+ public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
+ numReads = 0;
+ discardSomeReadBytes();
+ if (decodeWasNull) {
+ decodeWasNull = false;
+ if (!ctx.channel().config().isAutoRead()) {
+ ctx.read();
+ }
+ }
+ ctx.fireChannelReadComplete();
+ }
+
+ protected final void discardSomeReadBytes() {
+ if (cumulation != null && !first && cumulation.refCnt() == 1) {
+ // discard some bytes if possible to make more room in the
+ // buffer but only if the refCnt == 1 as otherwise the user may have
+ // used slice().retain() or duplicate().retain().
+ //
+ // See:
+ // - https://github.com/netty/netty/issues/2327
+ // - https://github.com/netty/netty/issues/1764
+ cumulation.discardSomeReadBytes();
+ }
+ }
+
+ @Override
+ public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+ RecyclableArrayList out = RecyclableArrayList.newInstance();
+ try {
+ if (cumulation != null) {
+ callDecode(ctx, cumulation, out);
+ decodeLast(ctx, cumulation, out);
+ } else {
+ decodeLast(ctx, Unpooled.EMPTY_BUFFER, out);
+ }
+ } catch (DecoderException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new DecoderException(e);
+ } finally {
+ try {
+ if (cumulation != null) {
+ cumulation.release();
+ cumulation = null;
+ }
+ int size = out.size();
+ for (int i = 0; i < size; i++) {
+ ctx.fireChannelRead(out.get(i));
+ }
+ if (size > 0) {
+ // Something was read, call fireChannelReadComplete()
+ ctx.fireChannelReadComplete();
+ }
+ ctx.fireChannelInactive();
+ } finally {
+ // recycle in all cases
+ out.recycle();
+ }
+ }
+ }
+
+ /**
+ * Called once data should be decoded from the given {@link ByteBuf}. This method will call {@link
+ * #decode(ChannelHandlerContext, ByteBuf, List)} as long as decoding should take place.
+ *
+ * @param ctx the {@link ChannelHandlerContext} which this {@link ByteToMessageDecoder} belongs to
+ * @param in the {@link ByteBuf} from which to read data
+ * @param out the {@link List} to which decoded messages should be added
+ */
+ protected void callDecode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
+ try {
+ while (in.isReadable()) {
+ int outSize = out.size();
+ int oldInputLength = in.readableBytes();
+ decode(ctx, in, out);
+
+ // Check if this handler was removed before continuing the loop.
+
+ // If it was removed, it is not safe to continue to operate on the buffer.
+ //
+ // See https://github.com/netty/netty/issues/1664
+ if (ctx.isRemoved()) {
+ break;
+ }
+
+ if (outSize == out.size()) {
+ if (oldInputLength == in.readableBytes()) {
+ break;
+ } else {
+ continue;
+ }
+ }
+
+ if (oldInputLength == in.readableBytes()) {
+ throw new DecoderException(
+ StringUtil.simpleClassName(getClass())
+ + ".decode() did not read anything but decoded a message.");
+ }
+
+ if (isSingleDecode()) {
+ break;
+ }
+ }
+ } catch (DecoderException e) {
+ throw e;
+ } catch (Throwable cause) {
+ throw new DecoderException(cause);
+ }
+ }
+
+ /**
+ * Is called one last time when the {@link ChannelHandlerContext} goes in-active. Which means the {@link
+ * #channelInactive(ChannelHandlerContext)} was triggered.
+ * <p>
+ * By default this will just call {@link #decode(ChannelHandlerContext, ByteBuf, List)} but sub-classes may override
+ * this for some special cleanup operation.
+ */
+ protected void decodeLast(ChannelHandlerContext ctx, ByteBuf in, List<Object> out)
+ throws Exception {
+ decode(ctx, in, out);
+ }
+
+ static ByteBuf expandCumulation(ByteBufAllocator alloc, ByteBuf cumulation, int readable) {
+ ByteBuf oldCumulation = cumulation;
+ cumulation = alloc.buffer(oldCumulation.readableBytes() + readable);
+ cumulation.writeBytes(oldCumulation);
+ oldCumulation.release();
+ return cumulation;
+ }
+
+ /**
+ * Cumulate {@link ByteBuf}s.
+ */
+ public interface Cumulator {
+ /**
+ * Cumulate the given {@link ByteBuf}s and return the {@link ByteBuf} that holds the cumulated bytes. The
+ * implementation is responsible to correctly handle the life-cycle of the given {@link ByteBuf}s and so call
+ * {@link ByteBuf#release()} if a {@link ByteBuf} is fully consumed.
+ */
+ ByteBuf cumulate(ByteBufAllocator alloc, ByteBuf cumulation, ByteBuf in);
+ }
+
+ /**
+ * Decode the from one {@link ByteBuf} to an other. This method will be called till either the input {@link ByteBuf}
+ * has nothing to read when return from this method or till nothing was read from the input {@link ByteBuf}.
+ *
+ * @param ctx the {@link ChannelHandlerContext} which this {@link ByteToMessageDecoder} belongs to
+ * @param in the {@link ByteBuf} from which to read data
+ * @param out the {@link List} to which decoded messages should be added
+ * @throws Exception is thrown if an error accour
+ */
+ protected abstract void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out)
+ throws Exception;
+}
\ No newline at end of file
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: commits-help@eventmesh.apache.org