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

[ISSUE #3949] Do some refactoring work.

Browse Source
This commit is contained in:
Jixiang.jjx
2022-07-13 11:29:15 +08:00
committed by zhouxiang
parent ff3117f322
commit 9cf17ebe49
6 changed files with 24 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIExt.java
+15 -14
View File
@@ -180,9 +180,10 @@ public class MQClientAPIExt {
ConsumerSendMsgBackRequestHeader requestHeader,
long timeoutMillis
) {
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.CONSUMER_SEND_MSG_BACK, requestHeader);
CompletableFuture<RemotingCommand> future = new CompletableFuture<>();
try {
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.CONSUMER_SEND_MSG_BACK, requestHeader);
this.getRemotingClient().invokeAsync(brokerAddr, request, timeoutMillis, responseFuture -> {
RemotingCommand response = responseFuture.getResponseCommand();
if (response != null) {
@@ -357,19 +358,19 @@ public class MQClientAPIExt {
}
public CompletableFuture<Long> getMaxOffset(String brokerAddr, String topic, int queueId, long timeoutMillis) {
GetMaxOffsetRequestHeader requestHeader = new GetMaxOffsetRequestHeader();
requestHeader.setTopic(topic);
requestHeader.setQueueId(queueId);
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_MAX_OFFSET, requestHeader);
CompletableFuture<Long> future = new CompletableFuture<>();
try {
GetMaxOffsetRequestHeader requestHeader = new GetMaxOffsetRequestHeader();
requestHeader.setTopic(topic);
requestHeader.setQueueId(queueId);
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_MAX_OFFSET, requestHeader);
this.getRemotingClient().invokeAsync(brokerAddr, request, timeoutMillis, responseFuture -> {
RemotingCommand response = responseFuture.getResponseCommand();
if (response != null) {
if (ResponseCode.SUCCESS == response.getCode()) {
try {
GetMaxOffsetResponseHeader responseHeader =
(GetMaxOffsetResponseHeader) response.decodeCommandCustomHeader(GetMaxOffsetResponseHeader.class);
GetMaxOffsetResponseHeader responseHeader = response.decodeCommandCustomHeader(GetMaxOffsetResponseHeader.class);
future.complete(responseHeader.getOffset());
} catch (Throwable t) {
future.completeExceptionally(t);
@@ -387,20 +388,20 @@ public class MQClientAPIExt {
}
public CompletableFuture<Long> searchOffset(String brokerAddr, String topic, int queueId , long timestamp, long timeoutMillis) {
SearchOffsetRequestHeader requestHeader = new SearchOffsetRequestHeader();
requestHeader.setTopic(topic);
requestHeader.setQueueId(queueId);
requestHeader.setTimestamp(timestamp);
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.SEARCH_OFFSET_BY_TIMESTAMP, requestHeader);
CompletableFuture<Long> future = new CompletableFuture<>();
try {
SearchOffsetRequestHeader requestHeader = new SearchOffsetRequestHeader();
requestHeader.setTopic(topic);
requestHeader.setQueueId(queueId);
requestHeader.setTimestamp(timestamp);
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.SEARCH_OFFSET_BY_TIMESTAMP, requestHeader);
this.getRemotingClient().invokeAsync(brokerAddr, request, timeoutMillis, responseFuture -> {
RemotingCommand response = responseFuture.getResponseCommand();
if (response != null) {
if (response.getCode() == ResponseCode.SUCCESS) {
try {
SearchOffsetResponseHeader responseHeader =
(SearchOffsetResponseHeader) response.decodeCommandCustomHeader(SearchOffsetResponseHeader.class);
SearchOffsetResponseHeader responseHeader = response.decodeCommandCustomHeader(SearchOffsetResponseHeader.class);
future.complete(responseHeader.getOffset());
} catch (Throwable t) {
future.completeExceptionally(t);
proxy/src/main/java/org/apache/rocketmq/proxy/connector/ForwardProducer.java
-1
View File
@@ -32,7 +32,6 @@ import org.apache.rocketmq.proxy.connector.transaction.TransactionId;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
public class ForwardProducer extends AbstractForwardClient {
private static final String PID_PREFIX = "PID_RMQ_PROXY_PUBLISH_MESSAGE_";
public ForwardProducer(ForwardClientManager clientFactory) {
proxy/src/main/java/org/apache/rocketmq/proxy/connector/transaction/TransactionId.java
+7 -2
View File
@@ -42,8 +42,13 @@ public class TransactionId {
private long tranStateTableOffset;
private String proxyTransactionId;
public TransactionId(SocketAddress brokerAddr, String brokerTransactionId, long commitLogOffset,
long tranStateTableOffset, String proxyTransactionId) {
public TransactionId(
SocketAddress brokerAddr,
String brokerTransactionId,
long commitLogOffset,
long tranStateTableOffset,
String proxyTransactionId
) {
this.brokerAddr = brokerAddr;
this.brokerTransactionId = brokerTransactionId;
this.commitLogOffset = commitLogOffset;
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/adapter/channel/ChannelType.java
+1 -1
View File
@@ -23,7 +23,7 @@ public enum ChannelType {
*/
LOCAL,
/**
* The channel sync from other proxy
* The channel synced from other proxy
*/
REMOTE
}
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/adapter/channel/GrpcClientChannel.java
+1 -2
View File
@@ -120,8 +120,7 @@ public class GrpcClientChannel extends SimpleChannel {
try {
switch (command.getCode()) {
case RequestCode.CHECK_TRANSACTION_STATE: {
final CheckTransactionStateRequestHeader requestHeader =
(CheckTransactionStateRequestHeader) command.decodeCommandCustomHeader(CheckTransactionStateRequestHeader.class);
final CheckTransactionStateRequestHeader requestHeader = command.decodeCommandCustomHeader(CheckTransactionStateRequestHeader.class);
MessageExt messageExt = MessageDecoder.decode(ByteBuffer.wrap(command.getBody()), true, false, false);
future.complete(PollCommandResponse.newBuilder()
.setRecoverOrphanedTransactionCommand(RecoverOrphanedTransactionCommand.newBuilder()
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/service/cluster/PullMessageService.java
-1
View File
@@ -41,7 +41,6 @@ import org.apache.rocketmq.proxy.grpc.adapter.ResponseBuilder;
import org.apache.rocketmq.proxy.grpc.adapter.ResponseHook;
public class PullMessageService extends BaseService {
private final DefaultForwardClient forwardClient;
private final ForwardReadConsumer readConsumer;