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

[ISSUE #3949] Implement pull and stream

Browse Source
This commit is contained in:
zhouxiang
2022-07-13 11:29:12 +08:00
parent 153b2ddd41
commit 527a38cefe
3 changed files with 140 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/adapter/channel/PullMessageChannel.java
+29
View File
@@ -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<PullMessageRequest, PullMessageResponse> {
public PullMessageChannel(PullMessageResponseHandler handler) {
super(handler);
}
}
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/adapter/handler/PullMessageResponseHandler.java
+56
View File
@@ -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<PullMessageRequest, PullMessageResponse> {
@Override public void handle(RemotingCommand responseCommand,
InvocationContext<PullMessageRequest, PullMessageResponse> context) {
try {
PullMessageResponseHeader responseHeader = (PullMessageResponseHeader) responseCommand.readCustomHeader();
PullMessageResponse.Builder builder = PullMessageResponse.newBuilder();
if (responseCommand.getCode() == ResponseCode.SUCCESS) {
ByteBuffer byteBuffer = ByteBuffer.wrap(responseCommand.getBody());
List<MessageExt> 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);
}
}
}
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/service/LocalGrpcService.java
+55 -2
View File
@@ -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<QueryOffsetResponse> 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<PullMessageResponse> 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<PullMessageResponse> future = new CompletableFuture<>();
InvocationContext<PullMessageRequest, PullMessageResponse> 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<PollCommandResponse> pollCommand(Context ctx, PollCommandRequest request) {