diff --git a/accurest-messaging/accurest-messaging-stream/src/main/java/io/codearte/accurest/messaging/stream/AccurestStreamMessaging.java b/accurest-messaging/accurest-messaging-stream/src/main/java/io/codearte/accurest/messaging/stream/AccurestStreamMessaging.java index a300178cc0..4fd456e637 100644 --- a/accurest-messaging/accurest-messaging-stream/src/main/java/io/codearte/accurest/messaging/stream/AccurestStreamMessaging.java +++ b/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 implements AccurestMessaging> 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 implements AccurestMessaging> 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 implements AccurestMessaging 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> receiveMessage(String destination) { return receiveMessage(destination, 5, TimeUnit.SECONDS); diff --git a/stub-runner/stub-runner-boot/src/test/groovy/io/codearte/accurest/stubrunner/boot/StubRunnerBootSpec.groovy b/stub-runner/stub-runner-boot/src/test/groovy/io/codearte/accurest/stubrunner/boot/StubRunnerBootSpec.groovy index 2c7f041e89..18653638ac 100644 --- a/stub-runner/stub-runner-boot/src/test/groovy/io/codearte/accurest/stubrunner/boot/StubRunnerBootSpec.groovy +++ b/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[] diff --git a/stub-runner/stub-runner-messaging/stub-runner-messaging-stream/README.adoc b/stub-runner/stub-runner-messaging/stub-runner-messaging-stream/README.adoc index 2201ae5f98..aa9af8950d 100644 --- a/stub-runner/stub-runner-messaging/stub-runner-messaging-stream/README.adoc +++ b/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] ---- diff --git a/stub-runner/stub-runner-messaging/stub-runner-messaging-stream/build.gradle b/stub-runner/stub-runner-messaging/stub-runner-messaging-stream/build.gradle index 6886c2201d..b03e790ea4 100644 --- a/stub-runner/stub-runner-messaging/stub-runner-messaging-stream/build.gradle +++ b/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}" diff --git a/stub-runner/stub-runner-messaging/stub-runner-messaging-stream/src/main/groovy/io/codearte/accurest/stubrunner/messaging/stream/StubRunnerStreamConfiguration.groovy b/stub-runner/stub-runner-messaging/stub-runner-messaging-stream/src/main/groovy/io/codearte/accurest/stubrunner/messaging/stream/StubRunnerStreamConfiguration.groovy new file mode 100644 index 0000000000..41813d1789 --- /dev/null +++ b/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> accurestContracts = batchStubRunner.accurestContracts + accurestContracts.each { StubConfiguration key, Collection 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 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 {} +} diff --git a/stub-runner/stub-runner-messaging/stub-runner-messaging-stream/src/main/resources/META-INF/spring.factories b/stub-runner/stub-runner-messaging/stub-runner-messaging-stream/src/main/resources/META-INF/spring.factories new file mode 100644 index 0000000000..7b69d1b85a --- /dev/null +++ b/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 diff --git a/stub-runner/stub-runner-messaging/stub-runner-messaging-stream/src/test/groovy/io/codearte/accurest/stubrunner/messaging/stream/StreamStubRunnerSpec.groovy b/stub-runner/stub-runner-messaging/stub-runner-messaging-stream/src/test/groovy/io/codearte/accurest/stubrunner/messaging/stream/StreamStubRunnerSpec.groovy index 722f41a36b..0233009851 100644 --- a/stub-runner/stub-runner-messaging/stub-runner-messaging-stream/src/test/groovy/io/codearte/accurest/stubrunner/messaging/stream/StreamStubRunnerSpec.groovy +++ b/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' ]) diff --git a/stub-runner/stub-runner-messaging/stub-runner-messaging-stream/src/test/resources/application.yml b/stub-runner/stub-runner-messaging/stub-runner-messaging-stream/src/test/resources/application.yml index 154bfe1b6a..e83bb967d7 100644 --- a/stub-runner/stub-runner-messaging/stub-runner-messaging-stream/src/test/resources/application.yml +++ b/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 \ No newline at end of file +stubrunner.stubs.ids: io.codearte.accurest.stubs:streamService + +spring: + cloud: + stream: + bindings: + output: + destination: returnBook + input: + destination: bookStorage \ No newline at end of file diff --git a/stub-runner/stub-runner-messaging/stub-runner-messaging-stream/src/test/resources/m2repo/repository/io/codearte/accurest/stubs/streamService/0.0.1-SNAPSHOT/streamService-0.0.1-SNAPSHOT-stubs.jar b/stub-runner/stub-runner-messaging/stub-runner-messaging-stream/src/test/resources/m2repo/repository/io/codearte/accurest/stubs/streamService/0.0.1-SNAPSHOT/streamService-0.0.1-SNAPSHOT-stubs.jar index 86c61d52ec..79d37267a7 100644 Binary files a/stub-runner/stub-runner-messaging/stub-runner-messaging-stream/src/test/resources/m2repo/repository/io/codearte/accurest/stubs/streamService/0.0.1-SNAPSHOT/streamService-0.0.1-SNAPSHOT-stubs.jar and b/stub-runner/stub-runner-messaging/stub-runner-messaging-stream/src/test/resources/m2repo/repository/io/codearte/accurest/stubs/streamService/0.0.1-SNAPSHOT/streamService-0.0.1-SNAPSHOT-stubs.jar differ