Stream uses destinations then channels

fixes #252

accurest-messaging/accurest-messaging-stream/src/main/java/io/codearte/accurest/messaging/stream/AccurestStreamMessaging.java

@@ -1,19 +1,20 @@
 package io.codearte.accurest.messaging.stream;
 
-import java.util.Map;
-import java.util.concurrent.TimeUnit;
-
+import io.codearte.accurest.messaging.AccurestMessage;
+import io.codearte.accurest.messaging.AccurestMessageBuilder;
+import io.codearte.accurest.messaging.AccurestMessaging;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.cloud.stream.config.BindingProperties;
+import org.springframework.cloud.stream.config.ChannelBindingServiceProperties;
 import org.springframework.cloud.stream.test.binder.MessageCollector;
 import org.springframework.context.ApplicationContext;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 
-import io.codearte.accurest.messaging.AccurestMessage;
-import io.codearte.accurest.messaging.AccurestMessageBuilder;
-import io.codearte.accurest.messaging.AccurestMessaging;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
 
 /**
  * @author Marcin Grzejszczak
@@ -43,7 +44,7 @@ public class AccurestStreamMessaging<T> implements AccurestMessaging<T, Message<
 	@Override
 	public void send(AccurestMessage<T, Message<T>> message, String destination) {
 		try {
-			MessageChannel messageChannel = context.getBean(destination, MessageChannel.class);
+			MessageChannel messageChannel = context.getBean(resolvedDestination(destination), MessageChannel.class);
 			messageChannel.send(message.convert());
 		} catch (Exception e) {
 			log.error("Exception occurred while trying to send a message [" + message + "] " +
@@ -56,7 +57,7 @@ public class AccurestStreamMessaging<T> implements AccurestMessaging<T, Message<
 	@SuppressWarnings("unchecked")
 	public AccurestMessage<T, Message<T>> receiveMessage(String destination, long timeout, TimeUnit timeUnit) {
 		try {
-			MessageChannel messageChannel = context.getBean(destination, MessageChannel.class);
+			MessageChannel messageChannel = context.getBean(resolvedDestination(destination), MessageChannel.class);
 			return builder.create(messageCollector.forChannel(messageChannel).poll(timeout, timeUnit));
 		} catch (Exception e) {
 			log.error("Exception occurred while trying to read a message from " +
@@ -65,6 +66,19 @@ public class AccurestStreamMessaging<T> implements AccurestMessaging<T, Message<
 		}
 	}
 
+	private String resolvedDestination(String destination) {
+		ChannelBindingServiceProperties channelBindingServiceProperties = context.getBean(ChannelBindingServiceProperties.class);
+		String resolvedDestination = destination;
+		for (Map.Entry<String, BindingProperties> entry : channelBindingServiceProperties.getBindings().entrySet()) {
+			if (entry.getValue().getDestination().equals(destination)) {
+				log.debug("Found a channel named [{}] with destination [{}]", entry.getKey(), destination);
+				return entry.getKey();
+			}
+		}
+		log.debug("No destination named [{}] was found. Assuming that the destination equals the channel name", destination);
+		return resolvedDestination;
+	}
+
 	@Override
 	public AccurestMessage<T, Message<T>> receiveMessage(String destination) {
 		return receiveMessage(destination, 5, TimeUnit.SECONDS);

stub-runner/stub-runner-boot/src/test/groovy/io/codearte/accurest/stubrunner/boot/StubRunnerBootSpec.groovy

@@ -5,6 +5,8 @@ import groovy.json.JsonSlurper
 import io.codearte.accurest.stubrunner.StubRunning
 import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.boot.test.SpringApplicationContextLoader
+import org.springframework.cloud.stream.annotation.EnableBinding
+import org.springframework.context.annotation.Configuration
 import org.springframework.test.context.ContextConfiguration
 import spock.lang.Specification
 
@@ -12,7 +14,9 @@ import spock.lang.Specification
  * @author Marcin Grzejszczak
  */
 // tag::boot_usage[]
-@ContextConfiguration(classes = StubRunnerBoot, loader = SpringApplicationContextLoader)
+@ContextConfiguration(classes = [StubRunnerBootSpec, StubRunnerBoot], loader = SpringApplicationContextLoader)
+@EnableBinding
+@Configuration
 class StubRunnerBootSpec extends Specification {
 
 	@Autowired StubRunning stubRunning
@@ -87,6 +91,6 @@ class StubRunnerBootSpec extends Specification {
 			response.statusCode == 404
 			response.body.asString() == '''{"io.codearte.accurest.stubs:streamService:stubs":["delete_book","return_book_1","return_book_2"]}'''
 	}
-// end::boot_usage[]
 
 }
