github-forks
/
spring-cloud-alibaba
mirror of https://github.com/alibaba/spring-cloud-alibaba.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Fix rocketmq test

Browse Source
pull/2750/head
sorie 3 years ago committed by Steve Rao
parent 851cd68984
commit 6d21378420
4 changed files with 29 additions and 149 deletions
Show Stats Download Patch File Download Diff File

110
spring-cloud-alibaba-starters/spring-cloud-starter-stream-rocketmq/src/main/java/com/alibaba/cloud/stream/binder/rocketmq/RocketMQMessageChannelBinder.java
Unescape Escape View File

@ -16,14 +16,6 @@
package com.alibaba.cloud.stream.binder.rocketmq;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.LinkedBlockingDeque;
import com.alibaba.cloud.stream.binder.rocketmq.custom.RocketMQBeanContainerCache;
import com.alibaba.cloud.stream.binder.rocketmq.extend.ErrorAcknowledgeHandler;
import com.alibaba.cloud.stream.binder.rocketmq.integration.inbound.RocketMQInboundChannelAdapter;
@ -35,7 +27,6 @@ import com.alibaba.cloud.stream.binder.rocketmq.properties.RocketMQConsumerPrope
import com.alibaba.cloud.stream.binder.rocketmq.properties.RocketMQExtendedBindingProperties;
import com.alibaba.cloud.stream.binder.rocketmq.properties.RocketMQProducerProperties;
import com.alibaba.cloud.stream.binder.rocketmq.provisioning.RocketMQTopicProvisioner;
import com.alibaba.cloud.stream.binder.rocketmq.support.MessageCollector;
import com.alibaba.cloud.stream.binder.rocketmq.utils.RocketMQUtils;
import org.apache.rocketmq.common.protocol.NamespaceUtil;
@ -45,8 +36,6 @@ import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder;
import org.springframework.cloud.stream.binding.MessageConverterConfigurer;
import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory;
import org.springframework.cloud.stream.converter.MessageConverterUtils;
import org.springframework.cloud.stream.provisioning.ConsumerDestination;
import org.springframework.cloud.stream.provisioning.ProducerDestination;
import org.springframework.integration.StaticMessageHeaderAccessor;
@ -55,18 +44,9 @@ import org.springframework.integration.channel.AbstractMessageChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.support.DefaultErrorMessageStrategy;
import org.springframework.integration.support.ErrorMessageStrategy;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.messaging.converter.DefaultContentTypeResolver;
import org.springframework.messaging.converter.MessageConverter;
import org.springframework.messaging.support.ChannelInterceptor;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.Assert;
import org.springframework.util.MimeType;
import org.springframework.util.MimeTypeUtils;
import org.springframework.util.StringUtils;
@ -85,9 +65,6 @@ public class RocketMQMessageChannelBinder extends
private final RocketMQBinderConfigurationProperties binderConfigurationProperties;
private final MessageCollectorImpl messageCollector = new MessageCollectorImpl();
private final ConcurrentMap<String, MessageChannel> messageChannels = new ConcurrentHashMap();
public RocketMQMessageChannelBinder(
RocketMQBinderConfigurationProperties binderConfigurationProperties,
RocketMQExtendedBindingProperties extendedBindingProperties,
@ -122,15 +99,9 @@ public class RocketMQMessageChannelBinder extends
messageHandler.setPartitioningInterceptor(partitioningInterceptor);
messageHandler.setBeanFactory(this.getApplicationContext().getBeanFactory());
messageHandler.setErrorMessageStrategy(this.getErrorMessageStrategy());
final BlockingQueue<Message<?>> queue = this.messageCollector.register(channel, extendedProducerProperties.isUseNativeEncoding());
((SubscribableChannel)channel).subscribe(queue::add);
this.messageChannels.put(destination.getName(), channel);
return messageHandler;
}
@Override
protected MessageHandler createProducerMessageHandler(ProducerDestination destination,
ExtendedProducerProperties<RocketMQProducerProperties> producerProperties,
@ -240,85 +211,4 @@ public class RocketMQMessageChannelBinder extends
public Class<? extends BinderSpecificPropertiesProvider> getExtendedPropertiesEntryClass() {
return this.extendedBindingProperties.getExtendedPropertiesEntryClass();
}
private static final class InboundMessageConvertingInterceptor implements ChannelInterceptor {
private final DefaultContentTypeResolver contentTypeResolver;
private final CompositeMessageConverterFactory converterFactory;
private InboundMessageConvertingInterceptor() {
this.contentTypeResolver = new DefaultContentTypeResolver();
this.converterFactory = new CompositeMessageConverterFactory();
}
private static boolean equalTypeAndSubType(MimeType m1, MimeType m2) {
return m1 != null && m2 != null && m1.getType().equalsIgnoreCase(m2.getType()) && m1.getSubtype().equalsIgnoreCase(m2.getSubtype());
}
public Message<?> preSend(Message<?> message, MessageChannel channel) {
Class<?> targetClass = null;
MessageConverter converter = null;
MimeType contentType = MimeType.valueOf(this.contentTypeResolver.resolve(message.getHeaders()).toString());
if (contentType != null && (equalTypeAndSubType(MessageConverterUtils.X_JAVA_SERIALIZED_OBJECT, contentType) || equalTypeAndSubType(MessageConverterUtils.X_JAVA_OBJECT, contentType))) {
message = MessageBuilder.fromMessage(message).setHeader("contentType", contentType).build();
converter = equalTypeAndSubType(MessageConverterUtils.X_JAVA_SERIALIZED_OBJECT, contentType) ? this.converterFactory.getMessageConverterForType(contentType) : this.converterFactory.getMessageConverterForAllRegistered();
String targetClassName = contentType.getParameter("type");
if (StringUtils.hasText(targetClassName)) {
try {
targetClass = Class.forName(targetClassName, false, Thread.currentThread().getContextClassLoader());
} catch (Exception var8) {
throw new IllegalStateException("Failed to determine class name for contentType: " + message.getHeaders(), var8);
}
}
}
Object payload;
if (converter != null) {
Assert.isTrue(!equalTypeAndSubType(MessageConverterUtils.X_JAVA_OBJECT, contentType) || targetClass != null, "Cannot deserialize into message since 'contentType` is not encoded with the actual target type.Consider 'application/x-java-object; type=foo.bar.MyClass'");
payload = converter.fromMessage(message, targetClass);
} else {
MimeType deserializeContentType = this.contentTypeResolver.resolve(message.getHeaders());
if (deserializeContentType == null) {
deserializeContentType = contentType;
}
payload = deserializeContentType == null ? message.getPayload() : this.deserializePayload(message.getPayload(), deserializeContentType);
}
message = MessageBuilder.withPayload(payload).copyHeaders(message.getHeaders()).setHeader("contentType", contentType).build();
return message;
}
private Object deserializePayload(Object payload, MimeType contentType) {
if (payload instanceof byte[] && ("text".equalsIgnoreCase(contentType.getType()) || equalTypeAndSubType(MimeTypeUtils.APPLICATION_JSON, contentType))) {
payload = new String((byte[])((byte[])payload), StandardCharsets.UTF_8);
}
return payload;
}
}
protected static class MessageCollectorImpl implements MessageCollector {
private final Map<MessageChannel, BlockingQueue<Message<?>>> results;
MessageCollectorImpl() {
this.results = new HashMap();
}
private BlockingQueue<Message<?>> register(MessageChannel channel, boolean useNativeEncoding) {
if (!useNativeEncoding) {
((AbstractMessageChannel)channel).addInterceptor(new InboundMessageConvertingInterceptor());
}
LinkedBlockingDeque<Message<?>> result = new LinkedBlockingDeque();
Assert.isTrue(!this.results.containsKey(channel), "Channel [" + channel + "] was already bound");
this.results.put(channel, result);
return result;
}
public BlockingQueue<Message<?>> forChannel(MessageChannel channel) {
BlockingQueue<Message<?>> queue = this.results.get(channel);
Assert.notNull(queue, "Channel [" + channel + "] was not bound by " + RocketMQMessageChannelBinder.class);
return queue;
}
}
}

