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] Add TopicMessageTypeValidator

Browse Source
This commit is contained in:
zhouxiang
2022-07-13 11:29:39 +08:00
parent 2fe29957cb
commit 1fa561cb94
7 changed files with 126 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
proxy/src/main/java/org/apache/rocketmq/proxy/common/ContextVariable.java
+1
View File
@@ -20,4 +20,5 @@ package org.apache.rocketmq.proxy.common;
public class ContextVariable {
public final static String REMOTE_ADDRESS = "remote-address";
public final static String LOCAL_ADDRESS = "local-address";
public final static String MESSAGE_TYPE = "message-type";
}
proxy/src/main/java/org/apache/rocketmq/proxy/common/ProxyExceptionCode.java
+1
View File
@@ -23,4 +23,5 @@ public enum ProxyExceptionCode {
INVALID_RECEIPT_HANDLE,
ILLEGAL_MESSAGE,
INTERNAL_SERVER_ERROR,
TOPIC_MESSAGE_TYPE_NOT_MATCH,
}
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/common/GrpcConverter.java
+15
View File
@@ -35,6 +35,7 @@ import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.common.attribute.TopicMessageType;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.consumer.ReceiptHandle;
import org.apache.rocketmq.common.filter.ExpressionType;
@@ -229,4 +230,18 @@ public class GrpcConverter {
.setName(NamespaceUtil.withoutNamespace(resourceNameWithNamespace))
.build();
}
public static TopicMessageType buildTopicMessageType(MessageType messageType) {
if (messageType.equals(MessageType.NORMAL)) {
return TopicMessageType.NORMAL;
} else if (messageType.equals(MessageType.FIFO)) {
return TopicMessageType.FIFO;
} else if (messageType.equals(MessageType.DELAY)) {
return TopicMessageType.DELAY;
} else if (messageType.equals(MessageType.TRANSACTION)) {
return TopicMessageType.TRANSACTION;
} else {
return TopicMessageType.UNSPECIFIED;
}
}
}
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/producer/SendMessageActivity.java
+24 -1
View File
@@ -40,11 +40,15 @@ import java.util.Set;
import java.util.concurrent.CompletableFuture;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.attribute.TopicMessageType;
import org.apache.rocketmq.common.message.MessageAccessor;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.sysflag.MessageSysFlag;
import org.apache.rocketmq.proxy.common.ContextVariable;
import org.apache.rocketmq.proxy.common.ProxyContext;
import org.apache.rocketmq.proxy.common.ProxyException;
import org.apache.rocketmq.proxy.common.ProxyExceptionCode;
import org.apache.rocketmq.proxy.grpc.v2.AbstractMessingActivity;
import org.apache.rocketmq.proxy.grpc.v2.common.GrpcClientSettingsManager;
import org.apache.rocketmq.proxy.grpc.v2.common.GrpcConverter;
@@ -52,14 +56,18 @@ import org.apache.rocketmq.proxy.grpc.v2.common.GrpcProxyException;
import org.apache.rocketmq.proxy.grpc.v2.common.ResponseBuilder;
import org.apache.rocketmq.proxy.processor.MessagingProcessor;
import org.apache.rocketmq.proxy.processor.QueueSelector;
import org.apache.rocketmq.proxy.processor.validator.DefaultTopicMessageTypeValidator;
import org.apache.rocketmq.proxy.processor.validator.TopicMessageTypeValidator;
import org.apache.rocketmq.proxy.service.route.MessageQueueView;
import org.apache.rocketmq.proxy.service.route.SelectableMessageQueue;
public class SendMessageActivity extends AbstractMessingActivity {
private final TopicMessageTypeValidator validator;
public SendMessageActivity(MessagingProcessor messagingProcessor,
GrpcClientSettingsManager grpcClientSettingsManager) {
super(messagingProcessor, grpcClientSettingsManager);
this.validator = new DefaultTopicMessageTypeValidator();
}
public CompletableFuture<SendMessageResponse> sendMessage(Context ctx, SendMessageRequest request) {
@@ -71,10 +79,13 @@ public class SendMessageActivity extends AbstractMessingActivity {
throw new GrpcProxyException(Code.MESSAGE_CORRUPTED, "no message to send");
}
MessageType messageType = parseMessageType(request.getMessagesList());
TopicMessageType topicMessageType = GrpcConverter.buildTopicMessageType(messageType);
List<Message> messageList = request.getMessagesList();
Resource topic = messageList.get(0).getTopic();
future = this.messagingProcessor.sendMessage(
context,
context.withVal(ContextVariable.MESSAGE_TYPE, topicMessageType.getValue()),
new SendMessageQueueSelector(request),
GrpcConverter.wrapResourceWithNamespace(topic),
buildMessage(context, request.getMessagesList(), topic)
@@ -241,6 +252,18 @@ public class SendMessageActivity extends AbstractMessingActivity {
return builder.build();
}
protected MessageType parseMessageType(List<Message> messageList) {
MessageType messageType = messageList.get(0).getSystemProperties().getMessageType();
for (Message message : messageList) {
MessageType messageType0 = message.getSystemProperties().getMessageType();
if (messageType0 != messageType) {
throw new GrpcProxyException(Code.ILLEGAL_MESSAGE, "message type is not match");
}
messageType = messageType0;
}
return messageType;
}
protected static class SendMessageQueueSelector implements QueueSelector {
private final SendMessageRequest request;
proxy/src/main/java/org/apache/rocketmq/proxy/processor/ProducerProcessor.java
+25 -11
View File
@@ -23,6 +23,7 @@ import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.attribute.TopicMessageType;
import org.apache.rocketmq.common.consumer.ReceiptHandle;
import org.apache.rocketmq.common.message.MessageAccessor;
import org.apache.rocketmq.common.message.MessageConst;
@@ -32,10 +33,14 @@ import org.apache.rocketmq.common.protocol.ResponseCode;
import org.apache.rocketmq.common.protocol.header.ConsumerSendMsgBackRequestHeader;
import org.apache.rocketmq.common.protocol.header.SendMessageRequestHeader;
import org.apache.rocketmq.common.sysflag.MessageSysFlag;
import org.apache.rocketmq.proxy.common.ContextVariable;
import org.apache.rocketmq.proxy.common.ProxyContext;
import org.apache.rocketmq.proxy.common.ProxyException;
import org.apache.rocketmq.proxy.common.ProxyExceptionCode;
import org.apache.rocketmq.proxy.common.utils.FutureUtils;
import org.apache.rocketmq.proxy.config.ConfigurationManager;
import org.apache.rocketmq.proxy.processor.validator.DefaultTopicMessageTypeValidator;
import org.apache.rocketmq.proxy.processor.validator.TopicMessageTypeValidator;
import org.apache.rocketmq.proxy.service.ServiceManager;
import org.apache.rocketmq.proxy.service.route.SelectableMessageQueue;
import org.apache.rocketmq.proxy.service.transaction.TransactionId;
@@ -44,11 +49,13 @@ import org.apache.rocketmq.remoting.protocol.RemotingCommand;
public class ProducerProcessor extends AbstractProcessor {
private final ExecutorService executor;
private final TopicMessageTypeValidator topicMessageTypeValidator;
public ProducerProcessor(MessagingProcessor messagingProcessor,
ServiceManager serviceManager, ExecutorService executor) {
super(messagingProcessor, serviceManager);
this.executor = executor;
this.topicMessageTypeValidator = new DefaultTopicMessageTypeValidator();
}
public CompletableFuture<List<SendResult>> sendMessage(ProxyContext ctx, QueueSelector queueSelector,
@@ -56,6 +63,13 @@ public class ProducerProcessor extends AbstractProcessor {
CompletableFuture<List<SendResult>> future = new CompletableFuture<>();
try {
String topic = messageExtList.get(0).getTopic();
if (ConfigurationManager.getProxyConfig().isEnableTopicMessageTypeCheck()) {
if (topicMessageTypeValidator != null) {
TopicMessageType topicMessageType = serviceManager.getMetadataService().getTopicMessageType(topic);
TopicMessageType messageType = TopicMessageType.valueOf(ctx.getVal(ContextVariable.MESSAGE_TYPE));
topicMessageTypeValidator.validate(topicMessageType, messageType);
}
}
SelectableMessageQueue messageQueue = queueSelector.select(ctx,
this.serviceManager.getTopicRouteService().getCurrentMessageQueueView(topic));
if (messageQueue == null) {
@@ -70,18 +84,18 @@ public class ProducerProcessor extends AbstractProcessor {
messageExtList,
requestHeader,
timeoutMillis)
.thenApplyAsync(sendResultList -> {
for (SendResult sendResult : sendResultList) {
int tranType = MessageSysFlag.getTransactionValue(requestHeader.getSysFlag());
if (SendStatus.SEND_OK.equals(sendResult.getSendStatus()) &&
tranType == MessageSysFlag.TRANSACTION_PREPARED_TYPE &&
StringUtils.isNotBlank(sendResult.getTransactionId())) {
TransactionId transactionId = TransactionId.genByBrokerTransactionId(messageQueue.getBrokerName(), sendResult);
sendResult.setTransactionId(transactionId.getProxyTransactionId());
.thenApplyAsync(sendResultList -> {
for (SendResult sendResult : sendResultList) {
int tranType = MessageSysFlag.getTransactionValue(requestHeader.getSysFlag());
if (SendStatus.SEND_OK.equals(sendResult.getSendStatus()) &&
tranType == MessageSysFlag.TRANSACTION_PREPARED_TYPE &&
StringUtils.isNotBlank(sendResult.getTransactionId())) {
TransactionId transactionId = TransactionId.genByBrokerTransactionId(messageQueue.getBrokerName(), sendResult);
sendResult.setTransactionId(transactionId.getProxyTransactionId());
}
}
}
return sendResultList;
}, this.executor);
return sendResultList;
}, this.executor);
} catch (Throwable t) {
future.completeExceptionally(t);
}
proxy/src/main/java/org/apache/rocketmq/proxy/processor/validator/DefaultTopicMessageTypeValidator.java
+31
View File
@@ -0,0 +1,31 @@
/*
* 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.processor.validator;
import org.apache.rocketmq.common.attribute.TopicMessageType;
import org.apache.rocketmq.proxy.common.ProxyException;
import org.apache.rocketmq.proxy.common.ProxyExceptionCode;
public class DefaultTopicMessageTypeValidator implements TopicMessageTypeValidator {
public void validate(TopicMessageType topicMessageType, TopicMessageType messageType) {
if (messageType.equals(TopicMessageType.UNSPECIFIED) || !messageType.equals(topicMessageType)) {
throw new ProxyException(ProxyExceptionCode.TOPIC_MESSAGE_TYPE_NOT_MATCH, messageType.name() + " " + topicMessageType.name());
}
}
}
proxy/src/main/java/org/apache/rocketmq/proxy/processor/validator/TopicMessageTypeValidator.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.processor.validator;
import org.apache.rocketmq.common.attribute.TopicMessageType;
public interface TopicMessageTypeValidator {
/**
* Will throw {@link org.apache.rocketmq.proxy.common.ProxyException} if validate failed.
* @param topicMessageType Target topic
* @param messageType Message's type
*/
void validate(TopicMessageType topicMessageType, TopicMessageType messageType);
}