mirror of
https://github.com/apache/rocketmq.git
synced 2026-10-29 17:25:35 +08:00
[ISSUE 3585] [Part B] Improve encode/decode performance (#3588)
* [ISSUE 3585] [Part B] Improve header encode/decode performance * [ISSUE 3585] [Part B] Improve RocketMQSerializable performance with zero-copy; fix securtiy problems. * [ISSUE 3585] [Part B] Fix unit test Co-authored-by: yuz10 <845238369@qq.com>
This commit is contained in:
huangli
+3
-88
@@ -19,7 +19,6 @@ package org.apache.rocketmq.broker.processor;
|
||||
import io.netty.channel.ChannelHandlerContext;
|
||||
import java.net.InetSocketAddress;
|
||||
import java.net.SocketAddress;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.ThreadLocalRandom;
|
||||
@@ -237,19 +236,6 @@ public abstract class AbstractSendMessageProcessor extends AsyncNettyRequestProc
|
||||
this.sendMessageHookList = sendMessageHookList;
|
||||
}
|
||||
|
||||
protected void doResponse(ChannelHandlerContext ctx, RemotingCommand request,
|
||||
final RemotingCommand response) {
|
||||
if (!request.isOnewayRPC()) {
|
||||
try {
|
||||
ctx.writeAndFlush(response);
|
||||
} catch (Throwable e) {
|
||||
log.error("SendMessageProcessor process request over, but response failed", e);
|
||||
log.error(request.toString());
|
||||
log.error(response.toString());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public void executeSendMessageHookBefore(final ChannelHandlerContext ctx, final RemotingCommand request,
|
||||
SendMessageContext context) {
|
||||
if (hasSendMessageHook()) {
|
||||
@@ -288,7 +274,9 @@ public abstract class AbstractSendMessageProcessor extends AsyncNettyRequestProc
|
||||
switch (request.getCode()) {
|
||||
case RequestCode.SEND_BATCH_MESSAGE:
|
||||
case RequestCode.SEND_MESSAGE_V2:
|
||||
requestHeaderV2 = decodeSendMessageHeaderV2(request);
|
||||
requestHeaderV2 =
|
||||
(SendMessageRequestHeaderV2) request
|
||||
.decodeCommandCustomHeader(SendMessageRequestHeaderV2.class);
|
||||
case RequestCode.SEND_MESSAGE:
|
||||
if (null == requestHeaderV2) {
|
||||
requestHeader =
|
||||
@@ -303,79 +291,6 @@ public abstract class AbstractSendMessageProcessor extends AsyncNettyRequestProc
|
||||
return requestHeader;
|
||||
}
|
||||
|
||||
static SendMessageRequestHeaderV2 decodeSendMessageHeaderV2(RemotingCommand request)
|
||||
throws RemotingCommandException {
|
||||
SendMessageRequestHeaderV2 r = new SendMessageRequestHeaderV2();
|
||||
HashMap<String, String> fields = request.getExtFields();
|
||||
if (fields == null) {
|
||||
throw new RemotingCommandException("the ext fields is null");
|
||||
}
|
||||
|
||||
String s = fields.get("a");
|
||||
checkNotNull(s, "the custom field <a> is null");
|
||||
r.setA(s);
|
||||
|
||||
s = fields.get("b");
|
||||
checkNotNull(s, "the custom field <b> is null");
|
||||
r.setB(s);
|
||||
|
||||
s = fields.get("c");
|
||||
checkNotNull(s, "the custom field <c> is null");
|
||||
r.setC(s);
|
||||
|
||||
s = fields.get("d");
|
||||
checkNotNull(s, "the custom field <d> is null");
|
||||
r.setD(Integer.parseInt(s));
|
||||
|
||||
s = fields.get("e");
|
||||
checkNotNull(s, "the custom field <e> is null");
|
||||
r.setE(Integer.parseInt(s));
|
||||
|
||||
s = fields.get("f");
|
||||
checkNotNull(s, "the custom field <f> is null");
|
||||
r.setF(Integer.parseInt(s));
|
||||
|
||||
s = fields.get("g");
|
||||
checkNotNull(s, "the custom field <g> is null");
|
||||
r.setG(Long.parseLong(s));
|
||||
|
||||
s = fields.get("h");
|
||||
checkNotNull(s, "the custom field <h> is null");
|
||||
r.setH(Integer.parseInt(s));
|
||||
|
||||
s = fields.get("i");
|
||||
if (s != null) {
|
||||
r.setI(s);
|
||||
}
|
||||
|
||||
s = fields.get("j");
|
||||
if (s != null) {
|
||||
r.setJ(Integer.parseInt(s));
|
||||
}
|
||||
|
||||
s = fields.get("k");
|
||||
if (s != null) {
|
||||
r.setK(Boolean.parseBoolean(s));
|
||||
}
|
||||
|
||||
s = fields.get("l");
|
||||
if (s != null) {
|
||||
r.setL(Integer.parseInt(s));
|
||||
}
|
||||
|
||||
s = fields.get("m");
|
||||
if (s != null) {
|
||||
r.setM(Boolean.parseBoolean(s));
|
||||
}
|
||||
return r;
|
||||
}
|
||||
|
||||
private static void checkNotNull(String s, String msg) throws RemotingCommandException {
|
||||
if (s == null) {
|
||||
throw new RemotingCommandException(msg);
|
||||
}
|
||||
}
|
||||
|
||||
public void executeSendMessageHookAfter(final RemotingCommand response, final SendMessageContext context) {
|
||||
if (hasSendMessageHook()) {
|
||||
for (SendMessageHook hook : this.sendMessageHookList) {
|
||||
|
||||
@@ -547,8 +547,6 @@ public class SendMessageProcessor extends AbstractSendMessageProcessor implement
|
||||
responseHeader.setQueueId(queueIdInt);
|
||||
responseHeader.setQueueOffset(putMessageResult.getAppendMessageResult().getLogicsOffset());
|
||||
|
||||
doResponse(ctx, request, response);
|
||||
|
||||
if (hasSendMessageHook()) {
|
||||
sendMessageContext.setMsgId(responseHeader.getMsgId());
|
||||
sendMessageContext.setQueueId(responseHeader.getQueueId());
|
||||
@@ -563,7 +561,7 @@ public class SendMessageProcessor extends AbstractSendMessageProcessor implement
|
||||
sendMessageContext.setCommercialSendSize(wroteSize);
|
||||
sendMessageContext.setCommercialOwner(owner);
|
||||
}
|
||||
return null;
|
||||
return response;
|
||||
} else {
|
||||
if (hasSendMessageHook()) {
|
||||
int wroteSize = request.getBody().length;
|
||||
|
||||
+1
-14
@@ -207,13 +207,7 @@ public class SendMessageProcessorTest {
|
||||
.thenReturn(CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.PUT_OK, new AppendMessageResult(AppendMessageStatus.PUT_OK))));
|
||||
RemotingCommand request = createSendTransactionMsgCommand(RequestCode.SEND_MESSAGE);
|
||||
final RemotingCommand[] response = new RemotingCommand[1];
|
||||
doAnswer(new Answer() {
|
||||
@Override
|
||||
public Object answer(InvocationOnMock invocation) throws Throwable {
|
||||
response[0] = invocation.getArgument(0);
|
||||
return null;
|
||||
}
|
||||
}).when(handlerContext).writeAndFlush(any(Object.class));
|
||||
|
||||
RemotingCommand responseToReturn = sendMessageProcessor.processRequest(handlerContext, request);
|
||||
if (responseToReturn != null) {
|
||||
assertThat(response[0]).isNull();
|
||||
@@ -275,13 +269,6 @@ public class SendMessageProcessorTest {
|
||||
private void assertPutResult(int responseCode) throws RemotingCommandException {
|
||||
final RemotingCommand request = createSendMsgCommand(RequestCode.SEND_MESSAGE);
|
||||
final RemotingCommand[] response = new RemotingCommand[1];
|
||||
doAnswer(new Answer() {
|
||||
@Override
|
||||
public Object answer(InvocationOnMock invocation) throws Throwable {
|
||||
response[0] = invocation.getArgument(0);
|
||||
return null;
|
||||
}
|
||||
}).when(handlerContext).writeAndFlush(any(Object.class));
|
||||
RemotingCommand responseToReturn = sendMessageProcessor.processRequest(handlerContext, request);
|
||||
if (responseToReturn != null) {
|
||||
assertThat(response[0]).isNull();
|
||||
|
||||
+79
-1
@@ -20,12 +20,17 @@
|
||||
*/
|
||||
package org.apache.rocketmq.common.protocol.header;
|
||||
|
||||
import java.util.HashMap;
|
||||
|
||||
import org.apache.rocketmq.remoting.CommandCustomHeader;
|
||||
import org.apache.rocketmq.remoting.annotation.CFNotNull;
|
||||
import org.apache.rocketmq.remoting.annotation.CFNullable;
|
||||
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
|
||||
import org.apache.rocketmq.remoting.protocol.FastCodesHeader;
|
||||
|
||||
public class PullMessageRequestHeader implements CommandCustomHeader {
|
||||
import io.netty.buffer.ByteBuf;
|
||||
|
||||
public class PullMessageRequestHeader implements CommandCustomHeader, FastCodesHeader {
|
||||
@CFNotNull
|
||||
private String consumerGroup;
|
||||
@CFNotNull
|
||||
@@ -52,6 +57,79 @@ public class PullMessageRequestHeader implements CommandCustomHeader {
|
||||
public void checkFields() throws RemotingCommandException {
|
||||
}
|
||||
|
||||
@Override
|
||||
public void encode(ByteBuf out) {
|
||||
writeIfNotNull(out, "consumerGroup", consumerGroup);
|
||||
writeIfNotNull(out, "topic", topic);
|
||||
writeIfNotNull(out, "queueId", queueId);
|
||||
writeIfNotNull(out, "queueOffset", queueOffset);
|
||||
writeIfNotNull(out, "maxMsgNums", maxMsgNums);
|
||||
writeIfNotNull(out, "sysFlag", sysFlag);
|
||||
writeIfNotNull(out, "commitOffset", commitOffset);
|
||||
writeIfNotNull(out, "suspendTimeoutMillis", suspendTimeoutMillis);
|
||||
writeIfNotNull(out, "subscription", subscription);
|
||||
writeIfNotNull(out, "subVersion", subVersion);
|
||||
writeIfNotNull(out, "expressionType", expressionType);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void decode(HashMap<String, String> fields) throws RemotingCommandException {
|
||||
String str = getAndCheckNotNull(fields, "consumerGroup");
|
||||
if (str != null) {
|
||||
this.consumerGroup = str;
|
||||
}
|
||||
|
||||
str = getAndCheckNotNull(fields, "topic");
|
||||
if (str != null) {
|
||||
this.topic = str;
|
||||
}
|
||||
|
||||
str = getAndCheckNotNull(fields, "queueId");
|
||||
if (str != null) {
|
||||
this.queueId = Integer.parseInt(str);
|
||||
}
|
||||
|
||||
str = getAndCheckNotNull(fields, "queueOffset");
|
||||
if (str != null) {
|
||||
this.queueOffset = Long.parseLong(str);
|
||||
}
|
||||
|
||||
str = getAndCheckNotNull(fields, "maxMsgNums");
|
||||
if (str != null) {
|
||||
this.maxMsgNums = Integer.parseInt(str);
|
||||
}
|
||||
|
||||
str = getAndCheckNotNull(fields, "sysFlag");
|
||||
if (str != null) {
|
||||
this.sysFlag = Integer.parseInt(str);
|
||||
}
|
||||
|
||||
str = getAndCheckNotNull(fields, "commitOffset");
|
||||
if (str != null) {
|
||||
this.commitOffset = Long.parseLong(str);
|
||||
}
|
||||
|
||||
str = getAndCheckNotNull(fields, "suspendTimeoutMillis");
|
||||
if (str != null) {
|
||||
this.suspendTimeoutMillis = Long.parseLong(str);
|
||||
}
|
||||
|
||||
str = fields.get("subscription");
|
||||
if (str != null) {
|
||||
this.subscription = str;
|
||||
}
|
||||
|
||||
str = getAndCheckNotNull(fields, "subVersion");
|
||||
if (str != null) {
|
||||
this.subVersion = Long.parseLong(str);
|
||||
}
|
||||
|
||||
str = fields.get("expressionType");
|
||||
if (str != null) {
|
||||
this.expressionType = str;
|
||||
}
|
||||
}
|
||||
|
||||
public String getConsumerGroup() {
|
||||
return consumerGroup;
|
||||
}
|
||||
|
||||
+37
-1
@@ -20,11 +20,16 @@
|
||||
*/
|
||||
package org.apache.rocketmq.common.protocol.header;
|
||||
|
||||
import java.util.HashMap;
|
||||
|
||||
import org.apache.rocketmq.remoting.CommandCustomHeader;
|
||||
import org.apache.rocketmq.remoting.annotation.CFNotNull;
|
||||
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
|
||||
import org.apache.rocketmq.remoting.protocol.FastCodesHeader;
|
||||
|
||||
public class PullMessageResponseHeader implements CommandCustomHeader {
|
||||
import io.netty.buffer.ByteBuf;
|
||||
|
||||
public class PullMessageResponseHeader implements CommandCustomHeader, FastCodesHeader {
|
||||
@CFNotNull
|
||||
private Long suggestWhichBrokerId;
|
||||
@CFNotNull
|
||||
@@ -38,6 +43,37 @@ public class PullMessageResponseHeader implements CommandCustomHeader {
|
||||
public void checkFields() throws RemotingCommandException {
|
||||
}
|
||||
|
||||
@Override
|
||||
public void encode(ByteBuf out) {
|
||||
writeIfNotNull(out, "suggestWhichBrokerId", suggestWhichBrokerId);
|
||||
writeIfNotNull(out, "nextBeginOffset", nextBeginOffset);
|
||||
writeIfNotNull(out, "minOffset", minOffset);
|
||||
writeIfNotNull(out, "maxOffset", maxOffset);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void decode(HashMap<String, String> fields) throws RemotingCommandException {
|
||||
String str = getAndCheckNotNull(fields, "suggestWhichBrokerId");
|
||||
if (str != null) {
|
||||
this.suggestWhichBrokerId = Long.parseLong(str);
|
||||
}
|
||||
|
||||
str = getAndCheckNotNull(fields, "nextBeginOffset");
|
||||
if (str != null) {
|
||||
this.nextBeginOffset = Long.parseLong(str);
|
||||
}
|
||||
|
||||
str = getAndCheckNotNull(fields, "minOffset");
|
||||
if (str != null) {
|
||||
this.minOffset = Long.parseLong(str);
|
||||
}
|
||||
|
||||
str = getAndCheckNotNull(fields, "maxOffset");
|
||||
if (str != null) {
|
||||
this.maxOffset = Long.parseLong(str);
|
||||
}
|
||||
}
|
||||
|
||||
public Long getNextBeginOffset() {
|
||||
return nextBeginOffset;
|
||||
}
|
||||
|
||||
+92
-1
@@ -17,15 +17,20 @@
|
||||
|
||||
package org.apache.rocketmq.common.protocol.header;
|
||||
|
||||
import java.util.HashMap;
|
||||
|
||||
import org.apache.rocketmq.remoting.protocol.FastCodesHeader;
|
||||
import org.apache.rocketmq.remoting.CommandCustomHeader;
|
||||
import org.apache.rocketmq.remoting.annotation.CFNotNull;
|
||||
import org.apache.rocketmq.remoting.annotation.CFNullable;
|
||||
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
|
||||
|
||||
import io.netty.buffer.ByteBuf;
|
||||
|
||||
/**
|
||||
* Use short variable name to speed up FastJson deserialization process.
|
||||
*/
|
||||
public class SendMessageRequestHeaderV2 implements CommandCustomHeader {
|
||||
public class SendMessageRequestHeaderV2 implements CommandCustomHeader, FastCodesHeader {
|
||||
@CFNotNull
|
||||
private String a; // producerGroup;
|
||||
@CFNotNull
|
||||
@@ -94,6 +99,92 @@ public class SendMessageRequestHeaderV2 implements CommandCustomHeader {
|
||||
public void checkFields() throws RemotingCommandException {
|
||||
}
|
||||
|
||||
@Override
|
||||
public void encode(ByteBuf out) {
|
||||
writeIfNotNull(out, "a", a);
|
||||
writeIfNotNull(out, "b", b);
|
||||
writeIfNotNull(out, "c", c);
|
||||
writeIfNotNull(out, "d", d);
|
||||
writeIfNotNull(out, "e", e);
|
||||
writeIfNotNull(out, "f", f);
|
||||
writeIfNotNull(out, "g", g);
|
||||
writeIfNotNull(out, "h", h);
|
||||
writeIfNotNull(out, "i", i);
|
||||
writeIfNotNull(out, "j", j);
|
||||
writeIfNotNull(out, "k", k);
|
||||
writeIfNotNull(out, "l", l);
|
||||
writeIfNotNull(out, "m", m);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void decode(HashMap<String, String> fields) throws RemotingCommandException {
|
||||
|
||||
String str = getAndCheckNotNull(fields, "a");
|
||||
if (str != null) {
|
||||
a = str;
|
||||
}
|
||||
|
||||
str = getAndCheckNotNull(fields, "b");
|
||||
if (str != null) {
|
||||
b = str;
|
||||
}
|
||||
|
||||
str = getAndCheckNotNull(fields, "c");
|
||||
if (str != null) {
|
||||
c = str;
|
||||
}
|
||||
|
||||
str = getAndCheckNotNull(fields, "d");
|
||||
if (str != null) {
|
||||
d = Integer.parseInt(str);
|
||||
}
|
||||
|
||||
str = getAndCheckNotNull(fields, "e");
|
||||
if (str != null) {
|
||||
e = Integer.parseInt(str);
|
||||
}
|
||||
|
||||
str = getAndCheckNotNull(fields, "f");
|
||||
if (str != null) {
|
||||
f = Integer.parseInt(str);
|
||||
}
|
||||
|
||||
str = getAndCheckNotNull(fields, "g");
|
||||
if (str != null) {
|
||||
g = Long.parseLong(str);
|
||||
}
|
||||
|
||||
str = getAndCheckNotNull(fields, "h");
|
||||
if (str != null) {
|
||||
h = Integer.parseInt(str);
|
||||
}
|
||||
|
||||
str = fields.get("i");
|
||||
if (str != null) {
|
||||
i = str;
|
||||
}
|
||||
|
||||
str = fields.get("j");
|
||||
if (str != null) {
|
||||
j = Integer.parseInt(str);
|
||||
}
|
||||
|
||||
str = fields.get("k");
|
||||
if (str != null) {
|
||||
k = Boolean.parseBoolean(str);
|
||||
}
|
||||
|
||||
str = fields.get("l");
|
||||
if (str != null) {
|
||||
l = Integer.parseInt(str);
|
||||
}
|
||||
|
||||
str = fields.get("m");
|
||||
if (str != null) {
|
||||
m = Boolean.parseBoolean(str);
|
||||
}
|
||||
}
|
||||
|
||||
public String getA() {
|
||||
return a;
|
||||
}
|
||||
|
||||
+37
-1
@@ -20,11 +20,16 @@
|
||||
*/
|
||||
package org.apache.rocketmq.common.protocol.header;
|
||||
|
||||
import java.util.HashMap;
|
||||
|
||||
import org.apache.rocketmq.remoting.CommandCustomHeader;
|
||||
import org.apache.rocketmq.remoting.annotation.CFNotNull;
|
||||
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
|
||||
import org.apache.rocketmq.remoting.protocol.FastCodesHeader;
|
||||
|
||||
public class SendMessageResponseHeader implements CommandCustomHeader {
|
||||
import io.netty.buffer.ByteBuf;
|
||||
|
||||
public class SendMessageResponseHeader implements CommandCustomHeader, FastCodesHeader {
|
||||
@CFNotNull
|
||||
private String msgId;
|
||||
@CFNotNull
|
||||
@@ -37,6 +42,37 @@ public class SendMessageResponseHeader implements CommandCustomHeader {
|
||||
public void checkFields() throws RemotingCommandException {
|
||||
}
|
||||
|
||||
@Override
|
||||
public void encode(ByteBuf out) {
|
||||
writeIfNotNull(out, "msgId", msgId);
|
||||
writeIfNotNull(out, "queueId", queueId);
|
||||
writeIfNotNull(out, "queueOffset", queueOffset);
|
||||
writeIfNotNull(out, "transactionId", transactionId);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void decode(HashMap<String, String> fields) throws RemotingCommandException {
|
||||
String str = getAndCheckNotNull(fields, "msgId");
|
||||
if (str != null) {
|
||||
this.msgId = str;
|
||||
}
|
||||
|
||||
str = getAndCheckNotNull(fields, "queueId");
|
||||
if (str != null) {
|
||||
this.queueId = Integer.parseInt(str);
|
||||
}
|
||||
|
||||
str = getAndCheckNotNull(fields, "queueOffset");
|
||||
if (str != null) {
|
||||
this.queueOffset = Long.parseLong(str);
|
||||
}
|
||||
|
||||
str = fields.get("transactionId");
|
||||
if (str != null) {
|
||||
this.transactionId = str;
|
||||
}
|
||||
}
|
||||
|
||||
public String getMsgId() {
|
||||
return msgId;
|
||||
}
|
||||
|
||||
+20
-11
@@ -14,24 +14,31 @@
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
package org.apache.rocketmq.broker.processor;
|
||||
package org.apache.rocketmq.common.protocol.header;
|
||||
|
||||
import java.lang.reflect.Field;
|
||||
import java.util.ArrayList;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.stream.Collectors;
|
||||
import java.util.stream.Stream;
|
||||
|
||||
import org.apache.rocketmq.common.protocol.header.SendMessageRequestHeaderV2;
|
||||
import org.apache.rocketmq.remoting.CommandCustomHeader;
|
||||
import org.apache.rocketmq.remoting.protocol.FastCodesHeader;
|
||||
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
|
||||
import org.junit.Assert;
|
||||
import org.junit.Test;
|
||||
|
||||
public class AbstractSendMessageProcessorTest {
|
||||
public class FastCodesHeaderTest {
|
||||
|
||||
@Test
|
||||
public void testDecodeSendMessageHeaderV2() throws Exception {
|
||||
Field[] declaredFields = SendMessageRequestHeaderV2.class.getDeclaredFields();
|
||||
public void testFastDecode() throws Exception {
|
||||
testFastDecode(SendMessageRequestHeaderV2.class);
|
||||
testFastDecode(SendMessageResponseHeader.class);
|
||||
testFastDecode(PullMessageRequestHeader.class);
|
||||
testFastDecode(PullMessageResponseHeader.class);
|
||||
}
|
||||
|
||||
private void testFastDecode(Class<? extends CommandCustomHeader> classHeader) throws Exception {
|
||||
Field[] declaredFields = classHeader.getDeclaredFields();
|
||||
List<Field> declaredFieldsList = new ArrayList<>();
|
||||
for (Field f : declaredFields) {
|
||||
if (f.getName().startsWith("$")) {
|
||||
@@ -43,7 +50,7 @@ public class AbstractSendMessageProcessorTest {
|
||||
RemotingCommand command = RemotingCommand.createRequestCommand(0, null);
|
||||
HashMap<String, String> m = buildExtFields(declaredFieldsList);
|
||||
command.setExtFields(m);
|
||||
check(command, declaredFieldsList);
|
||||
check(command, declaredFieldsList, classHeader);
|
||||
}
|
||||
|
||||
private HashMap<String, String> buildExtFields(List<Field> fields) {
|
||||
@@ -65,9 +72,11 @@ public class AbstractSendMessageProcessorTest {
|
||||
return extFields;
|
||||
}
|
||||
|
||||
private void check(RemotingCommand command, List<Field> fields) throws Exception {
|
||||
SendMessageRequestHeaderV2 o1 = (SendMessageRequestHeaderV2) command.decodeCommandCustomHeader(SendMessageRequestHeaderV2.class);
|
||||
SendMessageRequestHeaderV2 o2 = AbstractSendMessageProcessor.decodeSendMessageHeaderV2(command);
|
||||
private void check(RemotingCommand command, List<Field> fields,
|
||||
Class<? extends CommandCustomHeader> classHeader) throws Exception {
|
||||
CommandCustomHeader o1 = command.decodeCommandCustomHeader(classHeader, false);
|
||||
CommandCustomHeader o2 = classHeader.newInstance();
|
||||
((FastCodesHeader)o2).decode(command.getExtFields());
|
||||
for (Field f : fields) {
|
||||
Object value1 = f.get(o1);
|
||||
Object value2 = f.get(o2);
|
||||
@@ -47,6 +47,8 @@ import org.apache.rocketmq.logging.InternalLogger;
|
||||
import org.apache.rocketmq.common.message.Message;
|
||||
import org.apache.rocketmq.remoting.RPCHook;
|
||||
import org.apache.rocketmq.remoting.exception.RemotingException;
|
||||
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
|
||||
import org.apache.rocketmq.remoting.protocol.SerializeType;
|
||||
import org.apache.rocketmq.srvutil.ServerUtil;
|
||||
|
||||
public class BatchProducer {
|
||||
@@ -54,6 +56,7 @@ public class BatchProducer {
|
||||
private static byte[] msgBody;
|
||||
|
||||
public static void main(String[] args) throws MQClientException {
|
||||
System.setProperty(RemotingCommand.SERIALIZE_TYPE_PROPERTY, SerializeType.ROCKETMQ.name());
|
||||
|
||||
Options options = ServerUtil.buildCommandlineOptions(new Options());
|
||||
CommandLine commandLine = ServerUtil.parseCmdLine("benchmarkBatchProducer", args, buildCommandlineOptions(options), new PosixParser());
|
||||
|
||||
@@ -34,6 +34,8 @@ import org.apache.rocketmq.common.MixAll;
|
||||
import org.apache.rocketmq.common.filter.ExpressionType;
|
||||
import org.apache.rocketmq.common.message.MessageExt;
|
||||
import org.apache.rocketmq.remoting.RPCHook;
|
||||
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
|
||||
import org.apache.rocketmq.remoting.protocol.SerializeType;
|
||||
import org.apache.rocketmq.srvutil.ServerUtil;
|
||||
|
||||
import java.io.IOException;
|
||||
@@ -49,6 +51,7 @@ import java.util.concurrent.atomic.AtomicLong;
|
||||
public class Consumer {
|
||||
|
||||
public static void main(String[] args) throws MQClientException, IOException {
|
||||
System.setProperty(RemotingCommand.SERIALIZE_TYPE_PROPERTY, SerializeType.ROCKETMQ.name());
|
||||
Options options = ServerUtil.buildCommandlineOptions(new Options());
|
||||
CommandLine commandLine = ServerUtil.parseCmdLine("benchmarkConsumer", args, buildCommandlineOptions(options), new PosixParser());
|
||||
if (null == commandLine) {
|
||||
|
||||
@@ -36,6 +36,8 @@ import org.apache.rocketmq.common.message.Message;
|
||||
import org.apache.rocketmq.logging.InternalLogger;
|
||||
import org.apache.rocketmq.remoting.RPCHook;
|
||||
import org.apache.rocketmq.remoting.exception.RemotingException;
|
||||
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
|
||||
import org.apache.rocketmq.remoting.protocol.SerializeType;
|
||||
import org.apache.rocketmq.srvutil.ServerUtil;
|
||||
|
||||
import java.util.Arrays;
|
||||
@@ -56,6 +58,7 @@ public class Producer {
|
||||
private static final int SLEEP_FOR_A_WHILE = 100;
|
||||
|
||||
public static void main(String[] args) throws MQClientException {
|
||||
System.setProperty(RemotingCommand.SERIALIZE_TYPE_PROPERTY, SerializeType.ROCKETMQ.name());
|
||||
|
||||
Options options = ServerUtil.buildCommandlineOptions(new Options());
|
||||
CommandLine commandLine = ServerUtil.parseCmdLine("benchmarkProducer", args, buildCommandlineOptions(options), new PosixParser());
|
||||
|
||||
@@ -32,6 +32,8 @@ import org.apache.rocketmq.common.message.Message;
|
||||
import org.apache.rocketmq.common.message.MessageConst;
|
||||
import org.apache.rocketmq.common.message.MessageExt;
|
||||
import org.apache.rocketmq.remoting.RPCHook;
|
||||
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
|
||||
import org.apache.rocketmq.remoting.protocol.SerializeType;
|
||||
import org.apache.rocketmq.srvutil.ServerUtil;
|
||||
|
||||
import java.io.UnsupportedEncodingException;
|
||||
@@ -61,6 +63,7 @@ public class TransactionProducer {
|
||||
static final int MAX_CHECK_RESULT_IN_MSG = 20;
|
||||
|
||||
public static void main(String[] args) throws MQClientException, UnsupportedEncodingException {
|
||||
System.setProperty(RemotingCommand.SERIALIZE_TYPE_PROPERTY, SerializeType.ROCKETMQ.name());
|
||||
Options options = ServerUtil.buildCommandlineOptions(new Options());
|
||||
CommandLine commandLine = ServerUtil.parseCmdLine("TransactionProducer", args, buildCommandlineOptions(options), new PosixParser());
|
||||
TxSendConfig config = new TxSendConfig();
|
||||
|
||||
@@ -19,7 +19,6 @@ package org.apache.rocketmq.remoting.netty;
|
||||
import io.netty.buffer.ByteBuf;
|
||||
import io.netty.channel.ChannelHandlerContext;
|
||||
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
|
||||
import java.nio.ByteBuffer;
|
||||
import org.apache.rocketmq.remoting.common.RemotingHelper;
|
||||
import org.apache.rocketmq.remoting.common.RemotingUtil;
|
||||
import org.apache.rocketmq.logging.InternalLogger;
|
||||
@@ -44,10 +43,7 @@ public class NettyDecoder extends LengthFieldBasedFrameDecoder {
|
||||
if (null == frame) {
|
||||
return null;
|
||||
}
|
||||
|
||||
ByteBuffer byteBuffer = frame.nioBuffer();
|
||||
|
||||
return RemotingCommand.decode(byteBuffer);
|
||||
return RemotingCommand.decode(frame);
|
||||
} catch (Exception e) {
|
||||
log.error("decode exception, " + RemotingHelper.parseChannelRemoteAddr(ctx.channel()), e);
|
||||
RemotingUtil.closeChannel(ctx.channel());
|
||||
|
||||
@@ -20,7 +20,6 @@ import io.netty.buffer.ByteBuf;
|
||||
import io.netty.channel.ChannelHandler;
|
||||
import io.netty.channel.ChannelHandlerContext;
|
||||
import io.netty.handler.codec.MessageToByteEncoder;
|
||||
import java.nio.ByteBuffer;
|
||||
import org.apache.rocketmq.remoting.common.RemotingHelper;
|
||||
import org.apache.rocketmq.remoting.common.RemotingUtil;
|
||||
import org.apache.rocketmq.logging.InternalLogger;
|
||||
@@ -35,8 +34,7 @@ public class NettyEncoder extends MessageToByteEncoder<RemotingCommand> {
|
||||
public void encode(ChannelHandlerContext ctx, RemotingCommand remotingCommand, ByteBuf out)
|
||||
throws Exception {
|
||||
try {
|
||||
ByteBuffer header = remotingCommand.encodeHeader();
|
||||
out.writeBytes(header);
|
||||
remotingCommand.fastEncodeHeader(out);
|
||||
byte[] body = remotingCommand.getBody();
|
||||
if (body != null) {
|
||||
out.writeBytes(body);
|
||||
|
||||
@@ -209,6 +209,7 @@ public abstract class NettyRemotingAbstract {
|
||||
if (response != null) {
|
||||
response.setOpaque(opaque);
|
||||
response.markResponseType();
|
||||
response.setSerializeTypeCurrentRPC(cmd.getSerializeTypeCurrentRPC());
|
||||
try {
|
||||
ctx.writeAndFlush(response);
|
||||
} catch (Throwable e) {
|
||||
|
||||
@@ -0,0 +1,50 @@
|
||||
/*
|
||||
* 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.remoting.protocol;
|
||||
|
||||
import java.util.HashMap;
|
||||
|
||||
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
|
||||
|
||||
import io.netty.buffer.ByteBuf;
|
||||
|
||||
public interface FastCodesHeader {
|
||||
|
||||
default String getAndCheckNotNull(HashMap<String, String> fields, String field) {
|
||||
String value = fields.get(field);
|
||||
if (value == null) {
|
||||
String headerClass = this.getClass().getSimpleName();
|
||||
RemotingCommand.log.error("the custom field {}.{} is null", headerClass, field);
|
||||
// no exception throws, keep compatible with RemotingCommand.decodeCommandCustomHeader
|
||||
}
|
||||
return value;
|
||||
}
|
||||
|
||||
default void writeIfNotNull(ByteBuf out, String key, Object value) {
|
||||
if (value != null) {
|
||||
RocketMQSerializable.writeStr(out, true, key);
|
||||
RocketMQSerializable.writeStr(out, false, value.toString());
|
||||
}
|
||||
}
|
||||
|
||||
public void encode(ByteBuf out);
|
||||
|
||||
void decode(HashMap<String, String> fields) throws RemotingCommandException;
|
||||
|
||||
|
||||
}
|
||||
@@ -31,11 +31,14 @@ import org.apache.rocketmq.remoting.annotation.CFNotNull;
|
||||
import org.apache.rocketmq.remoting.common.RemotingHelper;
|
||||
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
|
||||
|
||||
import io.netty.buffer.ByteBuf;
|
||||
import io.netty.buffer.Unpooled;
|
||||
|
||||
public class RemotingCommand {
|
||||
public static final String SERIALIZE_TYPE_PROPERTY = "rocketmq.serialize.type";
|
||||
public static final String SERIALIZE_TYPE_ENV = "ROCKETMQ_SERIALIZE_TYPE";
|
||||
public static final String REMOTING_VERSION_KEY = "rocketmq.remoting.version";
|
||||
private static final InternalLogger log = InternalLoggerFactory.getLogger(RemotingHelper.ROCKETMQ_REMOTING);
|
||||
static final InternalLogger log = InternalLoggerFactory.getLogger(RemotingHelper.ROCKETMQ_REMOTING);
|
||||
private static final int RPC_TYPE = 0; // 0, REQUEST_COMMAND
|
||||
private static final int RPC_ONEWAY = 1; // 0, RPC
|
||||
private static final Map<Class<? extends CommandCustomHeader>, Field[]> CLASS_HASH_MAP =
|
||||
@@ -142,20 +145,24 @@ public class RemotingCommand {
|
||||
}
|
||||
|
||||
public static RemotingCommand decode(final ByteBuffer byteBuffer) throws RemotingCommandException {
|
||||
int length = byteBuffer.limit();
|
||||
int oriHeaderLen = byteBuffer.getInt();
|
||||
return decode(Unpooled.wrappedBuffer(byteBuffer));
|
||||
}
|
||||
|
||||
public static RemotingCommand decode(final ByteBuf byteBuffer) throws RemotingCommandException {
|
||||
int length = byteBuffer.readableBytes();
|
||||
int oriHeaderLen = byteBuffer.readInt();
|
||||
int headerLength = getHeaderLength(oriHeaderLen);
|
||||
if (headerLength > length - 4) {
|
||||
throw new RemotingCommandException("decode error, bad header length: " + headerLength);
|
||||
}
|
||||
|
||||
byte[] headerData = new byte[headerLength];
|
||||
byteBuffer.get(headerData);
|
||||
|
||||
RemotingCommand cmd = headerDecode(headerData, getProtocolType(oriHeaderLen));
|
||||
RemotingCommand cmd = headerDecode(byteBuffer, headerLength, getProtocolType(oriHeaderLen));
|
||||
|
||||
int bodyLength = length - 4 - headerLength;
|
||||
byte[] bodyData = null;
|
||||
if (bodyLength > 0) {
|
||||
bodyData = new byte[bodyLength];
|
||||
byteBuffer.get(bodyData);
|
||||
byteBuffer.readBytes(bodyData);
|
||||
}
|
||||
cmd.body = bodyData;
|
||||
|
||||
@@ -166,14 +173,16 @@ public class RemotingCommand {
|
||||
return length & 0xFFFFFF;
|
||||
}
|
||||
|
||||
private static RemotingCommand headerDecode(byte[] headerData, SerializeType type) throws RemotingCommandException {
|
||||
private static RemotingCommand headerDecode(ByteBuf byteBuffer, int len, SerializeType type) throws RemotingCommandException {
|
||||
switch (type) {
|
||||
case JSON:
|
||||
byte[] headerData = new byte[len];
|
||||
byteBuffer.readBytes(headerData);
|
||||
RemotingCommand resultJson = RemotingSerializable.decode(headerData, RemotingCommand.class);
|
||||
resultJson.setSerializeTypeCurrentRPC(type);
|
||||
return resultJson;
|
||||
case ROCKETMQ:
|
||||
RemotingCommand resultRMQ = RocketMQSerializable.rocketMQProtocolDecode(headerData);
|
||||
RemotingCommand resultRMQ = RocketMQSerializable.rocketMQProtocolDecode(byteBuffer, len);
|
||||
resultRMQ.setSerializeTypeCurrentRPC(type);
|
||||
return resultRMQ;
|
||||
default:
|
||||
@@ -208,14 +217,8 @@ public class RemotingCommand {
|
||||
return true;
|
||||
}
|
||||
|
||||
public static byte[] markProtocolType(int source, SerializeType type) {
|
||||
byte[] result = new byte[4];
|
||||
|
||||
result[0] = type.getCode();
|
||||
result[1] = (byte) ((source >> 16) & 0xFF);
|
||||
result[2] = (byte) ((source >> 8) & 0xFF);
|
||||
result[3] = (byte) (source & 0xFF);
|
||||
return result;
|
||||
public static int markProtocolType(int source, SerializeType type) {
|
||||
return (type.getCode() << 24) | (source & 0x00FFFFFF);
|
||||
}
|
||||
|
||||
public void markResponseType() {
|
||||
@@ -232,7 +235,12 @@ public class RemotingCommand {
|
||||
}
|
||||
|
||||
public CommandCustomHeader decodeCommandCustomHeader(
|
||||
Class<? extends CommandCustomHeader> classHeader) throws RemotingCommandException {
|
||||
Class<? extends CommandCustomHeader> classHeader) throws RemotingCommandException {
|
||||
return decodeCommandCustomHeader(classHeader, true);
|
||||
}
|
||||
|
||||
public CommandCustomHeader decodeCommandCustomHeader(Class<? extends CommandCustomHeader> classHeader,
|
||||
boolean useFastEncode) throws RemotingCommandException {
|
||||
CommandCustomHeader objectHeader;
|
||||
try {
|
||||
objectHeader = classHeader.newInstance();
|
||||
@@ -243,6 +251,11 @@ public class RemotingCommand {
|
||||
}
|
||||
|
||||
if (this.extFields != null) {
|
||||
if (objectHeader instanceof FastCodesHeader && useFastEncode) {
|
||||
((FastCodesHeader) objectHeader).decode(this.extFields);
|
||||
objectHeader.checkFields();
|
||||
return objectHeader;
|
||||
}
|
||||
|
||||
Field[] fields = getClazzFields(classHeader);
|
||||
for (Field field : fields) {
|
||||
@@ -344,7 +357,7 @@ public class RemotingCommand {
|
||||
result.putInt(length);
|
||||
|
||||
// header length
|
||||
result.put(markProtocolType(headerData.length, serializeTypeCurrentRPC));
|
||||
result.putInt(markProtocolType(headerData.length, serializeTypeCurrentRPC));
|
||||
|
||||
// header data
|
||||
result.put(headerData);
|
||||
@@ -396,6 +409,27 @@ public class RemotingCommand {
|
||||
}
|
||||
}
|
||||
|
||||
public void fastEncodeHeader(ByteBuf out) {
|
||||
int bodySize = this.body != null ? this.body.length : 0;
|
||||
int beginIndex = out.writerIndex();
|
||||
// skip 8 bytes
|
||||
out.writeLong(0);
|
||||
int headerSize;
|
||||
if (SerializeType.ROCKETMQ == serializeTypeCurrentRPC) {
|
||||
if (customHeader != null && !(customHeader instanceof FastCodesHeader)) {
|
||||
this.makeCustomHeaderToNet();
|
||||
}
|
||||
headerSize = RocketMQSerializable.rocketMQProtocolEncode(this, out);
|
||||
} else {
|
||||
this.makeCustomHeaderToNet();
|
||||
byte[] header = RemotingSerializable.encode(this);
|
||||
headerSize = header.length;
|
||||
out.writeBytes(header);
|
||||
}
|
||||
out.setInt(beginIndex, 4 + headerSize + bodySize);
|
||||
out.setInt(beginIndex + 4, markProtocolType(headerSize, serializeTypeCurrentRPC));
|
||||
}
|
||||
|
||||
public ByteBuffer encodeHeader() {
|
||||
return encodeHeader(this.body != null ? this.body.length : 0);
|
||||
}
|
||||
@@ -419,7 +453,7 @@ public class RemotingCommand {
|
||||
result.putInt(length);
|
||||
|
||||
// header length
|
||||
result.put(markProtocolType(headerData.length, serializeTypeCurrentRPC));
|
||||
result.putInt(markProtocolType(headerData.length, serializeTypeCurrentRPC));
|
||||
|
||||
// header data
|
||||
result.put(headerData);
|
||||
|
||||
+86
-41
@@ -24,9 +24,77 @@ import java.util.Iterator;
|
||||
import java.util.Map;
|
||||
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
|
||||
|
||||
import io.netty.buffer.ByteBuf;
|
||||
|
||||
public class RocketMQSerializable {
|
||||
private static final Charset CHARSET_UTF8 = StandardCharsets.UTF_8;
|
||||
|
||||
public static void writeStr(ByteBuf buf, boolean useShortLength, String str) {
|
||||
int lenIndex = buf.writerIndex();
|
||||
if (useShortLength) {
|
||||
buf.writeShort(0);
|
||||
} else {
|
||||
buf.writeInt(0);
|
||||
}
|
||||
int len = buf.writeCharSequence(str, StandardCharsets.UTF_8);
|
||||
if (useShortLength) {
|
||||
buf.setShort(lenIndex, len);
|
||||
} else {
|
||||
buf.setInt(lenIndex, len);
|
||||
}
|
||||
}
|
||||
|
||||
private static String readStr(ByteBuf buf, boolean useShortLength, int limit) throws RemotingCommandException {
|
||||
int len = useShortLength ? buf.readShort() : buf.readInt();
|
||||
if (len == 0) {
|
||||
return null;
|
||||
}
|
||||
if (len > limit) {
|
||||
throw new RemotingCommandException("string length exceed limit:" + limit);
|
||||
}
|
||||
CharSequence cs = buf.readCharSequence(len, StandardCharsets.UTF_8);
|
||||
return cs == null ? null : cs.toString();
|
||||
}
|
||||
|
||||
public static int rocketMQProtocolEncode(RemotingCommand cmd, ByteBuf out) {
|
||||
int beginIndex = out.writerIndex();
|
||||
// int code(~32767)
|
||||
out.writeShort(cmd.getCode());
|
||||
// LanguageCode language
|
||||
out.writeByte(cmd.getLanguage().getCode());
|
||||
// int version(~32767)
|
||||
out.writeShort(cmd.getVersion());
|
||||
// int opaque
|
||||
out.writeInt(cmd.getOpaque());
|
||||
// int flag
|
||||
out.writeInt(cmd.getFlag());
|
||||
// String remark
|
||||
String remark = cmd.getRemark();
|
||||
if (remark != null && !remark.isEmpty()) {
|
||||
writeStr(out, false, remark);
|
||||
} else {
|
||||
out.writeInt(0);
|
||||
}
|
||||
|
||||
int mapLenIndex = out.writerIndex();
|
||||
out.writeInt(0);
|
||||
if (cmd.readCustomHeader() instanceof FastCodesHeader) {
|
||||
((FastCodesHeader) cmd.readCustomHeader()).encode(out);
|
||||
}
|
||||
HashMap<String, String> map = cmd.getExtFields();
|
||||
if (map != null && !map.isEmpty()) {
|
||||
map.forEach((k, v) -> {
|
||||
if (k != null && v != null) {
|
||||
writeStr(out, true, k);
|
||||
writeStr(out, false, v);
|
||||
}
|
||||
});
|
||||
}
|
||||
out.setInt(mapLenIndex, out.writerIndex() - mapLenIndex - 4);
|
||||
return out.writerIndex() - beginIndex;
|
||||
}
|
||||
|
||||
|
||||
public static byte[] rocketMQProtocolEncode(RemotingCommand cmd) {
|
||||
// String remark
|
||||
byte[] remarkBytes = null;
|
||||
@@ -135,64 +203,41 @@ public class RocketMQSerializable {
|
||||
return length;
|
||||
}
|
||||
|
||||
public static RemotingCommand rocketMQProtocolDecode(final byte[] headerArray) throws RemotingCommandException {
|
||||
public static RemotingCommand rocketMQProtocolDecode(final ByteBuf headerBuffer, int headerLen) throws RemotingCommandException {
|
||||
RemotingCommand cmd = new RemotingCommand();
|
||||
ByteBuffer headerBuffer = ByteBuffer.wrap(headerArray);
|
||||
// int code(~32767)
|
||||
cmd.setCode(headerBuffer.getShort());
|
||||
cmd.setCode(headerBuffer.readShort());
|
||||
// LanguageCode language
|
||||
cmd.setLanguage(LanguageCode.valueOf(headerBuffer.get()));
|
||||
cmd.setLanguage(LanguageCode.valueOf(headerBuffer.readByte()));
|
||||
// int version(~32767)
|
||||
cmd.setVersion(headerBuffer.getShort());
|
||||
cmd.setVersion(headerBuffer.readShort());
|
||||
// int opaque
|
||||
cmd.setOpaque(headerBuffer.getInt());
|
||||
cmd.setOpaque(headerBuffer.readInt());
|
||||
// int flag
|
||||
cmd.setFlag(headerBuffer.getInt());
|
||||
cmd.setFlag(headerBuffer.readInt());
|
||||
// String remark
|
||||
int remarkLength = headerBuffer.getInt();
|
||||
if (remarkLength > 0) {
|
||||
if (remarkLength > headerArray.length) {
|
||||
throw new RemotingCommandException("RocketMQ protocol decoding failed, remark length: " + remarkLength + ", but header length: " + headerArray.length);
|
||||
}
|
||||
byte[] remarkContent = new byte[remarkLength];
|
||||
headerBuffer.get(remarkContent);
|
||||
cmd.setRemark(new String(remarkContent, CHARSET_UTF8));
|
||||
}
|
||||
cmd.setRemark(readStr(headerBuffer, false, headerLen));
|
||||
|
||||
// HashMap<String, String> extFields
|
||||
int extFieldsLength = headerBuffer.getInt();
|
||||
int extFieldsLength = headerBuffer.readInt();
|
||||
if (extFieldsLength > 0) {
|
||||
if (extFieldsLength > headerArray.length) {
|
||||
throw new RemotingCommandException("RocketMQ protocol decoding failed, extFields length: " + extFieldsLength + ", but header length: " + headerArray.length);
|
||||
if (extFieldsLength > headerLen) {
|
||||
throw new RemotingCommandException("RocketMQ protocol decoding failed, extFields length: " + extFieldsLength + ", but header length: " + headerLen);
|
||||
}
|
||||
byte[] extFieldsBytes = new byte[extFieldsLength];
|
||||
headerBuffer.get(extFieldsBytes);
|
||||
cmd.setExtFields(mapDeserialize(extFieldsBytes));
|
||||
cmd.setExtFields(mapDeserialize(headerBuffer, extFieldsLength));
|
||||
}
|
||||
return cmd;
|
||||
}
|
||||
|
||||
public static HashMap<String, String> mapDeserialize(byte[] bytes) {
|
||||
if (bytes == null || bytes.length <= 0)
|
||||
return null;
|
||||
public static HashMap<String, String> mapDeserialize(ByteBuf byteBuffer, int len) throws RemotingCommandException {
|
||||
|
||||
HashMap<String, String> map = new HashMap<String, String>();
|
||||
ByteBuffer byteBuffer = ByteBuffer.wrap(bytes);
|
||||
HashMap<String, String> map = new HashMap<>();
|
||||
int endIndex = byteBuffer.readerIndex() + len;
|
||||
|
||||
short keySize;
|
||||
byte[] keyContent;
|
||||
int valSize;
|
||||
byte[] valContent;
|
||||
while (byteBuffer.hasRemaining()) {
|
||||
keySize = byteBuffer.getShort();
|
||||
keyContent = new byte[keySize];
|
||||
byteBuffer.get(keyContent);
|
||||
|
||||
valSize = byteBuffer.getInt();
|
||||
valContent = new byte[valSize];
|
||||
byteBuffer.get(valContent);
|
||||
|
||||
map.put(new String(keyContent, CHARSET_UTF8), new String(valContent, CHARSET_UTF8));
|
||||
while (byteBuffer.readerIndex() < endIndex) {
|
||||
String k = readStr(byteBuffer, true, len);
|
||||
String v = readStr(byteBuffer, false, len);
|
||||
map.put(k, v);
|
||||
}
|
||||
return map;
|
||||
}
|
||||
|
||||
+13
-2
@@ -32,7 +32,13 @@ public class RemotingCommandTest {
|
||||
public void testMarkProtocolType_JSONProtocolType() {
|
||||
int source = 261;
|
||||
SerializeType type = SerializeType.JSON;
|
||||
byte[] result = RemotingCommand.markProtocolType(source, type);
|
||||
|
||||
byte[] result = new byte[4];
|
||||
int x = RemotingCommand.markProtocolType(source, type);
|
||||
result[0] = (byte) (x >> 24);
|
||||
result[1] = (byte) (x >> 16);
|
||||
result[2] = (byte) (x >> 8);
|
||||
result[3] = (byte) x;
|
||||
assertThat(result).isEqualTo(new byte[] {0, 0, 1, 5});
|
||||
}
|
||||
|
||||
@@ -40,7 +46,12 @@ public class RemotingCommandTest {
|
||||
public void testMarkProtocolType_ROCKETMQProtocolType() {
|
||||
int source = 16777215;
|
||||
SerializeType type = SerializeType.ROCKETMQ;
|
||||
byte[] result = RemotingCommand.markProtocolType(source, type);
|
||||
byte[] result = new byte[4];
|
||||
int x = RemotingCommand.markProtocolType(source, type);
|
||||
result[0] = (byte) (x >> 24);
|
||||
result[1] = (byte) (x >> 16);
|
||||
result[2] = (byte) (x >> 8);
|
||||
result[3] = (byte) x;
|
||||
assertThat(result).isEqualTo(new byte[] {1, -1, -1, -1});
|
||||
}
|
||||
|
||||
|
||||
+63
-5
@@ -17,12 +17,17 @@
|
||||
package org.apache.rocketmq.remoting.protocol;
|
||||
|
||||
import java.util.HashMap;
|
||||
import org.apache.rocketmq.remoting.CommandCustomHeader;
|
||||
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
|
||||
import org.junit.Assert;
|
||||
import org.junit.Test;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
import io.netty.buffer.ByteBuf;
|
||||
import io.netty.buffer.ByteBufAllocator;
|
||||
import io.netty.buffer.Unpooled;
|
||||
|
||||
public class RocketMQSerializableTest {
|
||||
@Test
|
||||
public void testRocketMQProtocolEncodeAndDecode_WithoutRemarkWithoutExtFields() {
|
||||
@@ -46,7 +51,7 @@ public class RocketMQSerializableTest {
|
||||
|
||||
RemotingCommand decodedCommand = null;
|
||||
try {
|
||||
decodedCommand = RocketMQSerializable.rocketMQProtocolDecode(result);
|
||||
decodedCommand = RocketMQSerializable.rocketMQProtocolDecode(Unpooled.wrappedBuffer(result), result.length);
|
||||
|
||||
assertThat(decodedCommand.getCode()).isEqualTo(code);
|
||||
assertThat(decodedCommand.getLanguage()).isEqualTo(LanguageCode.JAVA);
|
||||
@@ -90,7 +95,7 @@ public class RocketMQSerializableTest {
|
||||
assertThat(parseToInt(result, 30)).isEqualTo(0); //empty extFields
|
||||
|
||||
try {
|
||||
RemotingCommand decodedCommand = RocketMQSerializable.rocketMQProtocolDecode(result);
|
||||
RemotingCommand decodedCommand = RocketMQSerializable.rocketMQProtocolDecode(Unpooled.wrappedBuffer(result), result.length);
|
||||
|
||||
assertThat(decodedCommand.getCode()).isEqualTo(code);
|
||||
assertThat(decodedCommand.getLanguage()).isEqualTo(LanguageCode.JAVA);
|
||||
@@ -107,7 +112,7 @@ public class RocketMQSerializableTest {
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testRocketMQProtocolEncodeAndDecode_WithoutRemarkWithExtFields() {
|
||||
public void testRocketMQProtocolEncodeAndDecode_WithoutRemarkWithExtFields() throws Exception {
|
||||
System.setProperty(RemotingCommand.REMOTING_VERSION_KEY, "2333");
|
||||
|
||||
//org.apache.rocketmq.common.protocol.RequestCode.REGISTER_BROKER
|
||||
@@ -130,11 +135,12 @@ public class RocketMQSerializableTest {
|
||||
|
||||
byte[] extFieldsArray = new byte[14];
|
||||
System.arraycopy(result, 21, extFieldsArray, 0, 14);
|
||||
HashMap<String, String> extFields = RocketMQSerializable.mapDeserialize(extFieldsArray);
|
||||
HashMap<String, String> extFields =
|
||||
RocketMQSerializable.mapDeserialize(Unpooled.wrappedBuffer(extFieldsArray), extFieldsArray.length);
|
||||
assertThat(extFields).contains(new HashMap.SimpleEntry("key", "value"));
|
||||
|
||||
try {
|
||||
RemotingCommand decodedCommand = RocketMQSerializable.rocketMQProtocolDecode(result);
|
||||
RemotingCommand decodedCommand = RocketMQSerializable.rocketMQProtocolDecode(Unpooled.wrappedBuffer(result), result.length);
|
||||
assertThat(decodedCommand.getCode()).isEqualTo(code);
|
||||
assertThat(decodedCommand.getLanguage()).isEqualTo(LanguageCode.JAVA);
|
||||
assertThat(decodedCommand.getVersion()).isEqualTo(2333);
|
||||
@@ -170,4 +176,56 @@ public class RocketMQSerializableTest {
|
||||
return array[index] * 16777216 + array[++index] * 65536 + array[++index] * 256
|
||||
+ array[++index];
|
||||
}
|
||||
|
||||
public static class MyHeader1 implements CommandCustomHeader {
|
||||
private String str;
|
||||
private int num;
|
||||
|
||||
@Override
|
||||
public void checkFields() throws RemotingCommandException {
|
||||
}
|
||||
|
||||
public String getStr() {
|
||||
return str;
|
||||
}
|
||||
|
||||
public void setStr(String str) {
|
||||
this.str = str;
|
||||
}
|
||||
|
||||
public int getNum() {
|
||||
return num;
|
||||
}
|
||||
|
||||
public void setNum(int num) {
|
||||
this.num = num;
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testFastEncode() throws Exception {
|
||||
ByteBuf buf = ByteBufAllocator.DEFAULT.buffer(16);
|
||||
MyHeader1 header1 = new MyHeader1();
|
||||
header1.setStr("s1");
|
||||
header1.setNum(100);
|
||||
RemotingCommand cmd = RemotingCommand.createRequestCommand(1, header1);
|
||||
cmd.setRemark("remark");
|
||||
cmd.setOpaque(1001);
|
||||
cmd.setVersion(99);
|
||||
cmd.setLanguage(LanguageCode.JAVA);
|
||||
cmd.setFlag(3);
|
||||
cmd.makeCustomHeaderToNet();
|
||||
RocketMQSerializable.rocketMQProtocolEncode(cmd, buf);
|
||||
RemotingCommand cmd2 = RocketMQSerializable.rocketMQProtocolDecode(buf, buf.readableBytes());
|
||||
assertThat(cmd2.getRemark()).isEqualTo("remark");
|
||||
assertThat(cmd2.getCode()).isEqualTo(1);
|
||||
assertThat(cmd2.getOpaque()).isEqualTo(1001);
|
||||
assertThat(cmd2.getVersion()).isEqualTo(99);
|
||||
assertThat(cmd2.getLanguage()).isEqualTo(LanguageCode.JAVA);
|
||||
assertThat(cmd2.getFlag()).isEqualTo(3);
|
||||
|
||||
MyHeader1 h2 = (MyHeader1) cmd2.decodeCommandCustomHeader(MyHeader1.class);
|
||||
assertThat(h2.getStr()).isEqualTo("s1");
|
||||
assertThat(h2.getNum()).isEqualTo(100);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user