Move binders out of the main repo
Fixes spring-cloud/spring-cloud-stream#546 Fixing merge conflicts
This commit is contained in:
Eric Bottard
committed by
Soby Chacko
Soby Chacko
parent
daf454410c
commit
97f4b0f5f4
@@ -0,0 +1,210 @@
|
||||
/*
|
||||
* Copyright 2013-2016 the original author or authors.
|
||||
*
|
||||
* Licensed 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.springframework.cloud.stream.binder;
|
||||
|
||||
import java.util.UUID;
|
||||
|
||||
import org.junit.After;
|
||||
import org.junit.Test;
|
||||
|
||||
import org.springframework.integration.channel.DirectChannel;
|
||||
import org.springframework.integration.channel.QueueChannel;
|
||||
import org.springframework.integration.endpoint.AbstractEndpoint;
|
||||
import org.springframework.integration.support.MessageBuilder;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.MessageChannel;
|
||||
import org.springframework.messaging.MessageHeaders;
|
||||
import org.springframework.messaging.PollableChannel;
|
||||
import org.springframework.util.MimeTypeUtils;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
/**
|
||||
* @author Gary Russell
|
||||
* @author Ilayaperumal Gopinathan
|
||||
* @author David Turanski
|
||||
* @author Mark Fisher
|
||||
* @author Marius Bogoevici
|
||||
*/
|
||||
@SuppressWarnings("unchecked")
|
||||
public abstract class AbstractBinderTests<B extends AbstractTestBinder<? extends AbstractBinder<MessageChannel, CP, PP>, CP, PP>, CP extends ConsumerProperties, PP extends ProducerProperties> {
|
||||
|
||||
protected B testBinder;
|
||||
|
||||
/**
|
||||
* Subclasses may override this default value to have tests wait longer for a message receive, for example if
|
||||
* running
|
||||
* in an environment that is known to be slow (e.g. travis).
|
||||
*/
|
||||
protected double timeoutMultiplier = 1.0D;
|
||||
|
||||
/**
|
||||
* Attempt to receive a message on the given channel,
|
||||
* waiting up to 1s (times the {@link #timeoutMultiplier}).
|
||||
*/
|
||||
protected Message<?> receive(PollableChannel channel) {
|
||||
return receive(channel, 1);
|
||||
}
|
||||
|
||||
/**
|
||||
* Attempt to receive a message on the given channel,
|
||||
* waiting up to 1s * additionalMultiplier * {@link #timeoutMultiplier}).
|
||||
*
|
||||
* Allows accomodating tests which are slower than normal (e.g. retry).
|
||||
*/
|
||||
protected Message<?> receive(PollableChannel channel, int additionalMultiplier) {
|
||||
return channel.receive((int) (1000 * timeoutMultiplier * additionalMultiplier));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testClean() throws Exception {
|
||||
Binder binder = getBinder();
|
||||
Binding<MessageChannel> foo0ProducerBinding = binder.bindProducer("foo.0", new DirectChannel(),
|
||||
createProducerProperties());
|
||||
Binding<MessageChannel> foo0ConsumerBinding = binder.bindConsumer("foo.0", "test", new DirectChannel(),
|
||||
createConsumerProperties());
|
||||
Binding<MessageChannel> foo1ProducerBinding = binder.bindProducer("foo.1", new DirectChannel(),
|
||||
createProducerProperties());
|
||||
Binding<MessageChannel> foo1ConsumerBinding = binder.bindConsumer("foo.1", "test", new DirectChannel(),
|
||||
createConsumerProperties());
|
||||
Binding<MessageChannel> foo2ProducerBinding = binder.bindProducer("foo.2", new DirectChannel(),
|
||||
createProducerProperties());
|
||||
foo0ProducerBinding.unbind();
|
||||
assertThat(TestUtils.getPropertyValue(foo0ProducerBinding, "endpoint", AbstractEndpoint.class).isRunning())
|
||||
.isFalse();
|
||||
foo0ConsumerBinding.unbind();
|
||||
foo1ProducerBinding.unbind();
|
||||
assertThat(TestUtils.getPropertyValue(foo0ConsumerBinding, "endpoint", AbstractEndpoint.class).isRunning())
|
||||
.isFalse();
|
||||
assertThat(TestUtils.getPropertyValue(foo1ProducerBinding, "endpoint", AbstractEndpoint.class).isRunning())
|
||||
.isFalse();
|
||||
foo1ConsumerBinding.unbind();
|
||||
foo2ProducerBinding.unbind();
|
||||
assertThat(TestUtils.getPropertyValue(foo1ConsumerBinding, "endpoint", AbstractEndpoint.class).isRunning())
|
||||
.isFalse();
|
||||
assertThat(TestUtils.getPropertyValue(foo2ProducerBinding, "endpoint", AbstractEndpoint.class).isRunning())
|
||||
.isFalse();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testSendAndReceive() throws Exception {
|
||||
Binder binder = getBinder();
|
||||
DirectChannel moduleOutputChannel = new DirectChannel();
|
||||
QueueChannel moduleInputChannel = new QueueChannel();
|
||||
Binding<MessageChannel> producerBinding = binder.bindProducer("foo.0", moduleOutputChannel,
|
||||
createProducerProperties());
|
||||
Binding<MessageChannel> consumerBinding = binder.bindConsumer("foo.0", "test", moduleInputChannel,
|
||||
createConsumerProperties());
|
||||
Message<?> message = MessageBuilder.withPayload("foo").setHeader(MessageHeaders.CONTENT_TYPE, "foo/bar")
|
||||
.build();
|
||||
// Let the consumer actually bind to the producer before sending a msg
|
||||
binderBindUnbindLatency();
|
||||
moduleOutputChannel.send(message);
|
||||
Message<?> inbound = receive(moduleInputChannel);
|
||||
assertThat(inbound).isNotNull();
|
||||
assertThat(inbound.getPayload()).isEqualTo("foo");
|
||||
assertThat(inbound.getHeaders().get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE)).isNull();
|
||||
assertThat(inbound.getHeaders().get(MessageHeaders.CONTENT_TYPE)).isEqualTo("foo/bar");
|
||||
producerBinding.unbind();
|
||||
consumerBinding.unbind();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testSendAndReceiveMultipleTopics() throws Exception {
|
||||
Binder binder = getBinder();
|
||||
|
||||
DirectChannel moduleOutputChannel1 = new DirectChannel();
|
||||
DirectChannel moduleOutputChannel2 = new DirectChannel();
|
||||
|
||||
QueueChannel moduleInputChannel = new QueueChannel();
|
||||
|
||||
Binding<MessageChannel> producerBinding1 = binder.bindProducer("foo.x", moduleOutputChannel1, createProducerProperties());
|
||||
Binding<MessageChannel> producerBinding2 = binder.bindProducer("foo.y", moduleOutputChannel2, createProducerProperties());
|
||||
|
||||
Binding<MessageChannel> consumerBinding1 = binder.bindConsumer("foo.x", "test", moduleInputChannel, createConsumerProperties());
|
||||
Binding<MessageChannel> consumerBinding2 = binder.bindConsumer("foo.y", "test", moduleInputChannel, createConsumerProperties());
|
||||
|
||||
String testPayload1 = "foo" + UUID.randomUUID().toString();
|
||||
Message<?> message1 = MessageBuilder.withPayload(testPayload1.getBytes()).build();
|
||||
String testPayload2 = "foo" + UUID.randomUUID().toString();
|
||||
Message<?> message2 = MessageBuilder.withPayload(testPayload2.getBytes()).build();
|
||||
|
||||
// Let the consumer actually bind to the producer before sending a msg
|
||||
binderBindUnbindLatency();
|
||||
moduleOutputChannel1.send(message1);
|
||||
moduleOutputChannel2.send(message2);
|
||||
|
||||
|
||||
Message<?>[] messages = new Message[2];
|
||||
messages[0] = receive(moduleInputChannel);
|
||||
messages[1] = receive(moduleInputChannel);
|
||||
|
||||
assertThat(messages[0]).isNotNull();
|
||||
assertThat(messages[1]).isNotNull();
|
||||
assertThat(messages).extracting("payload").containsExactlyInAnyOrder(testPayload1.getBytes(),
|
||||
testPayload2.getBytes());
|
||||
|
||||
producerBinding1.unbind();
|
||||
producerBinding2.unbind();
|
||||
|
||||
consumerBinding1.unbind();
|
||||
consumerBinding2.unbind();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testSendAndReceiveNoOriginalContentType() throws Exception {
|
||||
Binder binder = getBinder();
|
||||
|
||||
DirectChannel moduleOutputChannel = new DirectChannel();
|
||||
QueueChannel moduleInputChannel = new QueueChannel();
|
||||
Binding<MessageChannel> producerBinding = binder.bindProducer("bar.0", moduleOutputChannel, createProducerProperties());
|
||||
Binding<MessageChannel> consumerBinding = binder.bindConsumer("bar.0", "test", moduleInputChannel, createConsumerProperties());
|
||||
binderBindUnbindLatency();
|
||||
|
||||
Message<?> message = MessageBuilder.withPayload("foo").build();
|
||||
moduleOutputChannel.send(message);
|
||||
Message<?> inbound = receive(moduleInputChannel);
|
||||
assertThat(inbound).isNotNull();
|
||||
assertThat(inbound.getPayload()).isEqualTo("foo");
|
||||
assertThat(inbound.getHeaders().get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE)).isNull();
|
||||
assertThat(inbound.getHeaders().get(MessageHeaders.CONTENT_TYPE)).isEqualTo(MimeTypeUtils.TEXT_PLAIN_VALUE);
|
||||
producerBinding.unbind();
|
||||
consumerBinding.unbind();
|
||||
}
|
||||
|
||||
|
||||
protected abstract B getBinder() throws Exception;
|
||||
|
||||
protected abstract CP createConsumerProperties();
|
||||
|
||||
protected abstract PP createProducerProperties();
|
||||
|
||||
@After
|
||||
public void cleanup() {
|
||||
if (testBinder != null) {
|
||||
testBinder.cleanup();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* If appropriate, let the binder middleware settle down a bit while binding/unbinding actually happens.
|
||||
*/
|
||||
protected void binderBindUnbindLatency() throws InterruptedException {
|
||||
// default none
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,69 @@
|
||||
/*
|
||||
* Copyright 2014-2016 the original author or authors.
|
||||
*
|
||||
* Licensed 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.springframework.cloud.stream.binder;
|
||||
|
||||
import java.util.HashSet;
|
||||
import java.util.Set;
|
||||
|
||||
import org.springframework.messaging.MessageChannel;
|
||||
|
||||
/**
|
||||
* Abstract class that adds test support for {@link Binder}.
|
||||
*
|
||||
* @author Ilayaperumal Gopinathan
|
||||
* @author Gary Russell
|
||||
* @author Mark Fisher
|
||||
*/
|
||||
public abstract class AbstractTestBinder<C extends AbstractBinder<MessageChannel, CP, PP>, CP extends ConsumerProperties, PP extends ProducerProperties> implements Binder<MessageChannel, CP, PP> {
|
||||
|
||||
protected Set<String> queues = new HashSet<String>();
|
||||
|
||||
private C binder;
|
||||
|
||||
public void setBinder(C binder) {
|
||||
try {
|
||||
binder.afterPropertiesSet();
|
||||
}
|
||||
catch (Exception e) {
|
||||
throw new RuntimeException("Failed to initialize binder", e);
|
||||
}
|
||||
this.binder = binder;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Binding<MessageChannel> bindConsumer(String name, String group, MessageChannel moduleInputChannel, CP properties) {
|
||||
queues.add(name);
|
||||
return binder.bindConsumer(name, group, moduleInputChannel, properties);
|
||||
}
|
||||
|
||||
@Override
|
||||
public Binding<MessageChannel> bindProducer(String name, MessageChannel moduleOutputChannel, PP properties) {
|
||||
queues.add(name);
|
||||
return binder.bindProducer(name, moduleOutputChannel, properties);
|
||||
}
|
||||
|
||||
public C getCoreBinder() {
|
||||
return binder;
|
||||
}
|
||||
|
||||
public abstract void cleanup();
|
||||
|
||||
public C getBinder() {
|
||||
return this.binder;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,49 @@
|
||||
/*
|
||||
* Copyright 2014 the original author or authors.
|
||||
*
|
||||
* Licensed 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.springframework.cloud.stream.binder;
|
||||
|
||||
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
|
||||
import org.springframework.context.support.AbstractApplicationContext;
|
||||
import org.springframework.integration.support.DefaultMessageBuilderFactory;
|
||||
import org.springframework.integration.support.MessageBuilderFactory;
|
||||
import org.springframework.integration.support.utils.IntegrationUtils;
|
||||
|
||||
import static org.mockito.Mockito.mock;
|
||||
import static org.mockito.Mockito.when;
|
||||
|
||||
|
||||
/**
|
||||
*
|
||||
* @author Gary Russell
|
||||
*/
|
||||
public abstract class BinderTestUtils {
|
||||
|
||||
private static final MessageBuilderFactory mbf = new DefaultMessageBuilderFactory();
|
||||
|
||||
public static final AbstractApplicationContext MOCK_AC = mock(AbstractApplicationContext.class);
|
||||
|
||||
public static final ConfigurableListableBeanFactory MOCK_BF = mock(ConfigurableListableBeanFactory.class);
|
||||
|
||||
static {
|
||||
when(MOCK_BF.getBean(IntegrationUtils.INTEGRATION_MESSAGE_BUILDER_FACTORY_BEAN_NAME,
|
||||
MessageBuilderFactory.class)).thenReturn(mbf);
|
||||
when(MOCK_AC.getBean(IntegrationUtils.INTEGRATION_MESSAGE_BUILDER_FACTORY_BEAN_NAME,
|
||||
MessageBuilderFactory.class)).thenReturn(mbf);
|
||||
when(MOCK_AC.getBeanFactory()).thenReturn(MOCK_BF);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,36 @@
|
||||
/*
|
||||
* Copyright 2014-2016 the original author or authors.
|
||||
*
|
||||
* Licensed 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.springframework.cloud.stream.binder;
|
||||
|
||||
import org.springframework.messaging.MessageChannel;
|
||||
|
||||
/**
|
||||
* Tests for binders that use an external broker.
|
||||
*
|
||||
* @author Gary Russell
|
||||
*/
|
||||
public abstract class BrokerBinderTests<B extends AbstractTestBinder<? extends AbstractBinder<MessageChannel, CP, PP>, CP, PP>, CP extends ConsumerProperties, PP extends ProducerProperties>
|
||||
extends AbstractBinderTests<B, CP, PP> {
|
||||
|
||||
/**
|
||||
* Create a new spy on the given 'queue'. This allows de-correlating the creation of
|
||||
* the 'connection' from its actual usage, which may be needed by some implementations
|
||||
* to see messages sent after connection creation.
|
||||
*/
|
||||
public abstract Spy spyOn(final String name);
|
||||
|
||||
}
|
||||
@@ -0,0 +1,348 @@
|
||||
/*
|
||||
* Copyright 2014-2016 the original author or authors.
|
||||
*
|
||||
* Licensed 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.springframework.cloud.stream.binder;
|
||||
|
||||
import java.util.Arrays;
|
||||
import java.util.List;
|
||||
import java.util.UUID;
|
||||
|
||||
import org.assertj.core.api.Condition;
|
||||
import org.junit.Test;
|
||||
|
||||
import org.springframework.beans.DirectFieldAccessor;
|
||||
import org.springframework.expression.spel.standard.SpelExpressionParser;
|
||||
import org.springframework.integration.IntegrationMessageHeaderAccessor;
|
||||
import org.springframework.integration.channel.DirectChannel;
|
||||
import org.springframework.integration.channel.QueueChannel;
|
||||
import org.springframework.integration.endpoint.AbstractEndpoint;
|
||||
import org.springframework.integration.support.MessageBuilder;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.MessageChannel;
|
||||
import org.springframework.messaging.support.GenericMessage;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
/**
|
||||
* Tests for binders that support partitioning.
|
||||
*
|
||||
* @author Gary Russell
|
||||
* @author Mark Fisher
|
||||
* @author Marius Bogoevici
|
||||
*/
|
||||
abstract public class PartitionCapableBinderTests<B extends AbstractTestBinder<? extends AbstractBinder<MessageChannel, CP, PP>, CP, PP>, CP extends ConsumerProperties, PP extends ProducerProperties>
|
||||
extends BrokerBinderTests<B, CP, PP> {
|
||||
|
||||
protected static final SpelExpressionParser spelExpressionParser = new SpelExpressionParser();
|
||||
|
||||
@Test
|
||||
@SuppressWarnings("unchecked")
|
||||
public void testAnonymousGroup() throws Exception {
|
||||
B binder = getBinder();
|
||||
DirectChannel output = new DirectChannel();
|
||||
Binding<MessageChannel> producerBinding = binder.bindProducer("defaultGroup.0", output,
|
||||
createProducerProperties());
|
||||
|
||||
QueueChannel input1 = new QueueChannel();
|
||||
Binding<MessageChannel> binding1 = binder.bindConsumer("defaultGroup.0", null, input1,
|
||||
createConsumerProperties());
|
||||
|
||||
QueueChannel input2 = new QueueChannel();
|
||||
Binding<MessageChannel> binding2 = binder.bindConsumer("defaultGroup.0", null, input2,
|
||||
createConsumerProperties());
|
||||
|
||||
String testPayload1 = "foo-" + UUID.randomUUID().toString();
|
||||
output.send(new GenericMessage<>(testPayload1.getBytes()));
|
||||
|
||||
Message<byte[]> receivedMessage1 = (Message<byte[]>) receive(input1);
|
||||
assertThat(receivedMessage1).isNotNull();
|
||||
assertThat(new String(receivedMessage1.getPayload())).isEqualTo(testPayload1);
|
||||
|
||||
Message<byte[]> receivedMessage2 = (Message<byte[]>) receive(input2);
|
||||
assertThat(receivedMessage2).isNotNull();
|
||||
assertThat(new String(receivedMessage2.getPayload())).isEqualTo(testPayload1);
|
||||
|
||||
binding2.unbind();
|
||||
|
||||
String testPayload2 = "foo-" + UUID.randomUUID().toString();
|
||||
output.send(new GenericMessage<>(testPayload2.getBytes()));
|
||||
|
||||
binding2 = binder.bindConsumer("defaultGroup.0", null, input2, createConsumerProperties());
|
||||
String testPayload3 = "foo-" + UUID.randomUUID().toString();
|
||||
output.send(new GenericMessage<>(testPayload3.getBytes()));
|
||||
|
||||
receivedMessage1 = (Message<byte[]>) receive(input1);
|
||||
assertThat(receivedMessage1).isNotNull();
|
||||
assertThat(new String(receivedMessage1.getPayload())).isEqualTo(testPayload2);
|
||||
receivedMessage1 = (Message<byte[]>) receive(input1);
|
||||
assertThat(receivedMessage1).isNotNull();
|
||||
assertThat(new String(receivedMessage1.getPayload())).isNotNull();
|
||||
|
||||
receivedMessage2 = (Message<byte[]>) receive(input2);
|
||||
assertThat(receivedMessage2).isNotNull();
|
||||
assertThat(new String(receivedMessage2.getPayload())).isEqualTo(testPayload3);
|
||||
|
||||
producerBinding.unbind();
|
||||
binding1.unbind();
|
||||
binding2.unbind();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testOneRequiredGroup() throws Exception {
|
||||
B binder = getBinder();
|
||||
DirectChannel output = new DirectChannel();
|
||||
|
||||
PP producerProperties = createProducerProperties();
|
||||
|
||||
String testDestination = "testDestination" + UUID.randomUUID().toString().replace("-", "");
|
||||
|
||||
producerProperties.setRequiredGroups("test1");
|
||||
Binding<MessageChannel> producerBinding = binder.bindProducer(testDestination, output, producerProperties);
|
||||
|
||||
String testPayload = "foo-" + UUID.randomUUID().toString();
|
||||
output.send(new GenericMessage<>(testPayload.getBytes()));
|
||||
|
||||
QueueChannel inbound1 = new QueueChannel();
|
||||
Binding<MessageChannel> consumerBinding = binder.bindConsumer(testDestination, "test1", inbound1,
|
||||
createConsumerProperties());
|
||||
|
||||
Message<?> receivedMessage1 = receive(inbound1);
|
||||
assertThat(receivedMessage1).isNotNull();
|
||||
assertThat(new String((byte[]) receivedMessage1.getPayload())).isEqualTo(testPayload);
|
||||
|
||||
producerBinding.unbind();
|
||||
consumerBinding.unbind();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testTwoRequiredGroups() throws Exception {
|
||||
B binder = getBinder();
|
||||
DirectChannel output = new DirectChannel();
|
||||
|
||||
String testDestination = "testDestination" + UUID.randomUUID().toString().replace("-", "");
|
||||
|
||||
PP producerProperties = createProducerProperties();
|
||||
producerProperties.setRequiredGroups("test1", "test2");
|
||||
Binding<MessageChannel> producerBinding = binder.bindProducer(testDestination, output, producerProperties);
|
||||
|
||||
String testPayload = "foo-" + UUID.randomUUID().toString();
|
||||
output.send(new GenericMessage<>(testPayload.getBytes()));
|
||||
|
||||
QueueChannel inbound1 = new QueueChannel();
|
||||
Binding<MessageChannel> consumerBinding1 = binder.bindConsumer(testDestination, "test1", inbound1,
|
||||
createConsumerProperties());
|
||||
QueueChannel inbound2 = new QueueChannel();
|
||||
Binding<MessageChannel> consumerBinding2 = binder.bindConsumer(testDestination, "test2", inbound2,
|
||||
createConsumerProperties());
|
||||
|
||||
Message<?> receivedMessage1 = receive(inbound1);
|
||||
assertThat(receivedMessage1).isNotNull();
|
||||
assertThat(new String((byte[]) receivedMessage1.getPayload())).isEqualTo(testPayload);
|
||||
Message<?> receivedMessage2 = receive(inbound2);
|
||||
assertThat(receivedMessage2).isNotNull();
|
||||
assertThat(new String((byte[]) receivedMessage2.getPayload())).isEqualTo(testPayload);
|
||||
|
||||
consumerBinding1.unbind();
|
||||
consumerBinding2.unbind();
|
||||
producerBinding.unbind();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testPartitionedModuleSpEL() throws Exception {
|
||||
B binder = getBinder();
|
||||
|
||||
CP consumerProperties = createConsumerProperties();
|
||||
consumerProperties.setConcurrency(2);
|
||||
consumerProperties.setInstanceIndex(0);
|
||||
consumerProperties.setInstanceCount(3);
|
||||
consumerProperties.setPartitioned(true);
|
||||
QueueChannel input0 = new QueueChannel();
|
||||
input0.setBeanName("test.input0S");
|
||||
Binding<MessageChannel> input0Binding = binder.bindConsumer("part.0", "test", input0, consumerProperties);
|
||||
consumerProperties.setInstanceIndex(1);
|
||||
QueueChannel input1 = new QueueChannel();
|
||||
input1.setBeanName("test.input1S");
|
||||
Binding<MessageChannel> input1Binding = binder.bindConsumer("part.0", "test", input1, consumerProperties);
|
||||
consumerProperties.setInstanceIndex(2);
|
||||
QueueChannel input2 = new QueueChannel();
|
||||
input2.setBeanName("test.input2S");
|
||||
Binding<MessageChannel> input2Binding = binder.bindConsumer("part.0", "test", input2, consumerProperties);
|
||||
|
||||
PP producerProperties = createProducerProperties();
|
||||
producerProperties.setPartitionKeyExpression(spelExpressionParser.parseExpression("payload"));
|
||||
producerProperties.setPartitionSelectorExpression(spelExpressionParser.parseExpression("hashCode()"));
|
||||
producerProperties.setPartitionCount(3);
|
||||
|
||||
DirectChannel output = new DirectChannel();
|
||||
output.setBeanName("test.output");
|
||||
Binding<MessageChannel> outputBinding = binder.bindProducer("part.0", output, producerProperties);
|
||||
try {
|
||||
AbstractEndpoint endpoint = extractEndpoint(outputBinding);
|
||||
assertThat(getEndpointRouting(endpoint))
|
||||
.contains(getExpectedRoutingBaseDestination("part.0", "test") + "-' + headers['partition']");
|
||||
}
|
||||
catch (UnsupportedOperationException ignored) {
|
||||
}
|
||||
|
||||
Message<Integer> message2 = MessageBuilder.withPayload(2)
|
||||
.setHeader(IntegrationMessageHeaderAccessor.CORRELATION_ID, "foo")
|
||||
.setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER, 42)
|
||||
.setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE, 43).build();
|
||||
output.send(message2);
|
||||
output.send(new GenericMessage<>(1));
|
||||
output.send(new GenericMessage<>(0));
|
||||
|
||||
Message<?> receive0 = receive(input0);
|
||||
assertThat(receive0).isNotNull();
|
||||
Message<?> receive1 = receive(input1);
|
||||
assertThat(receive1).isNotNull();
|
||||
Message<?> receive2 = receive(input2);
|
||||
assertThat(receive2).isNotNull();
|
||||
|
||||
Condition<Message<?>> correlationHeadersForPayload2 = new Condition<Message<?>>() {
|
||||
@Override
|
||||
public boolean matches(Message<?> value) {
|
||||
IntegrationMessageHeaderAccessor accessor = new IntegrationMessageHeaderAccessor(value);
|
||||
return "foo".equals(accessor.getCorrelationId()) && 42 == accessor.getSequenceNumber()
|
||||
&& 43 == accessor.getSequenceSize();
|
||||
}
|
||||
};
|
||||
|
||||
if (usesExplicitRouting()) {
|
||||
assertThat(receive0.getPayload()).isEqualTo(0);
|
||||
assertThat(receive1.getPayload()).isEqualTo(1);
|
||||
assertThat(receive2.getPayload()).isEqualTo(2);
|
||||
assertThat(receive2).has(correlationHeadersForPayload2);
|
||||
}
|
||||
else {
|
||||
List<Message<?>> receivedMessages = Arrays.asList(receive0, receive1, receive2);
|
||||
assertThat(receivedMessages).extracting("payload").containsExactlyInAnyOrder(0, 1, 2);
|
||||
Condition<Message<?>> payloadIs2 = new Condition<Message<?>>() {
|
||||
|
||||
@Override
|
||||
public boolean matches(Message<?> value) {
|
||||
return value.getPayload().equals(2);
|
||||
}
|
||||
};
|
||||
assertThat(receivedMessages).filteredOn(payloadIs2).areExactly(1, correlationHeadersForPayload2);
|
||||
|
||||
}
|
||||
input0Binding.unbind();
|
||||
input1Binding.unbind();
|
||||
input2Binding.unbind();
|
||||
outputBinding.unbind();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testPartitionedModuleJava() throws Exception {
|
||||
B binder = getBinder();
|
||||
|
||||
CP consumerProperties = createConsumerProperties();
|
||||
consumerProperties.setConcurrency(2);
|
||||
consumerProperties.setInstanceCount(3);
|
||||
consumerProperties.setInstanceIndex(0);
|
||||
consumerProperties.setPartitioned(true);
|
||||
QueueChannel input0 = new QueueChannel();
|
||||
input0.setBeanName("test.input0J");
|
||||
Binding<MessageChannel> input0Binding = binder.bindConsumer("partJ.0", "test", input0, consumerProperties);
|
||||
consumerProperties.setInstanceIndex(1);
|
||||
QueueChannel input1 = new QueueChannel();
|
||||
input1.setBeanName("test.input1J");
|
||||
Binding<MessageChannel> input1Binding = binder.bindConsumer("partJ.0", "test", input1, consumerProperties);
|
||||
consumerProperties.setInstanceIndex(2);
|
||||
QueueChannel input2 = new QueueChannel();
|
||||
input2.setBeanName("test.input2J");
|
||||
Binding<MessageChannel> input2Binding = binder.bindConsumer("partJ.0", "test", input2, consumerProperties);
|
||||
|
||||
PP producerProperties = createProducerProperties();
|
||||
producerProperties.setPartitionKeyExtractorClass(PartitionTestSupport.class);
|
||||
producerProperties.setPartitionSelectorClass(PartitionTestSupport.class);
|
||||
producerProperties.setPartitionCount(3);
|
||||
DirectChannel output = new DirectChannel();
|
||||
output.setBeanName("test.output");
|
||||
Binding<MessageChannel> outputBinding = binder.bindProducer("partJ.0", output, producerProperties);
|
||||
if (usesExplicitRouting()) {
|
||||
AbstractEndpoint endpoint = extractEndpoint(outputBinding);
|
||||
assertThat(getEndpointRouting(endpoint)).
|
||||
contains(getExpectedRoutingBaseDestination("partJ.0", "test") + "-' + headers['partition']");
|
||||
}
|
||||
|
||||
output.send(new GenericMessage<>(2));
|
||||
output.send(new GenericMessage<>(1));
|
||||
output.send(new GenericMessage<>(0));
|
||||
|
||||
Message<?> receive0 = receive(input0);
|
||||
assertThat(receive0).isNotNull();
|
||||
Message<?> receive1 = receive(input1);
|
||||
assertThat(receive1).isNotNull();
|
||||
Message<?> receive2 = receive(input2);
|
||||
assertThat(receive2).isNotNull();
|
||||
|
||||
if (usesExplicitRouting()) {
|
||||
assertThat(receive0.getPayload()).isEqualTo(0);
|
||||
assertThat(receive1.getPayload()).isEqualTo(1);
|
||||
assertThat(receive2.getPayload()).isEqualTo(2);
|
||||
}
|
||||
else {
|
||||
List<Message<?>> receivedMessages = Arrays.asList(receive0, receive1, receive2);
|
||||
assertThat(receivedMessages).extracting("payload").containsExactlyInAnyOrder(0, 1, 2);
|
||||
}
|
||||
|
||||
input0Binding.unbind();
|
||||
input1Binding.unbind();
|
||||
input2Binding.unbind();
|
||||
outputBinding.unbind();
|
||||
}
|
||||
|
||||
/**
|
||||
* Implementations should return whether the binder under test uses "explicit" routing
|
||||
* (e.g. Rabbit) whereby Spring Cloud Stream is responsible for assigning a partition
|
||||
* and knows which exact consumer will receive the message (i.e. honor
|
||||
* "partitionIndex") or "implicit" routing (e.g. Kafka) whereby the only guarantee is
|
||||
* that messages will be spread, but we don't control exactly which consumer gets
|
||||
* which message.
|
||||
*/
|
||||
protected abstract boolean usesExplicitRouting();
|
||||
|
||||
/**
|
||||
* For implementations that rely on explicit routing, return the routing expression.
|
||||
*/
|
||||
protected String getEndpointRouting(AbstractEndpoint endpoint) {
|
||||
throw new UnsupportedOperationException();
|
||||
}
|
||||
|
||||
/**
|
||||
* For implementations that rely on explicit routing, return the expected base
|
||||
* destination (the part that precedes '-partition' within the expression).
|
||||
*/
|
||||
protected String getExpectedRoutingBaseDestination(String name, String group) {
|
||||
throw new UnsupportedOperationException();
|
||||
}
|
||||
|
||||
/**
|
||||
* For implementations that rely on explicit routing, return the routing expression.
|
||||
*/
|
||||
protected String getPubSubEndpointRouting(AbstractEndpoint endpoint) {
|
||||
throw new UnsupportedOperationException();
|
||||
}
|
||||
|
||||
protected abstract String getClassUnderTestName();
|
||||
|
||||
protected AbstractEndpoint extractEndpoint(Binding<MessageChannel> binding) {
|
||||
DirectFieldAccessor accessor = new DirectFieldAccessor(binding);
|
||||
return (AbstractEndpoint) accessor.getPropertyValue("endpoint");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,37 @@
|
||||
/*
|
||||
* Copyright 2014 the original author or authors.
|
||||
*
|
||||
* Licensed 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.springframework.cloud.stream.binder;
|
||||
|
||||
import org.springframework.messaging.Message;
|
||||
|
||||
|
||||
/**
|
||||
* @author Gary Russell
|
||||
*/
|
||||
public class PartitionTestSupport implements PartitionKeyExtractorStrategy, PartitionSelectorStrategy {
|
||||
|
||||
@Override
|
||||
public int selectPartition(Object key, int divisor) {
|
||||
return key.hashCode() % divisor;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Object extractKey(Message<?> message) {
|
||||
return message.getPayload();
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,29 @@
|
||||
/*
|
||||
* Copyright 2015 the original author or authors.
|
||||
*
|
||||
* Licensed 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.springframework.cloud.stream.binder;
|
||||
|
||||
/**
|
||||
* Represents an out-of-band connection to the underlying middleware,
|
||||
* so that tests can check that some messages actually do (or do not)
|
||||
* transit through it.
|
||||
*
|
||||
* @author Eric Bottard
|
||||
*/
|
||||
public interface Spy {
|
||||
|
||||
Object receive(boolean expectNull) throws Exception;
|
||||
}
|
||||
@@ -0,0 +1,64 @@
|
||||
/*
|
||||
* Copyright 2015 the original author or authors.
|
||||
*
|
||||
* Licensed 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.springframework.cloud.stream.binder;
|
||||
|
||||
import org.springframework.beans.DirectFieldAccessor;
|
||||
import org.springframework.util.Assert;
|
||||
|
||||
/**
|
||||
* Copy of class in org.springframework.amqp.utils.test to avoid dependency on spring-amqp
|
||||
*/
|
||||
public abstract class TestUtils {
|
||||
|
||||
/**
|
||||
* Uses nested {@link DirectFieldAccessor}s to obtain a property using dotted notation
|
||||
* to traverse fields; e.g. "foo.bar.baz" will obtain a reference to the baz field of
|
||||
* the bar field of foo. Adopted from Spring Integration.
|
||||
* @param root The object.
|
||||
* @param propertyPath The path.
|
||||
* @return The field.
|
||||
*/
|
||||
public static Object getPropertyValue(Object root, String propertyPath) {
|
||||
Object value = null;
|
||||
DirectFieldAccessor accessor = new DirectFieldAccessor(root);
|
||||
String[] tokens = propertyPath.split("\\.");
|
||||
for (int i = 0; i < tokens.length; i++) {
|
||||
value = accessor.getPropertyValue(tokens[i]);
|
||||
if (value != null) {
|
||||
accessor = new DirectFieldAccessor(value);
|
||||
}
|
||||
else if (i == tokens.length - 1) {
|
||||
return null;
|
||||
}
|
||||
else {
|
||||
throw new IllegalArgumentException(
|
||||
"intermediate property '" + tokens[i] + "' is null");
|
||||
}
|
||||
}
|
||||
return value;
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
public static <T> T getPropertyValue(Object root, String propertyPath,
|
||||
Class<T> type) {
|
||||
Object value = getPropertyValue(root, propertyPath);
|
||||
if (value != null) {
|
||||
Assert.isAssignable(type, value.getClass());
|
||||
}
|
||||
return (T) value;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,12 @@
|
||||
<configuration>
|
||||
<appender name="stdout" class="ch.qos.logback.core.ConsoleAppender">
|
||||
<encoder>
|
||||
<pattern>%d{ISO8601} %5p %t %c{2}:%L - %m%n</pattern>
|
||||
</encoder>
|
||||
</appender>
|
||||
<logger name="org.springframework.web.client.RestTemplate" level="ERROR"/>
|
||||
<logger name="org.apache.hadoop.util.NativeCodeLoader" level="ERROR"/>
|
||||
<root level="WARN">
|
||||
<appender-ref ref="stdout"/>
|
||||
</root>
|
||||
</configuration>
|
||||
@@ -0,0 +1,283 @@
|
||||
/*
|
||||
* Copyright 2013-2016 the original author or authors.
|
||||
*
|
||||
* Licensed 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.springframework.cloud.stream.binder;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
|
||||
import com.esotericsoftware.kryo.Kryo;
|
||||
import com.esotericsoftware.kryo.Registration;
|
||||
import org.junit.Before;
|
||||
import org.junit.Test;
|
||||
|
||||
import org.springframework.cloud.stream.binder.AbstractBinder.JavaClassMimeTypeConversion;
|
||||
import org.springframework.integration.codec.kryo.KryoRegistrar;
|
||||
import org.springframework.integration.codec.kryo.PojoCodec;
|
||||
import org.springframework.integration.support.MessageBuilder;
|
||||
import org.springframework.integration.tuple.TupleKryoRegistrar;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.MessageChannel;
|
||||
import org.springframework.messaging.MessageHeaders;
|
||||
import org.springframework.messaging.converter.ContentTypeResolver;
|
||||
import org.springframework.messaging.support.GenericMessage;
|
||||
import org.springframework.tuple.DefaultTuple;
|
||||
import org.springframework.tuple.Tuple;
|
||||
import org.springframework.tuple.TupleBuilder;
|
||||
import org.springframework.util.MimeType;
|
||||
import org.springframework.util.MimeTypeUtils;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
|
||||
/**
|
||||
* @author Gary Russell
|
||||
* @author David Turanski
|
||||
* @author Ilayaperumal Gopinathan
|
||||
*/
|
||||
public class MessageChannelBinderSupportTests {
|
||||
|
||||
private final ContentTypeResolver contentTypeResolver = new StringConvertingContentTypeResolver();
|
||||
|
||||
private final TestMessageChannelBinder binder = new TestMessageChannelBinder();
|
||||
|
||||
@Before
|
||||
public void setUp() {
|
||||
binder.setCodec(new PojoCodec(new TupleRegistrar()));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testBytesPassThru() {
|
||||
byte[] payload = "foo".getBytes();
|
||||
Message<byte[]> message = MessageBuilder.withPayload(payload).build();
|
||||
MessageValues converted = binder.serializePayloadIfNecessary(message);
|
||||
assertThat(converted.getPayload()).isSameAs(payload);
|
||||
Message<?> convertedMessage = converted.toMessage();
|
||||
assertThat(convertedMessage.getPayload()).isSameAs(payload);
|
||||
assertThat(contentTypeResolver.resolve(convertedMessage.getHeaders()))
|
||||
.isEqualTo(MimeTypeUtils.APPLICATION_OCTET_STREAM);
|
||||
MessageValues reconstructed = binder.deserializePayloadIfNecessary(convertedMessage);
|
||||
payload = (byte[]) reconstructed.getPayload();
|
||||
assertThat(converted.getPayload()).isSameAs(payload);
|
||||
assertThat(reconstructed.get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE)).isNull();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testBytesPassThruContentType() {
|
||||
byte[] payload = "foo".getBytes();
|
||||
Message<byte[]> message = MessageBuilder.withPayload(payload)
|
||||
.setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_OCTET_STREAM_VALUE).build();
|
||||
MessageValues messageValues = binder.serializePayloadIfNecessary(message);
|
||||
Message<?> converted = messageValues.toMessage();
|
||||
assertThat(converted.getPayload()).isSameAs(payload);
|
||||
assertThat(contentTypeResolver.resolve(converted.getHeaders()))
|
||||
.isEqualTo(MimeTypeUtils.APPLICATION_OCTET_STREAM);
|
||||
MessageValues reconstructed = binder.deserializePayloadIfNecessary(converted);
|
||||
payload = (byte[]) reconstructed.getPayload();
|
||||
assertThat(converted.getPayload()).isSameAs(payload);
|
||||
assertThat(reconstructed.get(MessageHeaders.CONTENT_TYPE))
|
||||
.isEqualTo(MimeTypeUtils.APPLICATION_OCTET_STREAM_VALUE);
|
||||
assertThat(reconstructed.get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE)).isNull();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testString() throws IOException {
|
||||
MessageValues convertedValues = binder.serializePayloadIfNecessary(new GenericMessage<>("foo"));
|
||||
Message<?> converted = convertedValues.toMessage();
|
||||
assertThat(contentTypeResolver.resolve(converted.getHeaders())).isEqualTo(MimeTypeUtils.TEXT_PLAIN);
|
||||
MessageValues reconstructed = binder.deserializePayloadIfNecessary(converted);
|
||||
assertThat(reconstructed.getPayload()).isEqualTo("foo");
|
||||
assertThat(reconstructed.get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE)).isNull();
|
||||
assertThat(reconstructed.get(MessageHeaders.CONTENT_TYPE)).isEqualTo(MimeTypeUtils.TEXT_PLAIN_VALUE);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testStringXML() throws IOException {
|
||||
Message<?> message = MessageBuilder
|
||||
.withPayload("<?xml version=\"1.0\" encoding=\"UTF-8\" standalone=\"no\"?><test></test>")
|
||||
.setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_XML).build();
|
||||
Message<?> converted = binder.serializePayloadIfNecessary(message).toMessage();
|
||||
assertThat(contentTypeResolver.resolve(converted.getHeaders())).isEqualTo(MimeTypeUtils.TEXT_PLAIN);
|
||||
MessageValues reconstructed = binder.deserializePayloadIfNecessary(converted);
|
||||
assertThat(reconstructed.getPayload())
|
||||
.isEqualTo("<?xml version=\"1.0\" encoding=\"UTF-8\" standalone=\"no\"?><test></test>");
|
||||
assertThat(reconstructed.get(MessageHeaders.CONTENT_TYPE)).isEqualTo(MimeTypeUtils.TEXT_XML.toString());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testContentTypePreservedForJson() throws IOException {
|
||||
Message<String> inbound = MessageBuilder.withPayload("{\"foo\":\"foo\"}")
|
||||
.copyHeaders(Collections.singletonMap(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON))
|
||||
.build();
|
||||
MessageValues convertedValues = binder.serializePayloadIfNecessary(inbound);
|
||||
Message<?> converted = convertedValues.toMessage();
|
||||
assertThat(contentTypeResolver.resolve(converted.getHeaders())).isEqualTo(MimeTypeUtils.APPLICATION_JSON);
|
||||
assertThat(converted.getHeaders().get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE)).isNull();
|
||||
MessageValues reconstructed = binder.deserializePayloadIfNecessary(converted);
|
||||
assertThat(reconstructed.getPayload()).isEqualTo("{\"foo\":\"foo\"}");
|
||||
assertThat(reconstructed.get(MessageHeaders.CONTENT_TYPE)).isEqualTo(MimeTypeUtils.APPLICATION_JSON_VALUE);
|
||||
}
|
||||
|
||||
|
||||
@Test
|
||||
public void testContentTypePreservedForNonSCStApp() {
|
||||
Message<String> inbound = MessageBuilder.withPayload("{\"foo\":\"bar\"}")
|
||||
.copyHeaders(Collections.singletonMap(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON))
|
||||
.build();
|
||||
MessageValues reconstructed = binder.deserializePayloadIfNecessary(inbound);
|
||||
assertThat(reconstructed.getPayload()).isEqualTo("{\"foo\":\"bar\"}");
|
||||
assertThat(reconstructed.get(MessageHeaders.CONTENT_TYPE)).isEqualTo(MimeTypeUtils.APPLICATION_JSON);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testPojoSerialization() {
|
||||
MessageValues convertedValues = binder.serializePayloadIfNecessary(new GenericMessage<>(new Foo("bar")));
|
||||
Message<?> converted = convertedValues.toMessage();
|
||||
MimeType mimeType = contentTypeResolver.resolve(converted.getHeaders());
|
||||
assertThat(mimeType.getType()).isEqualTo("application");
|
||||
assertThat(mimeType.getSubtype()).isEqualTo("x-java-object");
|
||||
assertThat(mimeType.getParameter("type")).isEqualTo(Foo.class.getName());
|
||||
|
||||
MessageValues reconstructed = binder.deserializePayloadIfNecessary(converted);
|
||||
assertThat(((Foo) reconstructed.getPayload()).getBar()).isEqualTo("bar");
|
||||
assertThat(reconstructed.get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE)).isNull();
|
||||
assertThat(reconstructed.get(MessageHeaders.CONTENT_TYPE)).isEqualTo(
|
||||
"application/x-java-object;type=org.springframework.cloud.stream.binder.MessageChannelBinderSupportTests$Foo");
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testTupleSerialization() {
|
||||
Tuple payload = TupleBuilder.tuple().of("foo", "bar");
|
||||
MessageValues convertedValues = binder.serializePayloadIfNecessary(new GenericMessage<>(payload));
|
||||
Message<?> converted = convertedValues.toMessage();
|
||||
MimeType mimeType = contentTypeResolver.resolve(converted.getHeaders());
|
||||
assertThat(mimeType.getType()).isEqualTo("application");
|
||||
assertThat(mimeType.getSubtype()).isEqualTo("x-java-object");
|
||||
assertThat(mimeType.getParameter("type")).isEqualTo(DefaultTuple.class.getName());
|
||||
|
||||
MessageValues reconstructed = binder.deserializePayloadIfNecessary(converted);
|
||||
assertThat(((Tuple) reconstructed.getPayload()).getString("foo")).isEqualTo("bar");
|
||||
assertThat(reconstructed.get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE)).isNull();
|
||||
assertThat(reconstructed.get(MessageHeaders.CONTENT_TYPE))
|
||||
.isEqualTo("application/x-java-object;type=org.springframework.tuple.DefaultTuple");
|
||||
}
|
||||
|
||||
@Test
|
||||
public void mimeTypeIsSimpleObject() throws ClassNotFoundException {
|
||||
MimeType mt = JavaClassMimeTypeConversion.mimeTypeFromObject(new Object(), null);
|
||||
String className = JavaClassMimeTypeConversion.classNameFromMimeType(mt);
|
||||
assertThat(Class.forName(className)).isEqualTo(Object.class);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void mimeTypeIsObjectArray() throws ClassNotFoundException {
|
||||
MimeType mt = JavaClassMimeTypeConversion.mimeTypeFromObject(new String[0], null);
|
||||
String className = JavaClassMimeTypeConversion.classNameFromMimeType(mt);
|
||||
assertThat(Class.forName(className)).isEqualTo(String[].class);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void mimeTypeIsMultiDimensionalObjectArray() throws ClassNotFoundException {
|
||||
MimeType mt = JavaClassMimeTypeConversion.mimeTypeFromObject(new String[0][0][0], null);
|
||||
String className = JavaClassMimeTypeConversion.classNameFromMimeType(mt);
|
||||
assertThat(Class.forName(className)).isEqualTo(String[][][].class);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void mimeTypeIsPrimitiveArray() throws ClassNotFoundException {
|
||||
MimeType mt = JavaClassMimeTypeConversion.mimeTypeFromObject(new int[0], null);
|
||||
String className = JavaClassMimeTypeConversion.classNameFromMimeType(mt);
|
||||
assertThat(Class.forName(className)).isEqualTo(int[].class);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void mimeTypeIsMultiDimensionalPrimitiveArray() throws ClassNotFoundException {
|
||||
MimeType mt = JavaClassMimeTypeConversion.mimeTypeFromObject(new int[0][0][0], null);
|
||||
String className = JavaClassMimeTypeConversion.classNameFromMimeType(mt);
|
||||
assertThat(Class.forName(className)).isEqualTo(int[][][].class);
|
||||
}
|
||||
|
||||
public static class Foo {
|
||||
|
||||
private String bar;
|
||||
|
||||
public Foo() {
|
||||
}
|
||||
|
||||
public Foo(String bar) {
|
||||
this.bar = bar;
|
||||
}
|
||||
|
||||
public String getBar() {
|
||||
return bar;
|
||||
}
|
||||
|
||||
public void setBar(String bar) {
|
||||
this.bar = bar;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
public static class Bar {
|
||||
|
||||
private String foo;
|
||||
|
||||
public Bar() {
|
||||
}
|
||||
|
||||
public Bar(String foo) {
|
||||
this.foo = foo;
|
||||
}
|
||||
|
||||
public String getFoo() {
|
||||
return foo;
|
||||
}
|
||||
|
||||
public void setFoo(String foo) {
|
||||
this.foo = foo;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
public class TestMessageChannelBinder extends AbstractBinder<MessageChannel, ConsumerProperties, ProducerProperties> {
|
||||
|
||||
@Override
|
||||
protected Binding<MessageChannel> doBindConsumer(String name, String group, MessageChannel channel, ConsumerProperties properties) {
|
||||
return null;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Binding<MessageChannel> doBindProducer(String name, MessageChannel channel, ProducerProperties properties) {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
private static class TupleRegistrar implements KryoRegistrar {
|
||||
private final TupleKryoRegistrar delegate = new TupleKryoRegistrar();
|
||||
|
||||
@Override
|
||||
public void registerTypes(Kryo kryo) {
|
||||
delegate.registerTypes(kryo);
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<Registration> getRegistrations() {
|
||||
return delegate.getRegistrations();
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user