+// end::boot_usage[]

stub-runner/stub-runner-messaging/stub-runner-messaging-stream/README.adoc

@@ -1,12 +1,13 @@
-:input_name: input
-:output_name: output
-
 === Stub Runner Messaging Stream
 
 Accurest Stub Runner's messaging module gives you an easy way to integrate with Spring Stream.
 For the provided artifacts it will automatically download the stubs and register the required
 routes.
 
+WARNING: In Stub Runner's integration with Stream the `messageFrom` or `sentTo` Strings are resolved
+first as a `destination` of a channel, and then if there is no such `destination` it's resolved as a
+channel name.
+
 ==== Adding it to the project
 
 To use it you have to add the following dependency to your project (example for Gradle):
@@ -67,16 +68,24 @@ and number *2*
 include::src/test/groovy/io/codearte/accurest/stubrunner/messaging/stream/StreamStubRunnerSpec.groovy[tags=sample_dsl_2,indent=0]
 ----
 
+and the following Spring configuration:
+
+[source,yaml]
+----
+include::src/test/resources/application.yml[]
+----
+
+
 ===== Scenario 1 (no input message)
 
-So as to trigger a message via the `return_book_1` label we'll use the `StubTigger` interface as follows
+So as to trigger a message via the `return_book_1` label we'll use the `StubTrigger` interface as follows
 
 [source,groovy]
 ----
 include::src/test/groovy/io/codearte/accurest/stubrunner/messaging/stream/StreamStubRunnerSpec.groovy[tags=client_trigger,indent=0]
 ----
 
-Next we'll want to listen to the output of the message sent to `{output_name}`
+Next we'll want to listen to the output of the message sent to a channel whose `destination` is `returnBook`
 
 [source,groovy]
 ----
@@ -92,14 +101,14 @@ include::src/test/groovy/io/codearte/accurest/stubrunner/messaging/stream/Stream
 
 ===== Scenario 2 (output triggered by input)
 
-Since the route is set for you it's enough to just send a message to the `{output_name}` destination.
+Since the route is set for you it's enough to just send a message to the `bookStorage` `destination`.
 
 [source,groovy]
 ----
 include::src/test/groovy/io/codearte/accurest/stubrunner/messaging/stream/StreamStubRunnerSpec.groovy[tags=client_send,indent=0]
 ----
 
-Next we'll want to listen to the output of the message sent to `{output_name}`
+Next we'll want to listen to the output of the message sent to `returnBook`
 
 [source,groovy]
 ----

stub-runner/stub-runner-messaging/stub-runner-messaging-stream/build.gradle