5
spring-cloud-alibaba-tests/rocketmq-tests/pom.xml
Unescape Escape View File

@ -34,6 +34,11 @@
<version>4.11</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-test-support</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
<build>

55
spring-cloud-alibaba-tests/rocketmq-tests/rocketmq-stream-test/src/test/java/com/alibaba/cloud/stream/binder/rocketmq/RocketmqProduceAndConsumerTests.java
Unescape Escape View File

@ -5,9 +5,9 @@ import java.util.Map;
import java.util.concurrent.BlockingQueue;
import com.alibaba.cloud.stream.binder.rocketmq.fixture.RocketmqBinderProcessor;
import com.alibaba.cloud.stream.binder.rocketmq.support.MessageCollector;
import com.alibaba.cloud.testsupport.SpringCloudAlibaba;
import com.alibaba.cloud.testsupport.TestExtend;
import com.alibaba.fastjson.JSON;
import org.apache.rocketmq.common.message.MessageConst;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
@ -20,6 +20,7 @@ import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration;
import org.springframework.boot.autoconfigure.jdbc.DataSourceTransactionManagerAutoConfiguration;
import org.springframework.boot.autoconfigure.transaction.TransactionAutoConfiguration;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.cloud.stream.test.binder.MessageCollector;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.GenericMessage;
@ -29,19 +30,19 @@ import static com.alibaba.cloud.testsupport.Constant.TIME_OUT;
import static org.hamcrest.CoreMatchers.is;
import static org.junit.Assert.assertThat;
import static org.springframework.boot.test.context.SpringBootTest.WebEnvironment.NONE;
import static org.springframework.cloud.stream.test.matcher.MessageQueueMatcher.receivesPayloadThat;
@SpringCloudAlibaba(composeFiles = "docker/rocket-compose-test.yml", serviceName = "rocketmq-standalone")
@TestExtend(time = 6 * TIME_OUT)
@DirtiesContext
@ImportAutoConfiguration(value = {}, exclude = {
DataSourceAutoConfiguration.class,
TransactionAutoConfiguration.class,
DataSourceTransactionManagerAutoConfiguration.class})
@ImportAutoConfiguration(value = {}, exclude = { DataSourceAutoConfiguration.class,
TransactionAutoConfiguration.class,
DataSourceTransactionManagerAutoConfiguration.class })
@SpringBootTest(classes = RocketmqBinderProcessor.class, webEnvironment = NONE, properties = {
"spring.cloud.stream.rocketmq.binder.name-server=127.0.0.1:9876,127.0.0.1:9877",
"spring.cloud.stream.rocketmq.binder.group=flaky-group",
// "spring.cloud.stream.rocketmq.binder.consumer-group=flaky-group",
// "spring.cloud.stream.pollable-source=pollable",
// "spring.cloud.stream.rocketmq.binder.consumer-group=flaky-group",
// "spring.cloud.stream.pollable-source=pollable",
"spring.cloud.stream.bindings.uppercaseFunction-out-0.destination=TopicOrderTest",
"spring.cloud.stream.bindings.uppercaseFunction-out-0.content-type=application/json",
"spring.cloud.stream.bindings.uppercaseFunction-out-0.group=test-group1",
@ -49,11 +50,11 @@ import static org.springframework.boot.test.context.SpringBootTest.WebEnvironmen
"spring.cloud.stream.bindings.uppercaseFunction-in-0.content-type=application/json",
"spring.cloud.stream.bindings.uppercaseFunction-in-0.group=test-group1",
"spring.cloud.stream.bindings.uppercaseFunction-in-0.consumer.push.orderly=true",
"spring.cloud.stream.bindings.uppercaseFunction-in-0.consumer.maxAttempts=1",})
"spring.cloud.stream.bindings.uppercaseFunction-in-0.consumer.maxAttempts=1", })
public class RocketmqProduceAndConsumerTests {
private final MessageCollector collector = getInstance();
@Autowired
private MessageCollector collector;
@Autowired
@Qualifier("uppercaseFunction-in-0")
@ -64,19 +65,24 @@ public class RocketmqProduceAndConsumerTests {
private MessageChannel output;
@BeforeAll
public static void prepare(){
public static void prepare() {
}
public static RocketMQMessageChannelBinder.MessageCollectorImpl getInstance() {
return new RocketMQMessageChannelBinder.MessageCollectorImpl();
}
@BeforeEach
public void setup(){
public void setup() {
String key = "KEY";
String messageId = "1";
Map<String, Object> headers = new HashMap<>();
headers.put(MessageConst.PROPERTY_KEYS, key);
headers.put(MessageConst.PROPERTY_TAGS, "TagA");
headers.put(MessageConst.PROPERTY_ORIGIN_MESSAGE_ID, messageId);
Message<SimpleMsg> msg = new GenericMessage(new SimpleMsg("Hello RocketMQ"), headers);
Message<String> msg = new GenericMessage(JSON.toJSONString("Hello RocketMQ"),
headers);
input1.send(msg);
}
@ -84,28 +90,7 @@ public class RocketmqProduceAndConsumerTests {
public void testConsumeAndProduce() throws Exception {
BlockingQueue<Message<?>> messages = this.collector.forChannel(this.output);
assertThat(messages, is("Hello RocketMQ"));
}
public static RocketMQMessageChannelBinder.MessageCollectorImpl getInstance(){
return new RocketMQMessageChannelBinder.MessageCollectorImpl();
}
public class SimpleMsg {
private String msg;
public SimpleMsg(String msg) {
this.msg = msg;
}
public String getMsg() {
return msg;
}
public void setMsg(String msg) {
this.msg = msg;
}
assertThat(messages, receivesPayloadThat(is("\"HELLO ROCKETMQ\"")));
}
}

8
spring-cloud-alibaba-tests/rocketmq-tests/rocketmq-stream-test/src/test/java/com/alibaba/cloud/stream/binder/rocketmq/fixture/RocketmqBinderProcessor.java
Unescape Escape View File

@ -9,12 +9,12 @@ import org.springframework.context.annotation.Bean;
@SpringBootApplication
public class RocketmqBinderProcessor {
public static void main(String[] args) {
SpringApplication.run(RocketmqBinderProcessor.class, args);
}
@Bean
public Function<String, String> uppercaseFunction() {
return String::toUpperCase;
}
public static void main(String[] args) {
SpringApplication.run(RocketmqBinderProcessor.class, args);
}
}

Write Preview
Loading…
Cancel
Save