From b47944321e28a56f254576f93a8ef76b4388ed0d Mon Sep 17 00:00:00 2001 From: zhouxiang Date: Wed, 27 Apr 2022 17:50:10 +0800 Subject: [PATCH] [ISSUE #3949] Do refector and Fix unit test * Create Channel use Context * Use LocalReceiveMessageResponseStreamWriter to write ReceiveMessageResponse * Change polling time to 1 second to reduce test time --- .../proxy/channel/ChannelManager.java | 36 ++--- .../proxy/channel/InvocationChannel.java | 5 +- .../v1/adapter/channel/GrpcClientChannel.java | 3 +- .../proxy/grpc/v2/adapter/GrpcConverter.java | 23 +++ .../grpc/v2/adapter/ResponseBuilder.java | 3 + .../v2/adapter/channel/GrpcClientChannel.java | 3 +- .../channel/ReceiveMessageChannel.java | 11 +- .../adapter/channel/SendMessageChannel.java | 5 +- .../ReceiveMessageResponseHandler.java | 31 +--- .../grpc/v2/service/LocalGrpcService.java | 86 +++++++---- .../ReceiveMessageResponseStreamObserver.java | 67 ++++++++ .../ReceiveMessageResponseStreamWriter.java | 8 +- .../ReceiveMessageResultFilter.java | 2 +- .../v2/service/cluster/ConsumerService.java | 3 +- ...ultReceiveMessageResponseStreamWriter.java | 52 +------ .../DefaultReceiveMessageResultFilter.java | 27 +--- ...calReceiveMessageResponseStreamWriter.java | 143 ++++++++++++++++++ .../LocalReceiveMessageResultFilter.java | 123 +++++++++++++++ .../grpc/v2/service/LocalGrpcServiceTest.java | 2 +- .../rocketmq/test/grpc/v2/GrpcBaseTest.java | 10 +- 20 files changed, 480 insertions(+), 163 deletions(-) create mode 100644 proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/service/ReceiveMessageResponseStreamObserver.java rename proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/service/{cluster => }/ReceiveMessageResponseStreamWriter.java (88%) rename proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/service/{cluster => }/ReceiveMessageResultFilter.java (95%) create mode 100644 proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/service/local/LocalReceiveMessageResponseStreamWriter.java create mode 100644 proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/service/local/LocalReceiveMessageResultFilter.java diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/channel/ChannelManager.java b/proxy/src/main/java/org/apache/rocketmq/proxy/channel/ChannelManager.java index 925d923705..87b6ea2750 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/channel/ChannelManager.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/channel/ChannelManager.java @@ -24,7 +24,7 @@ import java.util.List; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; -import java.util.function.Supplier; +import java.util.function.Function; import org.apache.commons.lang3.StringUtils; import org.apache.rocketmq.common.constant.LoggerName; import org.apache.rocketmq.logging.InternalLogger; @@ -38,25 +38,25 @@ public class ChannelManager { protected final ConcurrentMap clientIdChannelMap = new ConcurrentHashMap<>(); protected final ConcurrentMap/* clientId */> groupClientIdMap = new ConcurrentHashMap<>(); - public SimpleChannel createChannel() { - return createChannel(anonymousChannelId()); + public SimpleChannel createChannel(Context ctx) { + return createChannel(ctx, anonymousChannelId(ctx)); } - public SimpleChannel createChannel(String clientId) { - return createChannel(clientId, ChannelManager::createSimpleChannelDirectly, SimpleChannel.class); + public SimpleChannel createChannel(Context ctx, String clientId) { + return createChannel(ctx, clientId, ChannelManager::createSimpleChannelDirectly, SimpleChannel.class); } - public T createChannel(Supplier creator, Class clazz) { - return createChannel(anonymousChannelId(clazz.getName()), creator, clazz); + public T createChannel(Context ctx, Function creator, Class clazz) { + return createChannel(ctx, anonymousChannelId(ctx, clazz.getName()), creator, clazz); } - public T createChannel(String clientId, Supplier creator, Class clazz) { + public T createChannel(Context ctx, String clientId, Function creator, Class clazz) { if (StringUtils.isBlank(clientId)) { log.warn("ClientId is unexpected null or empty"); - return creator.get(); + return creator.apply(ctx); } - clientIdChannelMap.computeIfAbsent(clientId, key -> creator.get()); + clientIdChannelMap.computeIfAbsent(clientId, key -> creator.apply(ctx)); T channel = clazz.cast(clientIdChannelMap.get(clientId)); channel.updateLastAccessTime(); @@ -83,26 +83,22 @@ public class ChannelManager { return clazz.cast(channel); } - private String anonymousChannelId() { - final String clientHost = InterceptorConstants.METADATA.get(Context.current()) + private String anonymousChannelId(Context ctx) { + final String clientHost = InterceptorConstants.METADATA.get(ctx) .get(InterceptorConstants.REMOTE_ADDRESS); - final String localAddress = InterceptorConstants.METADATA.get(Context.current()) + final String localAddress = InterceptorConstants.METADATA.get(ctx) .get(InterceptorConstants.LOCAL_ADDRESS); return clientHost + "@" + localAddress; } - private String anonymousChannelId(String className) { - final String clientHost = InterceptorConstants.METADATA.get(Context.current()) + private String anonymousChannelId(Context ctx, String className) { + final String clientHost = InterceptorConstants.METADATA.get(ctx) .get(InterceptorConstants.REMOTE_ADDRESS); - final String localAddress = InterceptorConstants.METADATA.get(Context.current()) + final String localAddress = InterceptorConstants.METADATA.get(ctx) .get(InterceptorConstants.LOCAL_ADDRESS); return className + "@" + clientHost + "@" + localAddress; } - public static SimpleChannel createSimpleChannelDirectly() { - return createSimpleChannelDirectly(Context.current()); - } - public static SimpleChannel createSimpleChannelDirectly(Context ctx) { final String clientHost = InterceptorConstants.METADATA.get(ctx) .get(InterceptorConstants.REMOTE_ADDRESS); diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/channel/InvocationChannel.java b/proxy/src/main/java/org/apache/rocketmq/proxy/channel/InvocationChannel.java index 83dc2428cc..2ae543f254 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/channel/InvocationChannel.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/channel/InvocationChannel.java @@ -17,6 +17,7 @@ package org.apache.rocketmq.proxy.channel; +import io.grpc.Context; import io.netty.channel.ChannelFuture; import java.util.Iterator; import java.util.Map; @@ -30,8 +31,8 @@ public abstract class InvocationChannel extends SimpleChannel implements C protected final ConcurrentMap> inFlightRequestMap; protected final ResponseHandler handler; - public InvocationChannel(ResponseHandler handler) { - super(ChannelManager.createSimpleChannelDirectly()); + public InvocationChannel(Context ctx, ResponseHandler handler) { + super(ChannelManager.createSimpleChannelDirectly(ctx)); this.inFlightRequestMap = new ConcurrentHashMap<>(); this.handler = handler; } diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v1/adapter/channel/GrpcClientChannel.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v1/adapter/channel/GrpcClientChannel.java index 99f9ec6601..5792d78cf6 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v1/adapter/channel/GrpcClientChannel.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v1/adapter/channel/GrpcClientChannel.java @@ -74,8 +74,9 @@ public class GrpcClientChannel extends SimpleChannel { TelemetryCommandManager manager ) { GrpcClientChannel channel = channelManager.createChannel( + ctx, buildKey(group, clientId), - () -> new GrpcClientChannel(ctx, group, clientId, manager), + context -> new GrpcClientChannel(context, group, clientId, manager), GrpcClientChannel.class ); diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/adapter/GrpcConverter.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/adapter/GrpcConverter.java index d8913cfedb..883abf6291 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/adapter/GrpcConverter.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/adapter/GrpcConverter.java @@ -237,6 +237,10 @@ public class GrpcConverter { return requestHeader; } + public static AckMessageRequestHeader buildAckMessageRequestHeader(ReceiveMessageRequest request, ReceiptHandle handle) { + return buildAckMessageRequestHeader(request.getMessageQueue().getTopic(), request.getGroup(), handle); + } + public static AckMessageRequestHeader buildAckMessageRequestHeader(AckMessageRequest request, ReceiptHandle handle) { return buildAckMessageRequestHeader(request.getTopic(), request.getGroup(), handle); } @@ -305,6 +309,25 @@ public class GrpcConverter { return changeInvisibleTimeRequestHeader; } + public static ChangeInvisibleTimeRequestHeader buildChangeInvisibleTimeRequestHeader(ReceiveMessageRequest request, ReceiptHandle handle) { + String groupName = GrpcConverter.wrapResourceWithNamespace(request.getGroup()); + String topicName = GrpcConverter.wrapResourceWithNamespace(request.getMessageQueue().getTopic()); + + ChangeInvisibleTimeRequestHeader changeInvisibleTimeRequestHeader = new ChangeInvisibleTimeRequestHeader(); + changeInvisibleTimeRequestHeader.setConsumerGroup(groupName); + changeInvisibleTimeRequestHeader.setTopic(handle.getRealTopic(topicName, groupName)); + changeInvisibleTimeRequestHeader.setQueueId(handle.getQueueId()); + changeInvisibleTimeRequestHeader.setExtraInfo(handle.getReceiptHandle()); + changeInvisibleTimeRequestHeader.setOffset(handle.getOffset()); + changeInvisibleTimeRequestHeader.setInvisibleTime(Durations.toMillis(request.getInvisibleDuration())); + return changeInvisibleTimeRequestHeader; + } + + public static ConsumerSendMsgBackRequestHeader buildConsumerSendMsgBackRequestHeader(ReceiveMessageRequest request, + ReceiptHandle handle, String messageId, int maxReconsumeTimes) { + return buildConsumerSendMsgBackRequestHeader(request.getMessageQueue().getTopic(), request.getGroup(), handle, messageId, maxReconsumeTimes); + } + public static ConsumerSendMsgBackRequestHeader buildConsumerSendMsgBackToDLQRequestHeader( NackMessageRequest request, int maxReconsumeTimes) { ReceiptHandle handle = ReceiptHandle.decode(request.getReceiptHandle()); diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/adapter/ResponseBuilder.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/adapter/ResponseBuilder.java index 994088bce6..0344f28193 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/adapter/ResponseBuilder.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/adapter/ResponseBuilder.java @@ -100,6 +100,9 @@ public class ResponseBuilder { case ResponseCode.FLUSH_SLAVE_TIMEOUT: { return Code.SLAVE_PERSISTENCE_TIMEOUT; } + case ResponseCode.POLLING_TIMEOUT: { + return Code.GATEWAY_TIMEOUT; + } default: { return Code.INTERNAL_SERVER_ERROR; } diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/adapter/channel/GrpcClientChannel.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/adapter/channel/GrpcClientChannel.java index 9a2650239c..7966ab7410 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/adapter/channel/GrpcClientChannel.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/adapter/channel/GrpcClientChannel.java @@ -72,8 +72,9 @@ public class GrpcClientChannel extends SimpleChannel { TelemetryCommandManager manager ) { GrpcClientChannel channel = channelManager.createChannel( + ctx, buildKey(group, clientId), - () -> new GrpcClientChannel(ctx, group, clientId, manager), + context -> new GrpcClientChannel(context, group, clientId, manager), GrpcClientChannel.class ); diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/adapter/channel/ReceiveMessageChannel.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/adapter/channel/ReceiveMessageChannel.java index 8593b66023..3187df724c 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/adapter/channel/ReceiveMessageChannel.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/adapter/channel/ReceiveMessageChannel.java @@ -18,13 +18,14 @@ package org.apache.rocketmq.proxy.grpc.v2.adapter.channel; import apache.rocketmq.v2.ReceiveMessageRequest; -import apache.rocketmq.v2.ReceiveMessageResponse; -import java.util.Iterator; +import io.grpc.Context; +import java.util.List; +import org.apache.rocketmq.common.message.MessageExt; import org.apache.rocketmq.proxy.channel.InvocationChannel; import org.apache.rocketmq.proxy.grpc.v2.adapter.handler.ReceiveMessageResponseHandler; -public class ReceiveMessageChannel extends InvocationChannel> { - public ReceiveMessageChannel(ReceiveMessageResponseHandler handler) { - super(handler); +public class ReceiveMessageChannel extends InvocationChannel> { + public ReceiveMessageChannel(Context ctx, ReceiveMessageResponseHandler handler) { + super(ctx, handler); } } diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/adapter/channel/SendMessageChannel.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/adapter/channel/SendMessageChannel.java index 39ce672299..a2b034529d 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/adapter/channel/SendMessageChannel.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/adapter/channel/SendMessageChannel.java @@ -19,11 +19,12 @@ package org.apache.rocketmq.proxy.grpc.v2.adapter.channel; import apache.rocketmq.v2.SendMessageRequest; import apache.rocketmq.v2.SendMessageResponse; +import io.grpc.Context; import org.apache.rocketmq.proxy.channel.InvocationChannel; import org.apache.rocketmq.proxy.grpc.v2.adapter.handler.SendMessageResponseHandler; public class SendMessageChannel extends InvocationChannel { - public SendMessageChannel(SendMessageResponseHandler handler) { - super(handler); + public SendMessageChannel(Context context, SendMessageResponseHandler handler) { + super(context, handler); } } diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/adapter/handler/ReceiveMessageResponseHandler.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/adapter/handler/ReceiveMessageResponseHandler.java index 23ffb5ff6b..ef867235cd 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/adapter/handler/ReceiveMessageResponseHandler.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/adapter/handler/ReceiveMessageResponseHandler.java @@ -18,13 +18,10 @@ package org.apache.rocketmq.proxy.grpc.v2.adapter.handler; import apache.rocketmq.v2.ReceiveMessageRequest; -import apache.rocketmq.v2.ReceiveMessageResponse; -import apache.rocketmq.v2.Resource; import com.google.common.base.Stopwatch; import java.nio.ByteBuffer; import java.util.ArrayList; import java.util.HashMap; -import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.concurrent.CompletableFuture; @@ -39,11 +36,10 @@ import org.apache.rocketmq.logging.InternalLogger; import org.apache.rocketmq.logging.InternalLoggerFactory; import org.apache.rocketmq.proxy.channel.InvocationContext; import org.apache.rocketmq.proxy.grpc.v2.adapter.GrpcConverter; -import org.apache.rocketmq.proxy.grpc.v2.adapter.ResponseBuilder; import org.apache.rocketmq.remoting.protocol.RemotingCommand; import org.apache.rocketmq.remoting.protocol.RemotingSysResponseCode; -public class ReceiveMessageResponseHandler implements ResponseHandler> { +public class ReceiveMessageResponseHandler implements ResponseHandler> { private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME); private final String brokerName; private final boolean fifo; @@ -55,17 +51,19 @@ public class ReceiveMessageResponseHandler implements ResponseHandler> context) { - CompletableFuture> future = context.getResponse(); + InvocationContext> context) { + CompletableFuture> future = context.getResponse(); long currentTimeInMillis = System.currentTimeMillis(); long popCosts = currentTimeInMillis - context.getTimestamp(); try { Stopwatch stopWatch = Stopwatch.createStarted(); PopMessageResponseHeader responseHeader = (PopMessageResponseHeader) responseCommand.readCustomHeader(); - List responseList = new ArrayList<>(); + List allMessageList = new ArrayList<>(); + ReceiveMessageRequest request = context.getRequest(); if (responseCommand.getCode() == RemotingSysResponseCode.SUCCESS) { + String topicName = GrpcConverter.wrapResourceWithNamespace(request.getMessageQueue().getTopic()); ByteBuffer byteBuffer = ByteBuffer.wrap(responseCommand.getBody()); List msgFoundList = MessageDecoder.decodes(byteBuffer); @@ -130,29 +128,16 @@ public class ReceiveMessageResponseHandler implements ResponseHandler String.valueOf(responseHeader.getPopTime())); } - - responseList.add(ReceiveMessageResponse.newBuilder() - .setStatus(ResponseBuilder.buildStatus(responseCommand.getCode(), responseCommand.getRemark())) - .build()); - - for (MessageExt messageExt : msgFoundList) { - ReceiveMessageResponse response = ReceiveMessageResponse.newBuilder() - .setMessage(GrpcConverter.buildMessage(messageExt)).build(); - responseList.add(response); - } + allMessageList.addAll(msgFoundList); } long elapsed = stopWatch.stop().elapsed(TimeUnit.MILLISECONDS); log.debug("Translating remoting response to gRPC response costs {}ms. Duration request received: {}", elapsed, popCosts); - future.complete(responseList.iterator()); + future.complete(allMessageList); } catch (Exception e) { log.error("Unexpected exception raised when handle pop remoting command", e); future.completeExceptionally(e); diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/service/LocalGrpcService.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/service/LocalGrpcService.java index effcc98ec5..6f4270a787 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/service/LocalGrpcService.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/service/LocalGrpcService.java @@ -52,7 +52,6 @@ import io.grpc.Context; import io.grpc.stub.StreamObserver; import io.netty.channel.Channel; import java.util.ArrayList; -import java.util.Iterator; import java.util.List; import java.util.concurrent.CompletableFuture; import java.util.concurrent.Executors; @@ -64,12 +63,14 @@ import org.apache.rocketmq.broker.client.ConsumerGroupEvent; import org.apache.rocketmq.broker.client.ConsumerIdsChangeListener; import org.apache.rocketmq.broker.client.ProducerChangeListener; import org.apache.rocketmq.broker.client.ProducerGroupEvent; +import org.apache.rocketmq.client.consumer.PopStatus; import org.apache.rocketmq.common.MQVersion; import org.apache.rocketmq.common.ThreadFactoryImpl; import org.apache.rocketmq.common.constant.LoggerName; import org.apache.rocketmq.common.consumer.ReceiptHandle; import org.apache.rocketmq.common.message.MessageBatch; import org.apache.rocketmq.common.message.MessageClientIDSetter; +import org.apache.rocketmq.common.message.MessageExt; import org.apache.rocketmq.common.message.MessageQueue; import org.apache.rocketmq.common.protocol.RequestCode; import org.apache.rocketmq.common.protocol.ResponseCode; @@ -97,16 +98,17 @@ import org.apache.rocketmq.proxy.common.TelemetryCommandRecord; import org.apache.rocketmq.proxy.connector.ConnectorManager; import org.apache.rocketmq.proxy.grpc.interceptor.InterceptorConstants; import org.apache.rocketmq.proxy.grpc.v2.adapter.GrpcConverter; -import org.apache.rocketmq.proxy.grpc.v2.adapter.ProxyException; import org.apache.rocketmq.proxy.grpc.v2.adapter.ResponseBuilder; -import org.apache.rocketmq.proxy.grpc.v2.adapter.ResponseWriter; +import org.apache.rocketmq.proxy.grpc.v2.adapter.ResponseHook; import org.apache.rocketmq.proxy.grpc.v2.adapter.channel.GrpcClientChannel; import org.apache.rocketmq.proxy.grpc.v2.adapter.channel.ReceiveMessageChannel; import org.apache.rocketmq.proxy.grpc.v2.adapter.channel.SendMessageChannel; import org.apache.rocketmq.proxy.grpc.v2.adapter.handler.ReceiveMessageResponseHandler; import org.apache.rocketmq.proxy.grpc.v2.adapter.handler.SendMessageResponseHandler; -import org.apache.rocketmq.proxy.grpc.v2.service.local.RouteService; +import org.apache.rocketmq.proxy.grpc.v2.service.local.LocalReceiveMessageResponseStreamWriter; +import org.apache.rocketmq.proxy.grpc.v2.service.local.LocalReceiveMessageResultFilter; import org.apache.rocketmq.proxy.grpc.v2.service.local.LocalWriteQueueSelector; +import org.apache.rocketmq.proxy.grpc.v2.service.local.RouteService; import org.apache.rocketmq.remoting.RemotingServer; import org.apache.rocketmq.remoting.netty.NettyRemotingAbstract; import org.apache.rocketmq.remoting.protocol.LanguageCode; @@ -125,6 +127,9 @@ public class LocalGrpcService extends AbstractStartAndShutdown implements GrpcFo private final ClientSettingsService clientSettingsService; private final DelayPolicy delayPolicy; private final LocalWriteQueueSelector localWriteQueueSelector; + private final ReceiveMessageResponseStreamWriter.Builder streamWriterBuilder; + + private volatile ResponseHook receiveMessageHook; public LocalGrpcService(BrokerController brokerController) { this(brokerController, new TelemetryCommandManager()); @@ -151,6 +156,14 @@ public class LocalGrpcService extends AbstractStartAndShutdown implements GrpcFo this.brokerController.getConsumerManager().appendConsumerIdsChangeListener(new ConsumerIdsChangeListenerImpl()); this.brokerController.getProducerManager().appendProducerChangeListener(new ProducerChangeListenerImpl()); + this.streamWriterBuilder = (observer, hook) -> new LocalReceiveMessageResponseStreamWriter( + observer, + hook, + channelManager, + brokerController, + new LocalReceiveMessageResultFilter(channelManager, brokerController, grpcClientManager) + ); + this.appendStartAndShutdown(connectorManager); } @@ -235,7 +248,7 @@ public class LocalGrpcService extends AbstractStartAndShutdown implements GrpcFo command.makeCustomHeaderToNet(); SendMessageResponseHandler handler = new SendMessageResponseHandler(messageId, requestHeader.getSysFlag(), brokerController.getBrokerAddr()); - SendMessageChannel channel = channelManager.createChannel(() -> new SendMessageChannel(handler), SendMessageChannel.class); + SendMessageChannel channel = channelManager.createChannel(ctx, context -> new SendMessageChannel(context, handler), SendMessageChannel.class); SimpleChannelHandlerContext channelHandlerContext = new SimpleChannelHandlerContext(channel); CompletableFuture future = new CompletableFuture<>(); InvocationContext context @@ -266,16 +279,16 @@ public class LocalGrpcService extends AbstractStartAndShutdown implements GrpcFo long pollTime = ctx.getDeadline().timeRemaining(TimeUnit.MILLISECONDS); // TODO: get fifo config from subscriptionGroupManager boolean fifo = false; + ReceiveMessageResponseStreamWriter writer = streamWriterBuilder.build(responseObserver, receiveMessageHook); + ReceiveMessageResponseHandler handler = new ReceiveMessageResponseHandler(brokerController.getBrokerConfig().getBrokerName(), fifo); + ReceiveMessageChannel channel = channelManager.createChannel(ctx, context -> new ReceiveMessageChannel(context, handler), ReceiveMessageChannel.class); + CompletableFuture> future = new CompletableFuture<>(); + InvocationContext> context + = new InvocationContext<>(request, future); + SimpleChannelHandlerContext channelHandlerContext = new SimpleChannelHandlerContext(channel); PopMessageRequestHeader requestHeader = GrpcConverter.buildPopMessageRequestHeader(request, pollTime, fifo); RemotingCommand command = RemotingCommand.createRequestCommand(RequestCode.POP_MESSAGE, requestHeader); command.makeCustomHeaderToNet(); - - ReceiveMessageResponseHandler handler = new ReceiveMessageResponseHandler(brokerController.getBrokerConfig().getBrokerName(), fifo); - ReceiveMessageChannel channel = channelManager.createChannel(() -> new ReceiveMessageChannel(handler), ReceiveMessageChannel.class); - SimpleChannelHandlerContext channelHandlerContext = new SimpleChannelHandlerContext(channel); - CompletableFuture> future = new CompletableFuture<>(); - InvocationContext> context - = new InvocationContext<>(request, future); channel.registerInvocationContext(command.getOpaque(), context); try { RemotingCommand response = brokerController.getPopMessageProcessor().processRequest(channelHandlerContext, command); @@ -288,26 +301,33 @@ public class LocalGrpcService extends AbstractStartAndShutdown implements GrpcFo channel.eraseInvocationContext(command.getOpaque()); future.completeExceptionally(e); } - future.thenAccept(r -> ResponseWriter.write(responseObserver, r)) - .exceptionally(e -> { - ResponseWriter.write( - responseObserver, - ReceiveMessageResponse.newBuilder().setStatus(ResponseBuilder.buildStatus(e)).build() - ); - return null; - }); + future.thenAccept(r -> { + try { + writer.write(ctx, request, PopStatus.FOUND, r); + } finally { + responseObserver.onCompleted(); + } + }).exceptionally(e -> { + writer.write(ctx, request, e); + return null; + }); } @Override public CompletableFuture ackMessage(Context ctx, AckMessageRequest request) { - Channel channel = channelManager.createChannel(); + Channel channel = channelManager.createChannel(ctx); SimpleChannelHandlerContext channelHandlerContext = new SimpleChannelHandlerContext(channel); CompletableFuture future = new CompletableFuture<>(); List ackMessageResultEntryList = new ArrayList<>(); for (AckMessageEntry entry : request.getEntriesList()) { ReceiptHandle receiptHandle = ReceiptHandle.decode(entry.getReceiptHandle()); if (receiptHandle.isExpired()) { - throw new ProxyException(Code.RECEIPT_HANDLE_EXPIRED, "handle has expired"); + ackMessageResultEntryList.add(AckMessageResultEntry.newBuilder() + .setReceiptHandle(entry.getReceiptHandle()) + .setMessageId(entry.getMessageId()) + .setStatus(ResponseBuilder.buildStatus(Code.RECEIPT_HANDLE_EXPIRED, "expired")) + .build()); + continue; } AckMessageRequestHeader requestHeader = GrpcConverter.buildAckMessageRequestHeader(request, receiptHandle); RemotingCommand command = RemotingCommand.createRequestCommand(RequestCode.ACK_MESSAGE, requestHeader); @@ -340,7 +360,7 @@ public class LocalGrpcService extends AbstractStartAndShutdown implements GrpcFo @Override public CompletableFuture nackMessage(Context ctx, NackMessageRequest request) { - Channel channel = channelManager.createChannel(); + Channel channel = channelManager.createChannel(ctx); SimpleChannelHandlerContext channelHandlerContext = new SimpleChannelHandlerContext(channel); CompletableFuture future = new CompletableFuture<>(); @@ -385,7 +405,7 @@ public class LocalGrpcService extends AbstractStartAndShutdown implements GrpcFo @Override public CompletableFuture forwardMessageToDeadLetterQueue(Context ctx, ForwardMessageToDeadLetterQueueRequest request) { - SimpleChannel channel = channelManager.createChannel(); + SimpleChannel channel = channelManager.createChannel(ctx); SimpleChannelHandlerContext channelHandlerContext = new SimpleChannelHandlerContext(channel); ConsumerSendMsgBackRequestHeader requestHeader = GrpcConverter.buildConsumerSendMsgBackRequestHeader(request); @@ -409,7 +429,7 @@ public class LocalGrpcService extends AbstractStartAndShutdown implements GrpcFo @Override public CompletableFuture endTransaction(Context ctx, EndTransactionRequest request) { - Channel channel = channelManager.createChannel(); + Channel channel = channelManager.createChannel(ctx); SimpleChannelHandlerContext channelHandlerContext = new SimpleChannelHandlerContext(channel); String topicName = GrpcConverter.wrapResourceWithNamespace(request.getTopic()); @@ -436,7 +456,7 @@ public class LocalGrpcService extends AbstractStartAndShutdown implements GrpcFo return future; } - public void reportThreadStackTrace(ThreadStackTrace request) { + public void reportThreadStackTrace(Context ctx, ThreadStackTrace request) { String nonce = request.getNonce(); String threadStack = request.getThreadStackTrace(); TelemetryCommandRecord pollCommandResponseFuture = telemetryCommandManager.getCommand(nonce); @@ -451,13 +471,13 @@ public class LocalGrpcService extends AbstractStartAndShutdown implements GrpcFo ConsumerRunningInfo runningInfo = new ConsumerRunningInfo(); runningInfo.setJstack(threadStack); remotingCommand.setBody(runningInfo.encode()); - nettyRemotingAbstract.processResponseCommand(new SimpleChannelHandlerContext(channelManager.createChannel()), remotingCommand); + nettyRemotingAbstract.processResponseCommand(new SimpleChannelHandlerContext(channelManager.createChannel(ctx)), remotingCommand); } } } } - public void reportVerifyMessageResult(VerifyMessageResult request) { + public void reportVerifyMessageResult(Context ctx, VerifyMessageResult request) { String nonce = request.getNonce(); TelemetryCommandRecord pollCommandResponseFuture = telemetryCommandManager.getCommand(nonce); if (pollCommandResponseFuture != null) { @@ -470,7 +490,7 @@ public class LocalGrpcService extends AbstractStartAndShutdown implements GrpcFo remotingCommand.setOpaque(pollCommandResponseFuture.getOpaque()); ConsumeMessageDirectlyResult result = GrpcConverter.buildConsumeMessageDirectlyResult(request); remotingCommand.setBody(result.encode()); - nettyRemotingAbstract.processResponseCommand(new SimpleChannelHandlerContext(channelManager.createChannel()), remotingCommand); + nettyRemotingAbstract.processResponseCommand(new SimpleChannelHandlerContext(channelManager.createChannel(ctx)), remotingCommand); } } } @@ -479,7 +499,7 @@ public class LocalGrpcService extends AbstractStartAndShutdown implements GrpcFo @Override public CompletableFuture notifyClientTermination(Context ctx, NotifyClientTerminationRequest request) { - Channel channel = channelManager.createChannel(); + Channel channel = channelManager.createChannel(ctx); SimpleChannelHandlerContext simpleChannelHandlerContext = new SimpleChannelHandlerContext(channel); String clientId = InterceptorConstants.METADATA.get(ctx).get(InterceptorConstants.CLIENT_ID); Settings clientSettings = grpcClientManager.getClientSettings(clientId); @@ -497,7 +517,7 @@ public class LocalGrpcService extends AbstractStartAndShutdown implements GrpcFo @Override public CompletableFuture changeInvisibleDuration(Context ctx, ChangeInvisibleDurationRequest request) { - Channel channel = channelManager.createChannel(); + Channel channel = channelManager.createChannel(ctx); SimpleChannelHandlerContext channelHandlerContext = new SimpleChannelHandlerContext(channel); ChangeInvisibleTimeRequestHeader requestHeader = GrpcConverter.buildChangeInvisibleTimeRequestHeader(request); @@ -546,11 +566,11 @@ public class LocalGrpcService extends AbstractStartAndShutdown implements GrpcFo break; } case THREAD_STACK_TRACE: { - reportThreadStackTrace(request.getThreadStackTrace()); + reportThreadStackTrace(ctx, request.getThreadStackTrace()); break; } case VERIFY_MESSAGE_RESULT: { - reportVerifyMessageResult(request.getVerifyMessageResult()); + reportVerifyMessageResult(ctx, request.getVerifyMessageResult()); break; } default: { diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/service/ReceiveMessageResponseStreamObserver.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/service/ReceiveMessageResponseStreamObserver.java new file mode 100644 index 0000000000..16a2f29d43 --- /dev/null +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/service/ReceiveMessageResponseStreamObserver.java @@ -0,0 +1,67 @@ +/* + * 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.rocketmq.proxy.grpc.v2.service; + +import apache.rocketmq.v2.ReceiveMessageRequest; +import apache.rocketmq.v2.ReceiveMessageResponse; +import io.grpc.Context; +import io.grpc.stub.StreamObserver; +import org.apache.rocketmq.proxy.grpc.v2.adapter.ResponseHook; +import org.apache.rocketmq.proxy.grpc.v2.adapter.ResponseWriter; + +public class ReceiveMessageResponseStreamObserver implements StreamObserver { + + private final Context context; + private final ReceiveMessageRequest request; + private final ResponseHook receiveMessageHook; + private final StreamObserver observer; + + public ReceiveMessageResponseStreamObserver(Context context, ReceiveMessageRequest request, + ResponseHook receiveMessageHook, + StreamObserver observer) { + this.context = context; + this.request = request; + this.receiveMessageHook = receiveMessageHook; + this.observer = observer; + } + + @Override + public void onNext(ReceiveMessageResponse response) { + if (receiveMessageHook != null) { + receiveMessageHook.beforeResponse(context, request, response, null); + } + observer.onNext(response); + } + + @Override + public void onError(Throwable throwable) { + if (receiveMessageHook != null) { + receiveMessageHook.beforeResponse(context, request, null, throwable); + } + observer.onError(throwable); + } + + @Override + public void onCompleted() { + observer.onCompleted(); + } + + public boolean isCancelled() { + return ResponseWriter.isCancelled(observer); + } +} diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/ReceiveMessageResponseStreamWriter.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/service/ReceiveMessageResponseStreamWriter.java similarity index 88% rename from proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/ReceiveMessageResponseStreamWriter.java rename to proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/service/ReceiveMessageResponseStreamWriter.java index c1d0625165..8723558b71 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/ReceiveMessageResponseStreamWriter.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/service/ReceiveMessageResponseStreamWriter.java @@ -14,13 +14,15 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.rocketmq.proxy.grpc.v2.service.cluster; +package org.apache.rocketmq.proxy.grpc.v2.service; import apache.rocketmq.v2.ReceiveMessageRequest; import apache.rocketmq.v2.ReceiveMessageResponse; import io.grpc.Context; import io.grpc.stub.StreamObserver; -import org.apache.rocketmq.client.consumer.PopResult; +import java.util.List; +import org.apache.rocketmq.client.consumer.PopStatus; +import org.apache.rocketmq.common.message.MessageExt; import org.apache.rocketmq.proxy.grpc.v2.adapter.ResponseHook; public abstract class ReceiveMessageResponseStreamWriter { @@ -41,7 +43,7 @@ public abstract class ReceiveMessageResponseStreamWriter { receiveMessageHook = hook; } - public abstract void write(Context ctx, ReceiveMessageRequest request, PopResult result); + public abstract void write(Context ctx, ReceiveMessageRequest request, PopStatus status, List messageFoundList); public abstract void write(Context ctx, ReceiveMessageRequest request, Throwable throwable); } diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/ReceiveMessageResultFilter.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/service/ReceiveMessageResultFilter.java similarity index 95% rename from proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/ReceiveMessageResultFilter.java rename to proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/service/ReceiveMessageResultFilter.java index cd291b493b..398dc5d790 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/ReceiveMessageResultFilter.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/service/ReceiveMessageResultFilter.java @@ -15,7 +15,7 @@ * limitations under the License. */ -package org.apache.rocketmq.proxy.grpc.v2.service.cluster; +package org.apache.rocketmq.proxy.grpc.v2.service; import apache.rocketmq.v2.Message; import apache.rocketmq.v2.ReceiveMessageRequest; diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/ConsumerService.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/ConsumerService.java index a900ae6bcc..e805708e69 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/ConsumerService.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/ConsumerService.java @@ -52,6 +52,7 @@ import org.apache.rocketmq.proxy.grpc.v2.adapter.ResponseBuilder; import org.apache.rocketmq.proxy.grpc.v2.adapter.ResponseHook; import org.apache.rocketmq.proxy.grpc.v2.service.BaseService; import org.apache.rocketmq.proxy.grpc.v2.service.GrpcClientManager; +import org.apache.rocketmq.proxy.grpc.v2.service.ReceiveMessageResponseStreamWriter; import org.apache.rocketmq.remoting.protocol.RemotingCommand; public class ConsumerService extends BaseService { @@ -106,7 +107,7 @@ public class ConsumerService extends BaseService { messageQueue.getBrokerName(), requestHeader, requestHeader.getPollTime()) - .thenAccept(result -> writer.write(ctx, request, result)) + .thenAccept(result -> writer.write(ctx, request, result.getPopStatus(), result.getMsgFoundList())) .exceptionally(e -> { writer.write(ctx, request, e); return null; diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/DefaultReceiveMessageResponseStreamWriter.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/DefaultReceiveMessageResponseStreamWriter.java index 8210b23f28..9595670646 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/DefaultReceiveMessageResponseStreamWriter.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/DefaultReceiveMessageResponseStreamWriter.java @@ -26,10 +26,10 @@ import java.time.Duration; import java.util.Iterator; import java.util.List; import org.apache.rocketmq.client.consumer.AckStatus; -import org.apache.rocketmq.client.consumer.PopResult; import org.apache.rocketmq.client.consumer.PopStatus; import org.apache.rocketmq.common.constant.LoggerName; import org.apache.rocketmq.common.consumer.ReceiptHandle; +import org.apache.rocketmq.common.message.MessageExt; import org.apache.rocketmq.common.protocol.header.ChangeInvisibleTimeRequestHeader; import org.apache.rocketmq.logging.InternalLogger; import org.apache.rocketmq.logging.InternalLoggerFactory; @@ -40,6 +40,9 @@ import org.apache.rocketmq.proxy.grpc.v2.adapter.ResponseBuilder; import org.apache.rocketmq.proxy.grpc.v2.adapter.ResponseHook; import org.apache.rocketmq.proxy.grpc.v2.adapter.ResponseWriter; import org.apache.rocketmq.proxy.grpc.v2.service.BaseService; +import org.apache.rocketmq.proxy.grpc.v2.service.ReceiveMessageResponseStreamObserver; +import org.apache.rocketmq.proxy.grpc.v2.service.ReceiveMessageResponseStreamWriter; +import org.apache.rocketmq.proxy.grpc.v2.service.ReceiveMessageResultFilter; public class DefaultReceiveMessageResponseStreamWriter extends ReceiveMessageResponseStreamWriter { protected static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME); @@ -62,8 +65,7 @@ public class DefaultReceiveMessageResponseStreamWriter extends ReceiveMessageRes } @Override - public void write(Context ctx, ReceiveMessageRequest request, PopResult result) { - PopStatus status = result.getPopStatus(); + public void write(Context ctx, ReceiveMessageRequest request, PopStatus status, List messageFoundList) { ReceiveMessageResponseStreamObserver responseStreamObserver = new ReceiveMessageResponseStreamObserver( ctx, request, @@ -72,7 +74,7 @@ public class DefaultReceiveMessageResponseStreamWriter extends ReceiveMessageRes try { switch (status) { case FOUND: - List messageList = this.receiveMessageResultFilter.filterMessage(ctx, request, result.getMsgFoundList()); + List messageList = this.receiveMessageResultFilter.filterMessage(ctx, request, messageFoundList); if (messageList.isEmpty()) { responseStreamObserver.onNext(ReceiveMessageResponse.newBuilder() .setStatus(ResponseBuilder.buildStatus(Code.OK, "no new message")) @@ -159,48 +161,6 @@ public class DefaultReceiveMessageResponseStreamWriter extends ReceiveMessageRes } } - protected static class ReceiveMessageResponseStreamObserver implements StreamObserver { - - private final Context context; - private final ReceiveMessageRequest request; - private final ResponseHook receiveMessageHook; - private final StreamObserver observer; - - public ReceiveMessageResponseStreamObserver(Context context, ReceiveMessageRequest request, - ResponseHook receiveMessageHook, - StreamObserver observer) { - this.context = context; - this.request = request; - this.receiveMessageHook = receiveMessageHook; - this.observer = observer; - } - - @Override - public void onNext(ReceiveMessageResponse response) { - if (receiveMessageHook != null) { - receiveMessageHook.beforeResponse(context, request, response, null); - } - observer.onNext(response); - } - - @Override - public void onError(Throwable throwable) { - if (receiveMessageHook != null) { - receiveMessageHook.beforeResponse(context, request, null, throwable); - } - observer.onError(throwable); - } - - @Override - public void onCompleted() { - observer.onCompleted(); - } - - public boolean isCancelled() { - return ResponseWriter.isCancelled(observer); - } - } - public ReceiveMessageResultFilter getReceiveMessageResultFilter() { return receiveMessageResultFilter; } diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/DefaultReceiveMessageResultFilter.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/DefaultReceiveMessageResultFilter.java index db9fac9088..684de08881 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/DefaultReceiveMessageResultFilter.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/DefaultReceiveMessageResultFilter.java @@ -39,6 +39,7 @@ import org.apache.rocketmq.proxy.connector.route.TopicRouteCache; import org.apache.rocketmq.proxy.grpc.v2.adapter.GrpcConverter; import org.apache.rocketmq.proxy.grpc.v2.adapter.ResponseHook; import org.apache.rocketmq.proxy.grpc.v2.service.GrpcClientManager; +import org.apache.rocketmq.proxy.grpc.v2.service.ReceiveMessageResultFilter; import org.apache.rocketmq.remoting.protocol.RemotingCommand; import static org.apache.rocketmq.proxy.grpc.v2.service.BaseService.getBrokerAddr; @@ -98,18 +99,12 @@ public class DefaultReceiveMessageResultFilter implements ReceiveMessageResultFi return; } String brokerAddr = getBrokerAddr(ctx, topicRouteCache, handle.getBrokerName()); - Resource topic = request.getMessageQueue().getTopic(); - Resource group = request.getGroup(); ConsumerSendMsgBackRequestHeader sendMsgBackRequestHeader = GrpcConverter.buildConsumerSendMsgBackRequestHeader( - topic, - group, + request, handle, messageExt.getMsgId(), maxReconsumeTimes); - AckMessageRequestHeader ackMessageRequestHeader = GrpcConverter.buildAckMessageRequestHeader( - topic, - group, - handle); + AckMessageRequestHeader ackMessageRequestHeader = GrpcConverter.buildAckMessageRequestHeader(request, handle); future = this.producer.sendMessageBackThenAckOrg(ctx, brokerAddr, sendMsgBackRequestHeader, ackMessageRequestHeader); } catch (Throwable t) { @@ -126,19 +121,13 @@ public class DefaultReceiveMessageResultFilter implements ReceiveMessageResultFi protected void ackNoMatchedMessage(Context ctx, ReceiveMessageRequest request, MessageExt messageExt) { CompletableFuture future = new CompletableFuture<>(); - AckMessageRequestHeader ackMessageRequestHeader = new AckMessageRequestHeader(); + ReceiptHandle handle = ReceiptHandle.create(messageExt); + if (handle == null) { + return; + } + AckMessageRequestHeader ackMessageRequestHeader = GrpcConverter.buildAckMessageRequestHeader(request, handle); try { - ReceiptHandle handle = ReceiptHandle.create(messageExt); - if (handle == null) { - return; - } String brokerAddr = getBrokerAddr(ctx, topicRouteCache, handle.getBrokerName()); - ackMessageRequestHeader.setConsumerGroup(GrpcConverter.wrapResourceWithNamespace(request.getGroup())); - ackMessageRequestHeader.setTopic(messageExt.getTopic()); - ackMessageRequestHeader.setQueueId(handle.getQueueId()); - ackMessageRequestHeader.setExtraInfo(handle.getReceiptHandle()); - ackMessageRequestHeader.setOffset(handle.getOffset()); - future = this.writeConsumer.ackMessage(ctx, brokerAddr, ackMessageRequestHeader); } catch (Throwable t) { future.completeExceptionally(t); diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/service/local/LocalReceiveMessageResponseStreamWriter.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/service/local/LocalReceiveMessageResponseStreamWriter.java new file mode 100644 index 0000000000..258afd1058 --- /dev/null +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/service/local/LocalReceiveMessageResponseStreamWriter.java @@ -0,0 +1,143 @@ +/* + * 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.rocketmq.proxy.grpc.v2.service.local; + +import apache.rocketmq.v2.Code; +import apache.rocketmq.v2.Message; +import apache.rocketmq.v2.ReceiveMessageRequest; +import apache.rocketmq.v2.ReceiveMessageResponse; +import io.grpc.Context; +import io.grpc.stub.StreamObserver; +import io.netty.channel.Channel; +import java.util.Iterator; +import java.util.List; +import org.apache.rocketmq.broker.BrokerController; +import org.apache.rocketmq.client.consumer.PopStatus; +import org.apache.rocketmq.common.constant.LoggerName; +import org.apache.rocketmq.common.consumer.ReceiptHandle; +import org.apache.rocketmq.common.message.MessageExt; +import org.apache.rocketmq.common.protocol.RequestCode; +import org.apache.rocketmq.common.protocol.header.ChangeInvisibleTimeRequestHeader; +import org.apache.rocketmq.proxy.channel.ChannelManager; +import org.apache.rocketmq.proxy.channel.SimpleChannelHandlerContext; +import org.apache.rocketmq.proxy.grpc.v2.adapter.GrpcConverter; +import org.apache.rocketmq.proxy.grpc.v2.adapter.ResponseBuilder; +import org.apache.rocketmq.proxy.grpc.v2.adapter.ResponseHook; +import org.apache.rocketmq.proxy.grpc.v2.adapter.ResponseWriter; +import org.apache.rocketmq.proxy.grpc.v2.service.ReceiveMessageResponseStreamObserver; +import org.apache.rocketmq.proxy.grpc.v2.service.ReceiveMessageResponseStreamWriter; +import org.apache.rocketmq.proxy.grpc.v2.service.ReceiveMessageResultFilter; +import org.apache.rocketmq.remoting.exception.RemotingCommandException; +import org.apache.rocketmq.remoting.protocol.RemotingCommand; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class LocalReceiveMessageResponseStreamWriter extends ReceiveMessageResponseStreamWriter { + private final static Logger log = LoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME); + private final ChannelManager channelManager; + private final BrokerController brokerController; + private final ReceiveMessageResultFilter receiveMessageResultFilter; + + public LocalReceiveMessageResponseStreamWriter( + StreamObserver observer, + ResponseHook hook, + ChannelManager channelManager, + BrokerController brokerController, + ReceiveMessageResultFilter receiveMessageResultFilter) { + super(observer, hook); + this.channelManager = channelManager; + this.brokerController = brokerController; + this.receiveMessageResultFilter = receiveMessageResultFilter; + } + + @Override + public void write(Context ctx, ReceiveMessageRequest request, PopStatus status, List messageFoundList) { + ReceiveMessageResponseStreamObserver responseStreamObserver = new ReceiveMessageResponseStreamObserver( + ctx, + request, + receiveMessageHook, + streamObserver); + try { + switch (status) { + case FOUND: + List messageList = this.receiveMessageResultFilter.filterMessage(ctx, request, messageFoundList); + if (messageList.isEmpty()) { + responseStreamObserver.onNext(ReceiveMessageResponse.newBuilder() + .setStatus(ResponseBuilder.buildStatus(Code.OK, "no new message")) + .build()); + } else { + responseStreamObserver.onNext(ReceiveMessageResponse.newBuilder() + .setStatus(ResponseBuilder.buildStatus(Code.OK, Code.OK.name())) + .build()); + Iterator messageIterator = messageList.iterator(); + while (messageIterator.hasNext()) { + if (responseStreamObserver.isCancelled()) { + break; + } + responseStreamObserver.onNext(ReceiveMessageResponse.newBuilder() + .setMessage(messageIterator.next()) + .build()); + } + messageIterator.forEachRemaining(message -> this.changeInvisibleTime(ctx, request, ReceiptHandle.decode(message.getSystemProperties().getReceiptHandle()))); + } + break; + case POLLING_FULL: + responseStreamObserver.onNext(ReceiveMessageResponse.newBuilder() + .setStatus(ResponseBuilder.buildStatus(Code.TOO_MANY_REQUESTS, "polling full")) + .build()); + break; + case NO_NEW_MSG: + case POLLING_NOT_FOUND: + default: + responseStreamObserver.onNext(ReceiveMessageResponse.newBuilder() + .setStatus(ResponseBuilder.buildStatus(Code.OK, "no new message")) + .build()); + break; + } + } catch (Throwable t) { + write(ctx, request, t); + } finally { + responseStreamObserver.onCompleted(); + } + } + + @Override public void write(Context ctx, ReceiveMessageRequest request, Throwable throwable) { + ReceiveMessageResponseStreamObserver responseStreamObserver = new ReceiveMessageResponseStreamObserver( + ctx, + request, + receiveMessageHook, + streamObserver); + ResponseWriter.write( + responseStreamObserver, + ReceiveMessageResponse.newBuilder().setStatus(ResponseBuilder.buildStatus(throwable)).build() + ); + } + + private void changeInvisibleTime(Context ctx, ReceiveMessageRequest request, ReceiptHandle handle) { + Channel channel = channelManager.createChannel(ctx); + SimpleChannelHandlerContext simpleChannelHandlerContext = new SimpleChannelHandlerContext(channel); + ChangeInvisibleTimeRequestHeader requestHeader = GrpcConverter.buildChangeInvisibleTimeRequestHeader(request, handle); + RemotingCommand command = RemotingCommand.createRequestCommand(RequestCode.CHANGE_MESSAGE_INVISIBLETIME, requestHeader); + command.makeCustomHeaderToNet(); + try { + brokerController.getChangeInvisibleTimeProcessor().processRequest(simpleChannelHandlerContext, command); + } catch (RemotingCommandException e) { + log.error("ChangeInvisibleTime error when write response", e); + } + } +} diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/service/local/LocalReceiveMessageResultFilter.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/service/local/LocalReceiveMessageResultFilter.java new file mode 100644 index 0000000000..62ea8fb78c --- /dev/null +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/service/local/LocalReceiveMessageResultFilter.java @@ -0,0 +1,123 @@ +/* + * 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.rocketmq.proxy.grpc.v2.service.local; + +import apache.rocketmq.v2.Message; +import apache.rocketmq.v2.ReceiveMessageRequest; +import apache.rocketmq.v2.Settings; +import io.grpc.Context; +import io.netty.channel.Channel; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import org.apache.rocketmq.broker.BrokerController; +import org.apache.rocketmq.common.constant.LoggerName; +import org.apache.rocketmq.common.consumer.ReceiptHandle; +import org.apache.rocketmq.common.message.MessageExt; +import org.apache.rocketmq.common.protocol.RequestCode; +import org.apache.rocketmq.common.protocol.ResponseCode; +import org.apache.rocketmq.common.protocol.header.AckMessageRequestHeader; +import org.apache.rocketmq.common.protocol.header.ConsumerSendMsgBackRequestHeader; +import org.apache.rocketmq.common.protocol.heartbeat.SubscriptionData; +import org.apache.rocketmq.proxy.channel.ChannelManager; +import org.apache.rocketmq.proxy.channel.SimpleChannelHandlerContext; +import org.apache.rocketmq.proxy.common.utils.FilterUtils; +import org.apache.rocketmq.proxy.grpc.v2.adapter.GrpcConverter; +import org.apache.rocketmq.proxy.grpc.v2.service.GrpcClientManager; +import org.apache.rocketmq.proxy.grpc.v2.service.ReceiveMessageResultFilter; +import org.apache.rocketmq.remoting.exception.RemotingCommandException; +import org.apache.rocketmq.remoting.protocol.RemotingCommand; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class LocalReceiveMessageResultFilter implements ReceiveMessageResultFilter { + private final static Logger log = LoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME); + + private final ChannelManager channelManager; + private final BrokerController brokerController; + private final GrpcClientManager grpcClientManager; + + public LocalReceiveMessageResultFilter(ChannelManager channelManager, BrokerController brokerController, GrpcClientManager grpcClientManager) { + this.channelManager = channelManager; + this.brokerController = brokerController; + this.grpcClientManager = grpcClientManager; + } + + @Override + public List filterMessage(Context ctx, ReceiveMessageRequest request, List messageExtList) { + if (messageExtList == null || messageExtList.isEmpty()) { + return Collections.emptyList(); + } + String topicName = GrpcConverter.wrapResourceWithNamespace(request.getMessageQueue().getTopic()); + SubscriptionData subscriptionData = GrpcConverter.buildSubscriptionData(topicName, request.getFilterExpression()); + Settings settings = grpcClientManager.getClientSettings(ctx); + int maxAttempts = settings.getBackoffPolicy().getMaxAttempts(); + List resMessageList = new ArrayList<>(); + for (MessageExt messageExt : messageExtList) { + if (!FilterUtils.isTagMatched(subscriptionData.getTagsSet(), messageExt.getTags())) { + ackMessage(ctx, request, messageExt); + continue; + } + if (messageExt.getReconsumeTimes() >= maxAttempts) { + forwardMessageToDLQ(ctx, request, messageExt, maxAttempts); + continue; + } + resMessageList.add(GrpcConverter.buildMessage(messageExt)); + } + return resMessageList; + } + + private void ackMessage(Context ctx, ReceiveMessageRequest request, MessageExt messageExt) { + ReceiptHandle handle = ReceiptHandle.create(messageExt); + if (handle == null) { + return; + } + Channel channel = channelManager.createChannel(ctx); + AckMessageRequestHeader requestHeader = GrpcConverter.buildAckMessageRequestHeader(request, handle); + RemotingCommand command = RemotingCommand.createRequestCommand(RequestCode.ACK_MESSAGE, requestHeader); + command.makeCustomHeaderToNet(); + try { + brokerController.getAckMessageProcessor().processRequest(new SimpleChannelHandlerContext(channel), command); + } catch (RemotingCommandException e) { + log.error("AckMessage failed in filterMessage", e); + } + } + + private void forwardMessageToDLQ(Context ctx, ReceiveMessageRequest request, MessageExt messageExt, int maxAttempt) { + try { + ReceiptHandle handle = ReceiptHandle.create(messageExt); + if (handle == null) { + return; + } + Channel channel = channelManager.createChannel(ctx); + SimpleChannelHandlerContext simpleChannelHandlerContext = new SimpleChannelHandlerContext(channel); + ConsumerSendMsgBackRequestHeader consumerSendMsgBackRequestHeader = GrpcConverter.buildConsumerSendMsgBackRequestHeader(request, handle, messageExt.getMsgId(), maxAttempt); + RemotingCommand command = RemotingCommand.createRequestCommand(RequestCode.CONSUMER_SEND_MSG_BACK, consumerSendMsgBackRequestHeader); + command.makeCustomHeaderToNet(); + RemotingCommand response = brokerController.getSendMessageProcessor().processRequest(simpleChannelHandlerContext, command); + if (response.getCode() == ResponseCode.SUCCESS) { + AckMessageRequestHeader ackMessageRequestHeader = GrpcConverter.buildAckMessageRequestHeader(request, handle); + command = RemotingCommand.createRequestCommand(RequestCode.ACK_MESSAGE, ackMessageRequestHeader); + command.makeCustomHeaderToNet(); + brokerController.getAckMessageProcessor().processRequest(simpleChannelHandlerContext, command); + } + } catch (Exception e) { + log.error("ForwardMessageToDLQ failed in filterMessage", e); + } + } +} diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/service/LocalGrpcServiceTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/service/LocalGrpcServiceTest.java index 28f705e75f..a290c62439 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/service/LocalGrpcServiceTest.java +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/service/LocalGrpcServiceTest.java @@ -332,7 +332,7 @@ public class LocalGrpcServiceTest extends InitConfigAndLoggerTest { .build()) .build(); ReceiveMessageResponse receiveMessageResponse1 = ReceiveMessageResponse.newBuilder() - .setStatus(ResponseBuilder.buildStatus(ResponseCode.SUCCESS, null)) + .setStatus(ResponseBuilder.buildStatus(Code.OK, Code.OK.name())) .build(); Message message = GrpcConverter.buildMessage(messageExt); ReceiveMessageResponse receiveMessageResponse2 = ReceiveMessageResponse.newBuilder() diff --git a/test/src/test/java/org/apache/rocketmq/test/grpc/v2/GrpcBaseTest.java b/test/src/test/java/org/apache/rocketmq/test/grpc/v2/GrpcBaseTest.java index 5aa4a24f0e..7c4379bacc 100644 --- a/test/src/test/java/org/apache/rocketmq/test/grpc/v2/GrpcBaseTest.java +++ b/test/src/test/java/org/apache/rocketmq/test/grpc/v2/GrpcBaseTest.java @@ -214,7 +214,7 @@ public class GrpcBaseTest extends BaseConf { // init consumer offset this.sendClientSettings(stub, buildPushConsumerClientSettings()).get(); - receiveMessage(blockingStub, topic, group); + receiveMessage(blockingStub, topic, group, 1); String messageId = createUniqID(); this.sendClientSettings(stub, buildProducerClientSettings(topic)).get(); @@ -235,7 +235,7 @@ public class GrpcBaseTest extends BaseConf { // init consumer offset this.sendClientSettings(stub, buildPushConsumerClientSettings()).get(); - receiveMessage(blockingStub, topic, group); + receiveMessage(blockingStub, topic, group, 1); this.sendClientSettings(stub, buildProducerClientSettings(topic)).get(); String messageId = createUniqID(); @@ -313,7 +313,7 @@ public class GrpcBaseTest extends BaseConf { }); telemetryCommandRef.set(null); // init consumer offset - receiveMessage(blockingStub, topic, group); + receiveMessage(blockingStub, topic, group, 1); requestStreamObserver.onNext(TelemetryCommand.newBuilder() .setSettings(buildProducerClientSettings(topic)) @@ -379,7 +379,7 @@ public class GrpcBaseTest extends BaseConf { // init consumer offset this.sendClientSettings(stub, buildSimpleConsumerClientSettings(maxDeliveryAttempts, fifo)).get(); - receiveMessage(blockingStub, topic, group); + receiveMessage(blockingStub, topic, group, 1); this.sendClientSettings(stub, buildProducerClientSettings(topic)).get(); String messageId = createUniqID(); @@ -436,7 +436,7 @@ public class GrpcBaseTest extends BaseConf { // init consumer offset this.sendClientSettings(stub, buildSimpleConsumerClientSettings(maxDeliveryAttempts, fifo)).get(); - receiveMessage(blockingStub, topic, group); + receiveMessage(blockingStub, topic, group, 1); this.sendClientSettings(stub, buildProducerClientSettings(topic)).get(); String messageId = createUniqID();