From 527a38cefe57788968b647bd11b6e726116fdee8 Mon Sep 17 00:00:00 2001 From: zhouxiang Date: Thu, 17 Mar 2022 19:13:23 +0800 Subject: [PATCH] [ISSUE #3949] Implement pull and stream --- .../adapter/channel/PullMessageChannel.java | 29 ++++++++++ .../handler/PullMessageResponseHandler.java | 56 ++++++++++++++++++ .../proxy/grpc/service/LocalGrpcService.java | 57 ++++++++++++++++++- 3 files changed, 140 insertions(+), 2 deletions(-) create mode 100644 proxy/src/main/java/org/apache/rocketmq/proxy/grpc/adapter/channel/PullMessageChannel.java create mode 100644 proxy/src/main/java/org/apache/rocketmq/proxy/grpc/adapter/handler/PullMessageResponseHandler.java diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/adapter/channel/PullMessageChannel.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/adapter/channel/PullMessageChannel.java new file mode 100644 index 0000000000..533bea1810 --- /dev/null +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/adapter/channel/PullMessageChannel.java @@ -0,0 +1,29 @@ +/* + * 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.adapter.channel; + +import apache.rocketmq.v1.PullMessageRequest; +import apache.rocketmq.v1.PullMessageResponse; +import org.apache.rocketmq.proxy.channel.InvocationChannel; +import org.apache.rocketmq.proxy.grpc.adapter.handler.PullMessageResponseHandler; + +public class PullMessageChannel extends InvocationChannel { + public PullMessageChannel(PullMessageResponseHandler handler) { + super(handler); + } +} diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/adapter/handler/PullMessageResponseHandler.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/adapter/handler/PullMessageResponseHandler.java new file mode 100644 index 0000000000..ce85e2e4fe --- /dev/null +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/adapter/handler/PullMessageResponseHandler.java @@ -0,0 +1,56 @@ +/* + * 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.adapter.handler; + +import apache.rocketmq.v1.PullMessageRequest; +import apache.rocketmq.v1.PullMessageResponse; +import java.nio.ByteBuffer; +import java.util.List; +import org.apache.rocketmq.common.message.MessageDecoder; +import org.apache.rocketmq.common.message.MessageExt; +import org.apache.rocketmq.common.protocol.ResponseCode; +import org.apache.rocketmq.common.protocol.header.PullMessageResponseHeader; +import org.apache.rocketmq.proxy.grpc.adapter.InvocationContext; +import org.apache.rocketmq.proxy.grpc.common.Converter; +import org.apache.rocketmq.proxy.grpc.common.ResponseBuilder; +import org.apache.rocketmq.remoting.protocol.RemotingCommand; + +public class PullMessageResponseHandler implements ResponseHandler { + @Override public void handle(RemotingCommand responseCommand, + InvocationContext context) { + try { + PullMessageResponseHeader responseHeader = (PullMessageResponseHeader) responseCommand.readCustomHeader(); + PullMessageResponse.Builder builder = PullMessageResponse.newBuilder(); + if (responseCommand.getCode() == ResponseCode.SUCCESS) { + ByteBuffer byteBuffer = ByteBuffer.wrap(responseCommand.getBody()); + List msgFoundList = MessageDecoder.decodes(byteBuffer); + for (MessageExt messageExt : msgFoundList) { + builder.addMessages(Converter.buildMessage(messageExt)); + } + } + PullMessageResponse response = builder.setCommon(ResponseBuilder.buildCommon(responseCommand.getCode(), responseCommand.getRemark())) + .setMinOffset(responseHeader.getMinOffset()) + .setNextOffset(responseHeader.getNextBeginOffset()) + .setMaxOffset(responseHeader.getMaxOffset()) + .build(); + context.getResponse().complete(response); + } catch (Exception e) { + context.getResponse().completeExceptionally(e); + } + } +} diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/service/LocalGrpcService.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/service/LocalGrpcService.java index 63340febff..a13e7c517c 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/service/LocalGrpcService.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/service/LocalGrpcService.java @@ -35,12 +35,14 @@ import apache.rocketmq.v1.NackMessageResponse; import apache.rocketmq.v1.NoopCommand; import apache.rocketmq.v1.NotifyClientTerminationRequest; import apache.rocketmq.v1.NotifyClientTerminationResponse; +import apache.rocketmq.v1.Partition; import apache.rocketmq.v1.PollCommandRequest; import apache.rocketmq.v1.PollCommandResponse; import apache.rocketmq.v1.PullMessageRequest; import apache.rocketmq.v1.PullMessageResponse; import apache.rocketmq.v1.QueryAssignmentRequest; import apache.rocketmq.v1.QueryAssignmentResponse; +import apache.rocketmq.v1.QueryOffsetPolicy; import apache.rocketmq.v1.QueryOffsetRequest; import apache.rocketmq.v1.QueryOffsetResponse; import apache.rocketmq.v1.QueryRouteRequest; @@ -54,6 +56,7 @@ import apache.rocketmq.v1.ReportThreadStackTraceResponse; import apache.rocketmq.v1.Resource; import apache.rocketmq.v1.SendMessageRequest; import apache.rocketmq.v1.SendMessageResponse; +import com.google.protobuf.util.Timestamps; import com.google.rpc.Code; import io.grpc.Context; import io.netty.channel.Channel; @@ -76,6 +79,7 @@ import org.apache.rocketmq.common.protocol.header.ChangeInvisibleTimeResponseHea import org.apache.rocketmq.common.protocol.header.ConsumerSendMsgBackRequestHeader; import org.apache.rocketmq.common.protocol.header.EndTransactionRequestHeader; import org.apache.rocketmq.common.protocol.header.PopMessageRequestHeader; +import org.apache.rocketmq.common.protocol.header.PullMessageRequestHeader; import org.apache.rocketmq.common.protocol.header.SendMessageRequestHeader; import org.apache.rocketmq.common.protocol.header.UnregisterClientRequestHeader; import org.apache.rocketmq.common.protocol.heartbeat.HeartbeatData; @@ -88,6 +92,8 @@ import org.apache.rocketmq.proxy.grpc.adapter.InvocationContext; import org.apache.rocketmq.proxy.grpc.adapter.channel.GrpcClientChannel; import org.apache.rocketmq.proxy.grpc.adapter.channel.ReceiveMessageChannel; import org.apache.rocketmq.proxy.grpc.adapter.channel.SendMessageChannel; +import org.apache.rocketmq.proxy.grpc.adapter.channel.PullMessageChannel; +import org.apache.rocketmq.proxy.grpc.adapter.handler.PullMessageResponseHandler; import org.apache.rocketmq.proxy.grpc.adapter.handler.ReceiveMessageResponseHandler; import org.apache.rocketmq.proxy.grpc.adapter.handler.SendMessageResponseHandler; import org.apache.rocketmq.proxy.grpc.common.Converter; @@ -352,11 +358,58 @@ public class LocalGrpcService implements GrpcForwardService { } @Override public CompletableFuture queryOffset(Context ctx, QueryOffsetRequest request) { - return null; + Partition partition = request.getPartition(); + String topicName = Converter.getResourceNameWithNamespace(partition.getTopic()); + int queueId = partition.getId(); + + long offset; + if (request.getPolicy() == QueryOffsetPolicy.BEGINNING) { + offset = 0L; + } else if (request.getPolicy() == QueryOffsetPolicy.END) { + offset = brokerController.getMessageStore() + .getMaxOffsetInQueue(topicName, queueId); + } else { + long timestamp = Timestamps.toMillis(request.getTimePoint()); + offset = brokerController.getMessageStore() + .getOffsetInQueueByTime(topicName, queueId, timestamp); + } + return CompletableFuture.completedFuture(QueryOffsetResponse.newBuilder() + .setCommon(ResponseBuilder.buildCommon(Code.OK, "ok")) + .setOffset(offset) + .build()); } @Override public CompletableFuture pullMessage(Context ctx, PullMessageRequest request) { - return null; + long timeRemaining = Context.current() + .getDeadline() + .timeRemaining(TimeUnit.MILLISECONDS); + long pollTime = timeRemaining - ConfigurationManager.getProxyConfig().getLongPollingReserveTimeInMillis(); + if (pollTime <= 0) { + pollTime = timeRemaining; + } + PullMessageRequestHeader requestHeader = Converter.buildPullMessageRequestHeader(request, pollTime); + RemotingCommand command = RemotingCommand.createRequestCommand(RequestCode.PULL_MESSAGE, requestHeader); + command.makeCustomHeaderToNet(); + + PullMessageResponseHandler handler = new PullMessageResponseHandler(); + PullMessageChannel channel = channelManager.createChannel(() -> new PullMessageChannel(handler), PullMessageChannel.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.getPullMessageProcessor() + .processRequest(channelHandlerContext, command); + if (response != null) { + handler.handle(response, context); + channel.eraseInvocationContext(command.getOpaque()); + } + } catch (Exception e) { + LOGGER.error("Failed to process pull message command", e); + channel.eraseInvocationContext(command.getOpaque()); + future.completeExceptionally(e); + } + return future; } @Override public CompletableFuture pollCommand(Context ctx, PollCommandRequest request) {