diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java index f6ac63313..94b7291ce 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java @@ -225,7 +225,9 @@ public class MessageConverterConfigurer implements MessageChannelConfigurer, Bea @Override public Message preSend(Message message, MessageChannel channel) { Message sentMessage = null; - if (this.klazz.isAssignableFrom(message.getPayload().getClass())) { + if (this.klazz.isAssignableFrom(message.getPayload().getClass()) || + (this.klazz.isAssignableFrom(String.class) && message.getPayload() instanceof byte[] + && !message.getHeaders().containsKey(MessageHeaders.CONTENT_TYPE) && !this.contentType.equals("text/plain"))) { Object contentTypeFromMessage = message.getHeaders().get(MessageHeaders.CONTENT_TYPE); if (contentTypeFromMessage == null) { sentMessage = MessageConverterConfigurer.this.messageBuilderFactory diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/contenttype/BinderConversionTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/contenttype/BinderConversionTests.java new file mode 100644 index 000000000..582eb07dc --- /dev/null +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/contenttype/BinderConversionTests.java @@ -0,0 +1,80 @@ +/* + * Copyright 2017 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.contenttype; + +import java.nio.charset.StandardCharsets; + +import org.junit.Test; + +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.boot.builder.SpringApplicationBuilder; +import org.springframework.cloud.stream.annotation.EnableBinding; +import org.springframework.cloud.stream.annotation.StreamListener; +import org.springframework.cloud.stream.binder.integration.SourceDestination; +import org.springframework.cloud.stream.binder.integration.SpringIntegrationBinderConfiguration; +import org.springframework.cloud.stream.binder.integration.TargetDestination; +import org.springframework.cloud.stream.messaging.Processor; +import org.springframework.context.ApplicationContext; +import org.springframework.context.annotation.Import; +import org.springframework.messaging.handler.annotation.SendTo; +import org.springframework.messaging.support.GenericMessage; + +/** + * Sort of a TCK test suite to validate payload conversion is + * done properly by interacting with binder's input/output destinations + * instead of its bridged channels. + * This means that all payloads (sent/received) must be expressed in the + * wire format (byte[]) + * + * @author Oleg Zhurakousky + * + */ +public class BinderConversionTests { + @Test + public void test() { + ApplicationContext context = new SpringApplicationBuilder(ApplicationJsonDefaultType.class).web(false) + .run("--spring.cloud.stream.default.contentType=application/json", "--spring.jmx.enabled=false"); + SourceDestination source = context.getBean(SourceDestination.class); + TargetDestination target = context.getBean(TargetDestination.class); + String jsonPayload = "{\"name\":\"oleg\"}"; + source.send(new GenericMessage(jsonPayload.getBytes())); + System.out.println(new String((byte[])target.receive().getPayload(), StandardCharsets.UTF_8)); + } + + @SpringBootApplication + @EnableBinding(Processor.class) + @Import(SpringIntegrationBinderConfiguration.class) + public static class ApplicationJsonDefaultType { + @StreamListener(Processor.INPUT) + @SendTo(Processor.OUTPUT) + public Person echo(Person value) { + return value; + } + + public static class Person { + private String name; + + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + } + } +} diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/AbstractDestination.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/AbstractDestination.java new file mode 100644 index 000000000..bbd5ab28d --- /dev/null +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/AbstractDestination.java @@ -0,0 +1,41 @@ +/* + * Copyright 2017 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.integration; + +import org.springframework.messaging.SubscribableChannel; + +/** + * @author Oleg Zhurakousky + * + */ +abstract class AbstractDestination { + + private SubscribableChannel channel; + + SubscribableChannel getChannel() { + return channel; + } + + void setChannel(SubscribableChannel channel) { + this.channel = channel; + this.afterChannelIsSet(); + } + + void afterChannelIsSet() { + // noop + } +} diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SampleStreamApp.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SampleStreamApp.java new file mode 100644 index 000000000..d3fabe8fc --- /dev/null +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SampleStreamApp.java @@ -0,0 +1,70 @@ +/* + * Copyright 2017 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.integration; + +import java.nio.charset.StandardCharsets; + +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.boot.builder.SpringApplicationBuilder; +import org.springframework.cloud.stream.annotation.EnableBinding; +import org.springframework.cloud.stream.annotation.StreamListener; +import org.springframework.cloud.stream.messaging.Processor; +import org.springframework.context.ApplicationContext; +import org.springframework.context.annotation.Import; +import org.springframework.integration.annotation.ServiceActivator; +import org.springframework.messaging.Message; +import org.springframework.messaging.handler.annotation.SendTo; +import org.springframework.messaging.support.GenericMessage; + +import static org.junit.Assert.assertEquals; + +/** + * Sample spring cloud stream application that demonstrates the usage of {@link SpringIntegrationChannelBinder}. + * + * @author Oleg Zhurakousky + * + */ +@SpringBootApplication +@EnableBinding(Processor.class) +@Import(SpringIntegrationBinderConfiguration.class) +public class SampleStreamApp { + + public static void main(String[] args) { + ApplicationContext context = new SpringApplicationBuilder(SampleStreamApp.class).web(false) + .run("--server.port=0"); + SourceDestination source = context.getBean(SourceDestination.class); + TargetDestination target = context.getBean(TargetDestination.class); + source.send(new GenericMessage("Hello".getBytes())); + Message message = target.receive(); + assertEquals("Hello", new String((byte[])message.getPayload(), StandardCharsets.UTF_8)); + System.out.println(); + } + + @StreamListener(Processor.INPUT) + @SendTo(Processor.OUTPUT) + public String receive(String value) { + System.out.println("Handling payload: " + value); + return value; + } + + @ServiceActivator(inputChannel="input.anonymous.errors") + public void error(String value) { + System.out.println("Handling ERROR payload: " + value); + } +} + + diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SourceDestination.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SourceDestination.java new file mode 100644 index 000000000..ee03161c4 --- /dev/null +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SourceDestination.java @@ -0,0 +1,40 @@ +/* + * Copyright 2017 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.integration; + +import org.springframework.messaging.Message; + +/** + * Implementation of binder endpoint that represents the source destination + * (e.g., destination from which messages will be received by Processor.INPUT). + *
+ * You can interact with it by calling {@link #send(Message)} operation. + * + * @author Oleg Zhurakousky + * + */ +public class SourceDestination extends AbstractDestination { + + /** + * Allows the {@link Message} to be sent to a Binder to be delegated + * to binder's input destination (e.g., Processor.INPUT). + * + */ + public void send(Message message) { + this.getChannel().send(message); + } +} diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SpringIntegrationBinderConfiguration.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SpringIntegrationBinderConfiguration.java new file mode 100644 index 000000000..56cb544ec --- /dev/null +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SpringIntegrationBinderConfiguration.java @@ -0,0 +1,99 @@ +/* + * Copyright 2017 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.integration; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Map; + +import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; +import org.springframework.cloud.stream.annotation.EnableBinding; +import org.springframework.cloud.stream.binder.Binder; +import org.springframework.cloud.stream.binder.BinderType; +import org.springframework.cloud.stream.binder.BinderTypeRegistry; +import org.springframework.cloud.stream.binder.ConsumerProperties; +import org.springframework.cloud.stream.binder.DefaultBinderTypeRegistry; +import org.springframework.cloud.stream.binder.ProducerProperties; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.Import; +import org.springframework.core.annotation.AnnotationUtils; +import org.springframework.integration.config.EnableIntegration; + +/** + * {@link Binder} configuration backed by Spring Integration. + * + * Please see {@link SpringIntegrationChannelBinder} for more details. + * + * @author Oleg Zhurakousky + * + * @see SpringIntegrationChannelBinder + */ +@Configuration +@ConditionalOnMissingBean(Binder.class) +@EnableIntegration +public class SpringIntegrationBinderConfiguration { + + public static final String NAME = "integration"; + + /** + * Utility operation to return an array of configuration classes + * defined in {@link EnableBinding} annotation. + * Typically used for tests that do not rely on creating an SCSt boot + * application annotated with {@link EnableBinding}, yet require + * full {@link Binder} configuration. + */ + public static Class[] getCompleteConfiguration() { + List> configClasses = new ArrayList<>(); + configClasses.add(SpringIntegrationBinderConfiguration.class); + Import annotation = AnnotationUtils.getAnnotation(EnableBinding.class, Import.class); + Map annotationAttributes = AnnotationUtils.getAnnotationAttributes(annotation); + configClasses.addAll(Arrays.asList((Class[])annotationAttributes.get("value"))); + return configClasses.toArray(new Class[] {}); + } + + @Bean + public BinderTypeRegistry binderTypeRegistry() { + BinderType binderType = new BinderType(NAME, new Class[] {SpringIntegrationBinderConfiguration.class}); + BinderTypeRegistry btr = new DefaultBinderTypeRegistry(Collections.singletonMap(NAME, binderType)); + return btr; + } + + @Bean + public SourceDestination sourceDestination() { + return new SourceDestination(); + } + + @Bean + public TargetDestination targetDestination() { + return new TargetDestination(); + } + + @SuppressWarnings("unchecked") + @Bean + public Binder springIntegrationChannelBinder(SpringIntegrationProvisioner provisioner) { + return (Binder) new SpringIntegrationChannelBinder(provisioner); + } + + @Bean + public SpringIntegrationProvisioner springIntegrationProvisioner() { + return new SpringIntegrationProvisioner(); + } + +} diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SpringIntegrationChannelBinder.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SpringIntegrationChannelBinder.java new file mode 100644 index 000000000..fdd190535 --- /dev/null +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SpringIntegrationChannelBinder.java @@ -0,0 +1,267 @@ +/* + * Copyright 2015-2017 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.integration; + +import java.util.function.Consumer; + +import org.springframework.beans.factory.BeanFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.cloud.stream.binder.AbstractMessageChannelBinder; +import org.springframework.cloud.stream.binder.Binder; +import org.springframework.cloud.stream.binder.ConsumerProperties; +import org.springframework.cloud.stream.binder.ProducerProperties; +import org.springframework.cloud.stream.binder.integration.SpringIntegrationProvisioner.SpringIntegrationConsumerDestination; +import org.springframework.cloud.stream.binder.integration.SpringIntegrationProvisioner.SpringIntegrationProducerDestination; +import org.springframework.cloud.stream.provisioning.ConsumerDestination; +import org.springframework.cloud.stream.provisioning.ProducerDestination; +import org.springframework.core.AttributeAccessor; +import org.springframework.integration.core.MessageProducer; +import org.springframework.integration.endpoint.MessageProducerSupport; +import org.springframework.integration.handler.BridgeHandler; +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.retry.RecoveryCallback; +import org.springframework.retry.RetryCallback; +import org.springframework.retry.RetryContext; +import org.springframework.retry.RetryListener; +import org.springframework.retry.support.RetryTemplate; +import org.springframework.util.Assert; +import org.springframework.util.StringUtils; + +/** + * Implementation of {@link Binder} backed by Spring Integration framework. + * It is useful for localized demos and testing. + *

+ * This binder extends from the same base class ({@link AbstractMessageChannelBinder}) as + * other binders (i.e., Rabbit, Kafka etc). Interaction with this binder is done via source + * and target destination which emulate real binder's destinations (i.e., Kafka topic) + *
+ * The destination classes are + *

    + *
  • {@link SourceDestination}
    • + *
  • {@link TargetDestination}
    • + *
+ * Simply autowire them in your your application and send/receive messages. + *

+ * You must also add {@link SpringIntegrationBinderConfiguration} to your configuration. + * Below is the example using Spring Boot test. + * 
+ *
+ * @RunWith(SpringJUnit4ClassRunner.class)
+ * @SpringBootTest(classes = {SpringIntegrationBinderConfiguration.class, TestWithSIBinder.MyProcessor.class})
+ * public class TestWithSIBinder {
+ *     @Autowired
+ *     private SourceDestination sourceDestination;
+ *
+ *     @Autowired
+ *     private TargetDestination targetDestination;
+ *
+ *     @Test
+ *     public void testWiring() {
+ *         sourceDestination.send(new GenericMessage("Hello"));
+ *         assertEquals("Hello world", new String((byte[])targetDestination.receive().getPayload(), StandardCharsets.UTF_8));
+ *     }
+ *
+ *     @SpringBootApplication
+ *     @EnableBinding(Processor.class)
+ *     public static class MyProcessor {
+ *         @StreamListener(Processor.INPUT)
+ *         @SendTo(Processor.OUTPUT)
+ *         public String transform(String in) {
+ *             return in + " world";
+ *         }
+ *     }
+ * }
+ *
+ * + * @author Oleg Zhurakousky + */ +class SpringIntegrationChannelBinder extends AbstractMessageChannelBinder { + + @Autowired + private BeanFactory beanFactory; + + SpringIntegrationChannelBinder(SpringIntegrationProvisioner provisioningProvider) { + super(true, new String[] {}, provisioningProvider); + } + + @Override + protected MessageHandler createProducerMessageHandler(ProducerDestination destination, + ProducerProperties producerProperties, MessageChannel errorChannel) throws Exception { + BridgeHandler handler = new BridgeHandler(); + handler.setBeanFactory(this.beanFactory); + handler.setOutputChannel(((SpringIntegrationProducerDestination)destination).getChannel()); + return handler; + } + + @Override + protected MessageProducer createConsumerEndpoint(ConsumerDestination destination, String group, ConsumerProperties properties) + throws Exception { + ErrorMessageStrategy errorMessageStrategy = new DefaultErrorMessageStrategy(); + SubscribableChannel siBinderInputChannel = ((SpringIntegrationConsumerDestination)destination).getChannel(); + + IntegrationMessageListeningContainer messageListenerContainer = new IntegrationMessageListeningContainer(); + IntegrationBinderInboundChannelAdapter adapter = new IntegrationBinderInboundChannelAdapter(messageListenerContainer); + + String groupName = StringUtils.hasText(group) ? group : "anonymous"; + ErrorInfrastructure errorInfrastructure = registerErrorInfrastructure(destination, groupName, properties); + if (properties.getMaxAttempts() > 1) { + adapter.setRetryTemplate(buildRetryTemplate(properties)); + adapter.setRecoveryCallback(errorInfrastructure.getRecoverer()); + } + else { + adapter.setErrorMessageStrategy(errorMessageStrategy); + adapter.setErrorChannel(errorInfrastructure.getErrorChannel()); + } + + siBinderInputChannel.subscribe(messageListenerContainer); + + return adapter; + } + + /** + * Implementation of simple message listener container modeled after AMQP SimpleMessageListenerContainer + */ + private static class IntegrationMessageListeningContainer implements MessageHandler { + + private Consumer> listener; + + @Override + public void handleMessage(Message message) throws MessagingException { + this.listener.accept(message); + } + + public void setMessageListener(Consumer> listener) { + this.listener = listener; + } + } + + /** + * Implementation of inbound channel adapter modeled after AmqpInboundChannelAdapter + */ + private static class IntegrationBinderInboundChannelAdapter extends MessageProducerSupport { + + private static final ThreadLocal attributesHolder = new ThreadLocal(); + + private final IntegrationMessageListeningContainer listenerContainer; + + private RetryTemplate retryTemplate; + + private RecoveryCallback recoveryCallback; + + IntegrationBinderInboundChannelAdapter(IntegrationMessageListeningContainer listenerContainer) { + this.listenerContainer = listenerContainer; + } + + @SuppressWarnings("unused") + // Temporarily unused until DLQ strategy for this binder becomes a requirement + public void setRecoveryCallback(RecoveryCallback recoveryCallback) { + this.recoveryCallback = recoveryCallback; + } + + public void setRetryTemplate(RetryTemplate retryTemplate) { + this.retryTemplate = retryTemplate; + } + + @Override + protected void onInit() { + if (this.retryTemplate != null) { + Assert.state(getErrorChannel() == null, "Cannot have an 'errorChannel' property when a 'RetryTemplate' is " + + "provided; use an 'ErrorMessageSendingRecoverer' in the 'recoveryCallback' property to " + + "send an error message when retries are exhausted"); + } + Listener messageListener = new Listener(); + if (this.retryTemplate != null) { + this.retryTemplate.registerListener(messageListener); + } + this.listenerContainer.setMessageListener(messageListener); + } + + protected class Listener implements RetryListener, Consumer> { + + @Override + @SuppressWarnings({ "unchecked", "rawtypes" }) + public void accept(final Message message) { + try { + if (IntegrationBinderInboundChannelAdapter.this.retryTemplate == null) { + try { + processMessage(message); + } + finally { + attributesHolder.remove(); + } + } + else { + try { + IntegrationBinderInboundChannelAdapter.this.retryTemplate.execute(new RetryCallback() { + + @Override + public Object doWithRetry(RetryContext context) throws Throwable { + processMessage(message); + return null; + } + },(RecoveryCallback) IntegrationBinderInboundChannelAdapter.this.recoveryCallback); + } + catch (Throwable e) { + // TODO Auto-generated catch block + e.printStackTrace(); + } + } + } + catch (RuntimeException e) { + if (getErrorChannel() != null) { + getMessagingTemplate().send(getErrorChannel(), buildErrorMessage(null, + new IllegalStateException("Message conversion failed: " + message, e))); + } + else { + throw e; + } + } + } + + private void processMessage(Message message) { + sendMessage(message); + } + + @Override + public boolean open(RetryContext context, RetryCallback callback) { + if (IntegrationBinderInboundChannelAdapter.this.recoveryCallback != null) { + attributesHolder.set(context); + } + return true; + } + + @Override + public void close(RetryContext context, RetryCallback callback, + Throwable throwable) { + attributesHolder.remove(); + } + + @Override + public void onError(RetryContext context, RetryCallback callback, + Throwable throwable) { + // Empty + } + } + } +} diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SpringIntegrationProvisioner.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SpringIntegrationProvisioner.java new file mode 100644 index 000000000..ddd3a8f46 --- /dev/null +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SpringIntegrationProvisioner.java @@ -0,0 +1,133 @@ +/* + * Copyright 2017 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.integration; + +import java.util.HashMap; +import java.util.Map; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.cloud.stream.binder.Binder; +import org.springframework.cloud.stream.binder.ConsumerProperties; +import org.springframework.cloud.stream.binder.ProducerProperties; +import org.springframework.cloud.stream.provisioning.ConsumerDestination; +import org.springframework.cloud.stream.provisioning.ProducerDestination; +import org.springframework.cloud.stream.provisioning.ProvisioningException; +import org.springframework.cloud.stream.provisioning.ProvisioningProvider; +import org.springframework.integration.channel.AbstractMessageChannel; +import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.channel.PublishSubscribeChannel; +import org.springframework.messaging.Message; +import org.springframework.messaging.SubscribableChannel; + +/** + * {@link ProvisioningProvider} to support {@link SpringIntegrationChannelBinder}. It + * exists primarily to support {@link AbstractMessageChannel} semantics for creating + * {@link ConsumerDestination} and {@link ProducerDestination}, to interact with this + * {@link Binder}. + * + * @author Oleg Zhurakousky + * + */ +class SpringIntegrationProvisioner implements ProvisioningProvider { + + private final Map provisionedDestinations = new HashMap<>(); + + @Autowired + private SourceDestination source; + + @Autowired + private TargetDestination target; + + /** + * Will provision producer destination as an SI {@link PublishSubscribeChannel}. + *
+ * This provides convenience of registering additional subscriber (handler in the test method) + * along side of being able to call {@link TargetDestination#receive()} to get a + * {@link Message} for additional assertions. + */ + @Override + public ProducerDestination provisionProducerDestination(String name, ProducerProperties properties) throws ProvisioningException { + SubscribableChannel destination = this.provisionDestination(name, true); + this.target.setChannel(destination); + return new SpringIntegrationProducerDestination(name, destination); + } + + /** + * Will provision consumer destination as SI {@link DirectChannel} + */ + @Override + public ConsumerDestination provisionConsumerDestination(String name, String group, ConsumerProperties properties) throws ProvisioningException { + SubscribableChannel destination = this.provisionDestination(name, false); + this.source.setChannel(destination); + return new SpringIntegrationConsumerDestination(name, destination); + } + + private SubscribableChannel provisionDestination(String name, boolean pubSub) { + String destinationName = name + ".destination"; + SubscribableChannel destination = this.provisionedDestinations.get(destinationName); + if (destination == null) { + destination = pubSub ? new PublishSubscribeChannel() : new DirectChannel(); + ((AbstractMessageChannel)destination).setBeanName(destinationName); + ((AbstractMessageChannel)destination).setComponentName(destinationName); + this.provisionedDestinations.put(destinationName, destination); + } + return destination; + } + + class SpringIntegrationConsumerDestination implements ConsumerDestination { + private final String name; + private final SubscribableChannel channel; + + SpringIntegrationConsumerDestination(String name, SubscribableChannel channel) { + this.name = name; + this.channel = channel; + } + + public SubscribableChannel getChannel() { + return this.channel; + } + + @Override + public String getName() { + return this.name; + } + } + + class SpringIntegrationProducerDestination implements ProducerDestination { + private final String name; + private final SubscribableChannel channel; + + SpringIntegrationProducerDestination(String name, SubscribableChannel channel) { + this.name = name; + this.channel = channel; + } + + @Override + public String getNameForPartition(int partition) { + return this.getName() + partition; + } + + public SubscribableChannel getChannel() { + return this.channel; + } + + @Override + public String getName() { + return this.name; + } + } +} diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/TargetDestination.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/TargetDestination.java new file mode 100644 index 000000000..217e68ee6 --- /dev/null +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/TargetDestination.java @@ -0,0 +1,72 @@ +/* + * Copyright 2017 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.integration; + +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.LinkedTransferQueue; +import java.util.concurrent.TimeUnit; + +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHandler; +import org.springframework.messaging.MessagingException; + +/** + * Implementation of binder endpoint that represents the target destination + * (e.g., destination which receives messages sent to Processor.OUTPUT) + *
+ * You can interact with it by calling {@link #receive()} operation. + * + * @author Oleg Zhurakousky + * + */ +public class TargetDestination extends AbstractDestination { + + private BlockingQueue> messages; + + /** + * Allows to access {@link Message}s received by this {@link TargetDestination}. + * @param timeout how long to wait before giving up + */ + public Message receive(long timeout) { + try { + return this.messages.poll(timeout, TimeUnit.MILLISECONDS); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return null; + } + + /** + * Allows to access {@link Message}s received by this {@link TargetDestination}. + */ + public Message receive() { + return this.receive(0); + } + + @Override + void afterChannelIsSet() { + this.messages = new LinkedTransferQueue<>(); + this.getChannel().subscribe(new MessageHandler() { + + @Override + public void handleMessage(Message message) throws MessagingException { + messages.offer(message); + } + }); + } +}