opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-09-24 16:04:00 +08:00
Code Issues Packages Projects Releases Wiki Activity

[ISSUE #3949] Do refector and Fix unit test

Browse Source 
* Create Channel use Context
* Use LocalReceiveMessageResponseStreamWriter to write ReceiveMessageResponse
* Change polling time to 1 second to reduce test time
This commit is contained in:
zhouxiang
2022-07-13 11:29:23 +08:00
parent f36f3c799d
commit b47944321e
20 changed files with 480 additions and 163 deletions
Download Patch File Download Diff File Expand all files Collapse all files
proxy/src/main/java/org/apache/rocketmq/proxy/channel/ChannelManager.java
+16 -20
View File
@@ -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<String /* clientId */, SimpleChannel> clientIdChannelMap = new ConcurrentHashMap<>();
protected final ConcurrentMap<String /* group */, Set<String>/* 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 extends SimpleChannel> T createChannel(Supplier<T> creator, Class<T> clazz) {
return createChannel(anonymousChannelId(clazz.getName()), creator, clazz);
public <T extends SimpleChannel> T createChannel(Context ctx, Function<Context, T> creator, Class<T> clazz) {
return createChannel(ctx, anonymousChannelId(ctx, clazz.getName()), creator, clazz);
}
public <T extends SimpleChannel> T createChannel(String clientId, Supplier<T> creator, Class<T> clazz) {
public <T extends SimpleChannel> T createChannel(Context ctx, String clientId, Function<Context, T> creator, Class<T> 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);
proxy/src/main/java/org/apache/rocketmq/proxy/channel/InvocationChannel.java
+3 -2
View File
@@ -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<R, W> extends SimpleChannel implements C
protected final ConcurrentMap<Integer, InvocationContext<R, W>> inFlightRequestMap;
protected final ResponseHandler<R, W> handler;
public InvocationChannel(ResponseHandler<R, W> handler) {
super(ChannelManager.createSimpleChannelDirectly());
public InvocationChannel(Context ctx, ResponseHandler<R, W> handler) {
super(ChannelManager.createSimpleChannelDirectly(ctx));
this.inFlightRequestMap = new ConcurrentHashMap<>();
this.handler = handler;
}
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v1/adapter/channel/GrpcClientChannel.java
+2 -1
View File
@@ -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
);
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/adapter/GrpcConverter.java
+23
View File
@@ -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());
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/adapter/ResponseBuilder.java
+3
View File
@@ -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;
}
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/adapter/channel/GrpcClientChannel.java
+2 -1
View File
@@ -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
);
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/adapter/channel/ReceiveMessageChannel.java
+6 -5
View File
@@ -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<ReceiveMessageRequest, Iterator<ReceiveMessageResponse>> {
public ReceiveMessageChannel(ReceiveMessageResponseHandler handler) {
super(handler);
public class ReceiveMessageChannel extends InvocationChannel<ReceiveMessageRequest, List<MessageExt>> {
public ReceiveMessageChannel(Context ctx, ReceiveMessageResponseHandler handler) {
super(ctx, handler);
}
}
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/adapter/channel/SendMessageChannel.java
+3 -2
View File
@@ -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<SendMessageRequest, SendMessageResponse> {
public SendMessageChannel(SendMessageResponseHandler handler) {
super(handler);
public SendMessageChannel(Context context, SendMessageResponseHandler handler) {
super(context, handler);
}
}
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/adapter/handler/ReceiveMessageResponseHandler.java
+8 -23
View File
@@ -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<ReceiveMessageRequest, Iterator<ReceiveMessageResponse>> {
public class ReceiveMessageResponseHandler implements ResponseHandler<ReceiveMessageRequest, List<MessageExt>> {
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<ReceiveMes
@Override
public void handle(RemotingCommand responseCommand,
InvocationContext<ReceiveMessageRequest, Iterator<ReceiveMessageResponse>> context) {
CompletableFuture<Iterator<ReceiveMessageResponse>> future = context.getResponse();
InvocationContext<ReceiveMessageRequest, List<MessageExt>> context) {
CompletableFuture<List<MessageExt>> future = context.getResponse();
long currentTimeInMillis = System.currentTimeMillis();
long popCosts = currentTimeInMillis - context.getTimestamp();
try {
Stopwatch stopWatch = Stopwatch.createStarted();
PopMessageResponseHeader responseHeader = (PopMessageResponseHeader) responseCommand.readCustomHeader();
List<ReceiveMessageResponse> responseList = new ArrayList<>();
List<MessageExt> 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<MessageExt> msgFoundList = MessageDecoder.decodes(byteBuffer);
@@ -130,29 +128,16 @@ public class ReceiveMessageResponseHandler implements ResponseHandler<ReceiveMes
}
}
}
Resource topic = context.getRequest()
.getMessageQueue()
.getTopic();
String topicName = GrpcConverter.wrapResourceWithNamespace(topic);
messageExt.setTopic(topicName);
messageExt.setBrokerName(brokerName);
messageExt.getProperties().computeIfAbsent(MessageConst.PROPERTY_FIRST_POP_TIME,
k -> 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);
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/service/LocalGrpcService.java
+53 -33
View File
@@ -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<ReceiveMessageRequest, ReceiveMessageResponse> 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<SendMessageResponse> future = new CompletableFuture<>();
InvocationContext<SendMessageRequest, SendMessageResponse> 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<List<MessageExt>> future = new CompletableFuture<>();
InvocationContext<ReceiveMessageRequest, List<MessageExt>> 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<Iterator<ReceiveMessageResponse>> future = new CompletableFuture<>();
InvocationContext<ReceiveMessageRequest, Iterator<ReceiveMessageResponse>> 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<AckMessageResponse> ackMessage(Context ctx, AckMessageRequest request) {
Channel channel = channelManager.createChannel();
Channel channel = channelManager.createChannel(ctx);
SimpleChannelHandlerContext channelHandlerContext = new SimpleChannelHandlerContext(channel);
CompletableFuture<AckMessageResponse> future = new CompletableFuture<>();
List<AckMessageResultEntry> 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<NackMessageResponse> nackMessage(Context ctx, NackMessageRequest request) {
Channel channel = channelManager.createChannel();
Channel channel = channelManager.createChannel(ctx);
SimpleChannelHandlerContext channelHandlerContext = new SimpleChannelHandlerContext(channel);
CompletableFuture<NackMessageResponse> future = new CompletableFuture<>();
@@ -385,7 +405,7 @@ public class LocalGrpcService extends AbstractStartAndShutdown implements GrpcFo
@Override
public CompletableFuture<ForwardMessageToDeadLetterQueueResponse> 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<EndTransactionResponse> 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<NotifyClientTerminationResponse> 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<ChangeInvisibleDurationResponse> 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: {
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/service/ReceiveMessageResponseStreamObserver.java
+67
View File
@@ -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<ReceiveMessageResponse> {
private final Context context;
private final ReceiveMessageRequest request;
private final ResponseHook<ReceiveMessageRequest, ReceiveMessageResponse> receiveMessageHook;
private final StreamObserver<ReceiveMessageResponse> observer;
public ReceiveMessageResponseStreamObserver(Context context, ReceiveMessageRequest request,
ResponseHook<ReceiveMessageRequest, ReceiveMessageResponse> receiveMessageHook,
StreamObserver<ReceiveMessageResponse> 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);
}
}
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/ReceiveMessageResponseStreamWriter.java → proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/service/ReceiveMessageResponseStreamWriter.java
+5 -3
View File
@@ -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<MessageExt> messageFoundList);
public abstract void write(Context ctx, ReceiveMessageRequest request, Throwable throwable);
}
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/ReceiveMessageResultFilter.java → proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/service/ReceiveMessageResultFilter.java
+1 -1
View File
@@ -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;
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/ConsumerService.java
+2 -1
View File
@@ -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;
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/DefaultReceiveMessageResponseStreamWriter.java
+6 -46
View File
@@ -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<MessageExt> messageFoundList) {
ReceiveMessageResponseStreamObserver responseStreamObserver = new ReceiveMessageResponseStreamObserver(
ctx,
request,
@@ -72,7 +74,7 @@ public class DefaultReceiveMessageResponseStreamWriter extends ReceiveMessageRes
try {
switch (status) {
case FOUND:
List<Message> messageList = this.receiveMessageResultFilter.filterMessage(ctx, request, result.getMsgFoundList());
List<Message> 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<ReceiveMessageResponse> {
private final Context context;
private final ReceiveMessageRequest request;
private final ResponseHook<ReceiveMessageRequest, ReceiveMessageResponse> receiveMessageHook;
private final StreamObserver<ReceiveMessageResponse> observer;
public ReceiveMessageResponseStreamObserver(Context context, ReceiveMessageRequest request,
ResponseHook<ReceiveMessageRequest, ReceiveMessageResponse> receiveMessageHook,
StreamObserver<ReceiveMessageResponse> 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;
}
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/DefaultReceiveMessageResultFilter.java
+8 -19
View File
@@ -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<AckResult> 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);
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/service/local/LocalReceiveMessageResponseStreamWriter.java
+143
View File
@@ -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<ReceiveMessageResponse> observer,
ResponseHook<ReceiveMessageRequest, ReceiveMessageResponse> 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<MessageExt> messageFoundList) {
ReceiveMessageResponseStreamObserver responseStreamObserver = new ReceiveMessageResponseStreamObserver(
ctx,
request,
receiveMessageHook,
streamObserver);
try {
switch (status) {
case FOUND:
List<Message> 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<Message> 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);
}
}
}
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/service/local/LocalReceiveMessageResultFilter.java
+123
View File
@@ -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<Message> filterMessage(Context ctx, ReceiveMessageRequest request, List<MessageExt> 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<Message> 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);
}
}
}
proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/service/LocalGrpcServiceTest.java
+1 -1
View File
@@ -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()
test/src/test/java/org/apache/rocketmq/test/grpc/v2/GrpcBaseTest.java
+5 -5
View File
@@ -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();