@@ -11,7 +11,6 @@ repositories {
 
 dependencies {
 	compile project(':stub-runner-root:stub-runner-spring')
-	compile project(':stub-runner-root:stub-runner-messaging-root:stub-runner-messaging-integration')
 	compile project(':accurest-messaging-root:accurest-messaging-stream')
 	compile "org.springframework.integration:spring-integration-java-dsl:${springIntegrationDslVersion}"
 

stub-runner/stub-runner-messaging/stub-runner-messaging-stream/src/main/groovy/io/codearte/accurest/stubrunner/messaging/stream/StubRunnerStreamConfiguration.groovy

@@ -0,0 +1,78 @@
+package io.codearte.accurest.stubrunner.messaging.stream
+
+import groovy.transform.CompileStatic
+import io.codearte.accurest.dsl.GroovyDsl
+import io.codearte.accurest.stubrunner.BatchStubRunner
+import io.codearte.accurest.stubrunner.StubConfiguration
+import io.codearte.accurest.stubrunner.spring.StubRunnerConfiguration
+import org.slf4j.Logger
+import org.slf4j.LoggerFactory
+import org.springframework.beans.factory.config.AutowireCapableBeanFactory
+import org.springframework.cloud.stream.config.BindingProperties
+import org.springframework.cloud.stream.config.ChannelBindingServiceProperties
+import org.springframework.context.Lifecycle
+import org.springframework.context.annotation.Bean
+import org.springframework.context.annotation.Configuration
+import org.springframework.context.annotation.Import
+import org.springframework.integration.dsl.FilterEndpointSpec
+import org.springframework.integration.dsl.GenericEndpointSpec
+import org.springframework.integration.dsl.IntegrationFlowBuilder
+import org.springframework.integration.dsl.IntegrationFlows
+import org.springframework.messaging.Message
+/**
+ * Spring Cloud Stream configuration that iterates over the downloaded Groovy DSLs
+ * and registers a flow for each DSL.
+ *
+ * @author Marcin Grzejszczak
+ */
+@Configuration
+@Import(StubRunnerConfiguration)
+@CompileStatic
+class StubRunnerStreamConfiguration {
+
+	private static final Logger log = LoggerFactory.getLogger(StubRunnerStreamConfiguration)
+
+	@Bean
+	FlowRegistrar flowRegistrar(AutowireCapableBeanFactory beanFactory, BatchStubRunner batchStubRunner) {
+		Map<StubConfiguration, Collection<GroovyDsl>> accurestContracts = batchStubRunner.accurestContracts
+		accurestContracts.each { StubConfiguration key, Collection<GroovyDsl> value ->
+			String name = "${key.groupId}_${key.artifactId}"
+			value.findAll { it?.input?.messageFrom?.clientValue }.each { GroovyDsl dsl ->
+				String flowName = "${name}_${dsl.label}_${dsl.hashCode()}"
+				String from = resolvedDestination(beanFactory, dsl.input.messageFrom.clientValue)
+				IntegrationFlowBuilder builder = IntegrationFlows.from(from)
+						.filter(new StubRunnerStreamMessageSelector(dsl), { FilterEndpointSpec e -> e.id("${flowName}.filter") } )
+						.transform(new StubRunnerStreamTransformer(dsl), { GenericEndpointSpec e -> e.id("${flowName}.transformer") })
+				if (dsl.outputMessage?.sentTo) {
+					builder = builder.channel(resolvedDestination(beanFactory, dsl.outputMessage.sentTo.clientValue))
+				} else {
+					builder = builder.handle(new DummyMessageHandler(), "handle")
+				}
+				beanFactory.initializeBean(builder.get(), flowName)
+				beanFactory.getBean("${flowName}.filter", Lifecycle.class).start();
+				beanFactory.getBean("${flowName}.transformer", Lifecycle.class).start();
+			}
+		}
+		return new FlowRegistrar()
+	}
+
+	private String resolvedDestination(AutowireCapableBeanFactory context, String destination) {
+		ChannelBindingServiceProperties channelBindingServiceProperties = context.getBean(ChannelBindingServiceProperties.class);
+		String resolvedDestination = destination;
+		for (Map.Entry<String, BindingProperties> entry : channelBindingServiceProperties.getBindings().entrySet()) {
+			if (entry.getValue().getDestination().equals(destination)) {
+				log.debug("Found a channel named [{}] with destination [{}]", entry.getKey(), destination);
+				return entry.getKey();
+			}
+		}
+		log.debug("No destination named [{}] was found. Assuming that the destination equals the channel name", destination);
+		return resolvedDestination;
+	}
+
+	@CompileStatic
+	private static class DummyMessageHandler {
+		void handle(Message message) {}
+	}
+
+	static class FlowRegistrar {}
+}

stub-runner/stub-runner-messaging/stub-runner-messaging-stream/src/main/resources/META-INF/spring.factories

@@ -0,0 +1,3 @@
+# Auto Configuration
+org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
+io.codearte.accurest.stubrunner.messaging.stream.StubRunnerStreamConfiguration

stub-runner/stub-runner-messaging/stub-runner-messaging-stream/src/test/groovy/io/codearte/accurest/stubrunner/messaging/stream/StreamStubRunnerSpec.groovy

@@ -33,17 +33,17 @@ class StreamStubRunnerSpec extends Specification {
 
 	def setup() {
 		// ensure that message were taken from the queue
-		messaging.receiveMessage('output', 100, TimeUnit.MILLISECONDS)
+		messaging.receiveMessage('returnBook', 100, TimeUnit.MILLISECONDS)
 	}
 
 	def 'should download the stub and register a route for it'() {
 		when:
 		// tag::client_send[]
-			messaging.send(new BookReturned('foo'), [sample: 'header'], 'input')
+			messaging.send(new BookReturned('foo'), [sample: 'header'], 'bookStorage')
 		// end::client_send[]
 		then:
 		// tag::client_receive[]
-			AccurestMessage receivedMessage = messaging.receiveMessage('output')
+			AccurestMessage receivedMessage = messaging.receiveMessage('returnBook')
 		// end::client_receive[]
 		and:
 		// tag::client_receive_message[]
@@ -60,7 +60,7 @@ class StreamStubRunnerSpec extends Specification {
 		// end::client_trigger[]
 		then:
 		// tag::client_trigger_receive[]
-			AccurestMessage receivedMessage = messaging.receiveMessage('output')
+			AccurestMessage receivedMessage = messaging.receiveMessage('returnBook')
 		// end::client_trigger_receive[]
 		and:
 		// tag::client_trigger_message[]
@@ -76,7 +76,7 @@ class StreamStubRunnerSpec extends Specification {
 			stubFinder.trigger('io.codearte.accurest.stubs:streamService', 'return_book_1')
 		// end::trigger_group_artifact[]
 		then:
-			AccurestMessage receivedMessage = messaging.receiveMessage('output')
+			AccurestMessage receivedMessage = messaging.receiveMessage('returnBook')
 		and:
 			receivedMessage != null
 			assertJsons(receivedMessage.payload)
@@ -89,7 +89,7 @@ class StreamStubRunnerSpec extends Specification {
 			stubFinder.trigger('streamService', 'return_book_1')
 		// end::trigger_artifact[]
 		then:
-			AccurestMessage receivedMessage = messaging.receiveMessage('output')
+			AccurestMessage receivedMessage = messaging.receiveMessage('returnBook')
 		and:
 			receivedMessage != null
 			assertJsons(receivedMessage.payload)
@@ -116,7 +116,7 @@ class StreamStubRunnerSpec extends Specification {
 			stubFinder.trigger()
 		// end::trigger_all[]
 		then:
-			AccurestMessage receivedMessage = messaging.receiveMessage('output')
+			AccurestMessage receivedMessage = messaging.receiveMessage('returnBook')
 		and:
 			receivedMessage != null
 			assertJsons(receivedMessage.payload)
@@ -134,9 +134,9 @@ class StreamStubRunnerSpec extends Specification {
 
 	def 'should not trigger a message that does not match input'() {
 		when:
-			messaging.send(new BookReturned('not_matching'), [wrong: 'header_value'], 'input')
+			messaging.send(new BookReturned('not_matching'), [wrong: 'header_value'], 'bookStorage')
 		then:
-			AccurestMessage receivedMessage = messaging.receiveMessage('output', 100, TimeUnit.MILLISECONDS)
+			AccurestMessage receivedMessage = messaging.receiveMessage('returnBook', 100, TimeUnit.MILLISECONDS)
 		and:
 			receivedMessage == null
 	}
@@ -156,7 +156,7 @@ class StreamStubRunnerSpec extends Specification {
 			triggeredBy('bookReturnedTriggered()')
 		}
 		outputMessage {
-			sentTo('jms:output')
+			sentTo('returnBook')
 			body('''{ "bookName" : "foo" }''')
 			headers {
 				header('BOOK-NAME', 'foo')
@@ -170,7 +170,7 @@ class StreamStubRunnerSpec extends Specification {
 	io.codearte.accurest.dsl.GroovyDsl.make {
 		label 'return_book_2'
 		input {
-			messageFrom('jms:input')
+			messageFrom('bookStorage')
 			messageBody([
 					bookName: 'foo'
 			])
@@ -179,7 +179,7 @@ class StreamStubRunnerSpec extends Specification {
 			}
 		}
 		outputMessage {
-			sentTo('jms:output')
+			sentTo('returnBook')
 			body([
 					bookName: 'foo'
 			])
@@ -195,7 +195,7 @@ class StreamStubRunnerSpec extends Specification {
 			io.codearte.accurest.dsl.GroovyDsl.make {
 				label 'delete_book'
 				input {
-					messageFrom('jms:delete')
+					messageFrom('delete')
 					messageBody([
 							bookName: 'foo'
 					])

stub-runner/stub-runner-messaging/stub-runner-messaging-stream/src/test/resources/application.yml

@@ -1,2 +1,11 @@
 stubrunner.stubs.repository.root: classpath:m2repo/repository/
-stubrunner.stubs.ids: io.codearte.accurest.stubs:streamService
+stubrunner.stubs.ids: io.codearte.accurest.stubs:streamService
+
+spring:
+  cloud:
+    stream:
+      bindings:
+        output:
+          destination: returnBook
+        input:
+          destination: bookStorage