diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/service/message/LocalMessageService.java b/proxy/src/main/java/org/apache/rocketmq/proxy/service/message/LocalMessageService.java index ca7dcc9eb0c..aaa688fee64 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/service/message/LocalMessageService.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/service/message/LocalMessageService.java @@ -104,7 +104,7 @@ public CompletableFuture> sendMessage(ProxyContext ctx, Address body = message.getBody(); messageId = MessageClientIDSetter.getUniqID(message); } - RemotingCommand request = LocalRemotingCommand.createRequestCommand(RequestCode.SEND_MESSAGE, requestHeader); + RemotingCommand request = LocalRemotingCommand.createRequestCommand(RequestCode.SEND_MESSAGE, requestHeader, ctx.getLanguage()); request.setBody(body); CompletableFuture future = new CompletableFuture<>(); SimpleChannel channel = channelManager.createInvocationChannel(ctx); @@ -162,7 +162,7 @@ public CompletableFuture sendMessageBack(ProxyContext ctx, Rece ConsumerSendMsgBackRequestHeader requestHeader, long timeoutMillis) { SimpleChannel channel = channelManager.createChannel(ctx); ChannelHandlerContext channelHandlerContext = channel.getChannelHandlerContext(); - RemotingCommand command = LocalRemotingCommand.createRequestCommand(RequestCode.CONSUMER_SEND_MSG_BACK, requestHeader); + RemotingCommand command = LocalRemotingCommand.createRequestCommand(RequestCode.CONSUMER_SEND_MSG_BACK, requestHeader, ctx.getLanguage()); CompletableFuture future = new CompletableFuture<>(); try { RemotingCommand response = brokerController.getSendMessageProcessor() @@ -181,7 +181,7 @@ public CompletableFuture endTransactionOneway(ProxyContext ctx, String bro CompletableFuture future = new CompletableFuture<>(); SimpleChannel channel = channelManager.createChannel(ctx); ChannelHandlerContext channelHandlerContext = channel.getChannelHandlerContext(); - RemotingCommand command = LocalRemotingCommand.createRequestCommand(RequestCode.END_TRANSACTION, requestHeader); + RemotingCommand command = LocalRemotingCommand.createRequestCommand(RequestCode.END_TRANSACTION, requestHeader, ctx.getLanguage()); try { brokerController.getEndTransactionProcessor() .processRequest(channelHandlerContext, command); @@ -196,7 +196,7 @@ public CompletableFuture endTransactionOneway(ProxyContext ctx, String bro public CompletableFuture popMessage(ProxyContext ctx, AddressableMessageQueue messageQueue, PopMessageRequestHeader requestHeader, long timeoutMillis) { requestHeader.setBornTime(System.currentTimeMillis()); - RemotingCommand request = LocalRemotingCommand.createRequestCommand(RequestCode.POP_MESSAGE, requestHeader); + RemotingCommand request = LocalRemotingCommand.createRequestCommand(RequestCode.POP_MESSAGE, requestHeader, ctx.getLanguage()); CompletableFuture future = new CompletableFuture<>(); SimpleChannel channel = channelManager.createInvocationChannel(ctx); InvocationContext invocationContext = new InvocationContext(future); @@ -307,7 +307,7 @@ public CompletableFuture changeInvisibleTime(ProxyContext ctx, Receip ChangeInvisibleTimeRequestHeader requestHeader, long timeoutMillis) { SimpleChannel channel = channelManager.createChannel(ctx); ChannelHandlerContext channelHandlerContext = channel.getChannelHandlerContext(); - RemotingCommand command = LocalRemotingCommand.createRequestCommand(RequestCode.CHANGE_MESSAGE_INVISIBLETIME, requestHeader); + RemotingCommand command = LocalRemotingCommand.createRequestCommand(RequestCode.CHANGE_MESSAGE_INVISIBLETIME, requestHeader, ctx.getLanguage()); CompletableFuture future = new CompletableFuture<>(); try { RemotingCommand response = brokerController.getChangeInvisibleTimeProcessor() @@ -346,7 +346,7 @@ public CompletableFuture ackMessage(ProxyContext ctx, ReceiptHandle h AckMessageRequestHeader requestHeader, long timeoutMillis) { SimpleChannel channel = channelManager.createChannel(ctx); ChannelHandlerContext channelHandlerContext = channel.getChannelHandlerContext(); - RemotingCommand command = LocalRemotingCommand.createRequestCommand(RequestCode.ACK_MESSAGE, requestHeader); + RemotingCommand command = LocalRemotingCommand.createRequestCommand(RequestCode.ACK_MESSAGE, requestHeader, ctx.getLanguage()); CompletableFuture future = new CompletableFuture<>(); try { RemotingCommand response = brokerController.getAckMessageProcessor() diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/service/message/LocalRemotingCommand.java b/proxy/src/main/java/org/apache/rocketmq/proxy/service/message/LocalRemotingCommand.java index 73048dbbc24..915cafcd579 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/service/message/LocalRemotingCommand.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/service/message/LocalRemotingCommand.java @@ -16,16 +16,19 @@ */ package org.apache.rocketmq.proxy.service.message; -import java.util.HashMap; import org.apache.rocketmq.remoting.CommandCustomHeader; import org.apache.rocketmq.remoting.exception.RemotingCommandException; +import org.apache.rocketmq.remoting.protocol.LanguageCode; import org.apache.rocketmq.remoting.protocol.RemotingCommand; +import java.util.HashMap; + public class LocalRemotingCommand extends RemotingCommand { - public static LocalRemotingCommand createRequestCommand(int code, CommandCustomHeader customHeader) { + public static LocalRemotingCommand createRequestCommand(int code, CommandCustomHeader customHeader, String language) { LocalRemotingCommand cmd = new LocalRemotingCommand(); cmd.setCode(code); + cmd.setLanguage(LanguageCode.getCode(language)); cmd.writeCustomHeader(customHeader); cmd.setExtFields(new HashMap<>()); setCmdVersion(cmd); @@ -37,4 +40,5 @@ public CommandCustomHeader decodeCommandCustomHeader( Class classHeader) throws RemotingCommandException { return classHeader.cast(readCustomHeader()); } + } diff --git a/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/LanguageCode.java b/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/LanguageCode.java index 19280f99672..2df9fbf0278 100644 --- a/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/LanguageCode.java +++ b/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/LanguageCode.java @@ -17,6 +17,11 @@ package org.apache.rocketmq.remoting.protocol; +import java.util.Arrays; +import java.util.Map; +import java.util.function.Function; +import java.util.stream.Collectors; + public enum LanguageCode { JAVA((byte) 0), CPP((byte) 1), @@ -50,4 +55,10 @@ public static LanguageCode valueOf(byte code) { public byte getCode() { return code; } + + private static final Map MAP = Arrays.stream(LanguageCode.values()).collect(Collectors.toMap(LanguageCode::name, Function.identity())); + + public static LanguageCode getCode(String language) { + return MAP.get(language); + } }