diff --git a/spring-cloud-contract-stub-runner/src/main/resources/META-INF/spring.factories b/spring-cloud-contract-stub-runner/src/main/resources/META-INF/spring.factories index 1556b08c2f..0f3c598395 100644 --- a/spring-cloud-contract-stub-runner/src/main/resources/META-INF/spring.factories +++ b/spring-cloud-contract-stub-runner/src/main/resources/META-INF/spring.factories @@ -4,6 +4,7 @@ org.springframework.cloud.contract.stubrunner.spring.StubRunnerConfiguration,\ org.springframework.cloud.contract.stubrunner.spring.cloud.StubRunnerSpringCloudAutoConfiguration,\ org.springframework.cloud.contract.stubrunner.spring.cloud.ribbon.StubRunnerRibbonAutoConfiguration,\ org.springframework.cloud.contract.stubrunner.messaging.integration.StubRunnerIntegrationConfiguration,\ +org.springframework.cloud.contract.stubrunner.messaging.stream.StubRunnerStreamConfiguration,\ org.springframework.cloud.contract.stubrunner.spring.cloud.zookeeper.StubRunnerSpringCloudZookeeperAutoConfiguration,\ org.springframework.cloud.contract.stubrunner.spring.cloud.eureka.StubRunnerSpringCloudEurekaAutoConfiguration,\ org.springframework.cloud.contract.stubrunner.spring.cloud.consul.StubRunnerSpringCloudConsulAutoConfiguration,\ diff --git a/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/amqp/RabbitMockConnectionFactoryAutoConfiguration.java b/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/amqp/RabbitMockConnectionFactoryAutoConfiguration.java index e92f02bbf6..e7ef0691fc 100644 --- a/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/amqp/RabbitMockConnectionFactoryAutoConfiguration.java +++ b/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/amqp/RabbitMockConnectionFactoryAutoConfiguration.java @@ -1,5 +1,6 @@ package org.springframework.cloud.contract.verifier.messaging.amqp; +import com.rabbitmq.client.AMQP; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; @@ -32,7 +33,7 @@ public class RabbitMockConnectionFactoryAutoConfiguration { @Bean public ConnectionFactory connectionFactory() { final Connection mockConnection = mock(Connection.class); - final Channel mockChannel = mock(Channel.class); + final AMQP.Queue.DeclareOk mockDeclareOk = mock(AMQP.Queue.DeclareOk.class); com.rabbitmq.client.ConnectionFactory mockConnectionFactory = mock(com.rabbitmq.client.ConnectionFactory.class, new Answer() { @Override public Object answer(InvocationOnMock invocationOnMock) throws Throwable { @@ -44,12 +45,20 @@ public class RabbitMockConnectionFactoryAutoConfiguration { } }); try { + final Channel mockChannel = mock(Channel.class, invocationOnMock -> { + if ("queueDeclare".equals(invocationOnMock.getMethod().getName())) { + return mockDeclareOk; + } + return Mockito.RETURNS_DEFAULTS.answer(invocationOnMock); + }); when(mockConnection.isOpen()).thenReturn(true); when(mockConnection.createChannel()).thenReturn(mockChannel); when(mockConnection.createChannel(Mockito.anyInt())).thenReturn(mockChannel); } catch (Exception e) { throw new RuntimeException(e); } - return new CachingConnectionFactory(mockConnectionFactory); + return new CachingConnectionFactory(mockConnectionFactory) { + + }; } } diff --git a/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/amqp/SpringAmqpStubMessages.java b/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/amqp/SpringAmqpStubMessages.java index bf2834f225..8c923121c7 100644 --- a/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/amqp/SpringAmqpStubMessages.java +++ b/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/amqp/SpringAmqpStubMessages.java @@ -29,6 +29,7 @@ import org.springframework.amqp.core.MessageListener; import org.springframework.amqp.core.MessagePropertiesBuilder; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; +import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener; import org.springframework.amqp.support.AmqpHeaders; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.cloud.contract.verifier.messaging.MessageVerifier; @@ -97,8 +98,18 @@ public class SpringAmqpStubMessages implements throw new IllegalStateException("no listeners found for destination " + destination); } for (SimpleMessageListenerContainer listenerContainer : listenerContainers) { - MessageListener messageListener = (MessageListener) listenerContainer.getMessageListener(); - messageListener.onMessage(message); + Object messageListener = listenerContainer.getMessageListener(); + if (messageListener instanceof ChannelAwareMessageListener && listenerContainer.getConnectionFactory() != null) { + try { + ((ChannelAwareMessageListener) messageListener).onMessage(message, + listenerContainer.getConnectionFactory().createConnection().createChannel(true)); + } + catch (Exception e) { + throw new RuntimeException(e); + } + } else { + ((MessageListener) messageListener).onMessage(message); + } } } diff --git a/tests/spring-cloud-contract-stub-runner-stream/src/test/groovy/org/springframework/cloud/contract/stubrunner/messaging/stream/StreamStubRunnerSpec.groovy b/tests/spring-cloud-contract-stub-runner-stream/src/test/groovy/org/springframework/cloud/contract/stubrunner/messaging/stream/StreamStubRunnerSpec.groovy index 2ee278ee21..32a71d28db 100644 --- a/tests/spring-cloud-contract-stub-runner-stream/src/test/groovy/org/springframework/cloud/contract/stubrunner/messaging/stream/StreamStubRunnerSpec.groovy +++ b/tests/spring-cloud-contract-stub-runner-stream/src/test/groovy/org/springframework/cloud/contract/stubrunner/messaging/stream/StreamStubRunnerSpec.groovy @@ -58,107 +58,107 @@ class StreamStubRunnerSpec extends Specification { def 'should download the stub and register a route for it'() { when: - // tag::client_send[] - messaging.send(new BookReturned('foo'), [sample: 'header'], 'bookStorage') - // end::client_send[] + // tag::client_send[] + messaging.send(new BookReturned('foo'), [sample: 'header'], 'bookStorage') + // end::client_send[] then: - // tag::client_receive[] - Message receivedMessage = messaging.receive('returnBook') - // end::client_receive[] + // tag::client_receive[] + Message receivedMessage = messaging.receive('returnBook') + // end::client_receive[] and: - // tag::client_receive_message[] - receivedMessage != null - assertJsons(receivedMessage.payload) - receivedMessage.headers.get('BOOK-NAME') == 'foo' - // end::client_receive_message[] + // tag::client_receive_message[] + receivedMessage != null + assertJsons(receivedMessage.payload) + receivedMessage.headers.get('BOOK-NAME') == 'foo' + // end::client_receive_message[] } def 'should trigger a message by label'() { when: - // tag::client_trigger[] - stubFinder.trigger('return_book_1') - // end::client_trigger[] + // tag::client_trigger[] + stubFinder.trigger('return_book_1') + // end::client_trigger[] then: - // tag::client_trigger_receive[] - Message receivedMessage = messaging.receive('returnBook') - // end::client_trigger_receive[] + // tag::client_trigger_receive[] + Message receivedMessage = messaging.receive('returnBook') + // end::client_trigger_receive[] and: - // tag::client_trigger_message[] - receivedMessage != null - assertJsons(receivedMessage.payload) - receivedMessage.headers.get('BOOK-NAME') == 'foo' - // end::client_trigger_message[] + // tag::client_trigger_message[] + receivedMessage != null + assertJsons(receivedMessage.payload) + receivedMessage.headers.get('BOOK-NAME') == 'foo' + // end::client_trigger_message[] } def 'should trigger a label for the existing groupId:artifactId'() { when: - // tag::trigger_group_artifact[] - stubFinder.trigger('org.springframework.cloud.contract.verifier.stubs:streamService', 'return_book_1') - // end::trigger_group_artifact[] + // tag::trigger_group_artifact[] + stubFinder.trigger('org.springframework.cloud.contract.verifier.stubs:streamService', 'return_book_1') + // end::trigger_group_artifact[] then: - Message receivedMessage = messaging.receive('returnBook') + Message receivedMessage = messaging.receive('returnBook') and: - receivedMessage != null - assertJsons(receivedMessage.payload) - receivedMessage.headers.get('BOOK-NAME') == 'foo' + receivedMessage != null + assertJsons(receivedMessage.payload) + receivedMessage.headers.get('BOOK-NAME') == 'foo' } def 'should trigger a label for the existing artifactId'() { when: - // tag::trigger_artifact[] - stubFinder.trigger('streamService', 'return_book_1') - // end::trigger_artifact[] + // tag::trigger_artifact[] + stubFinder.trigger('streamService', 'return_book_1') + // end::trigger_artifact[] then: - Message receivedMessage = messaging.receive('returnBook') + Message receivedMessage = messaging.receive('returnBook') and: - receivedMessage != null - assertJsons(receivedMessage.payload) - receivedMessage.headers.get('BOOK-NAME') == 'foo' + receivedMessage != null + assertJsons(receivedMessage.payload) + receivedMessage.headers.get('BOOK-NAME') == 'foo' } def 'should throw exception when missing label is passed'() { when: - stubFinder.trigger('missing label') + stubFinder.trigger('missing label') then: - thrown(IllegalArgumentException) + thrown(IllegalArgumentException) } def 'should throw exception when missing label and artifactid is passed'() { when: - stubFinder.trigger('some:service', 'return_book_1') + stubFinder.trigger('some:service', 'return_book_1') then: - thrown(IllegalArgumentException) + thrown(IllegalArgumentException) } def 'should trigger messages by running all triggers'() { when: - // tag::trigger_all[] - stubFinder.trigger() - // end::trigger_all[] + // tag::trigger_all[] + stubFinder.trigger() + // end::trigger_all[] then: - Message receivedMessage = messaging.receive('returnBook') + Message receivedMessage = messaging.receive('returnBook') and: - receivedMessage != null - assertJsons(receivedMessage.payload) - receivedMessage.headers.get('BOOK-NAME') == 'foo' + receivedMessage != null + assertJsons(receivedMessage.payload) + receivedMessage.headers.get('BOOK-NAME') == 'foo' } def 'should trigger a label with no output message'() { when: - // tag::trigger_no_output[] - messaging.send(new BookReturned('foo'), [sample: 'header'], 'delete') - // end::trigger_no_output[] + // tag::trigger_no_output[] + messaging.send(new BookReturned('foo'), [sample: 'header'], 'delete') + // end::trigger_no_output[] then: - noExceptionThrown() + noExceptionThrown() } def 'should not trigger a message that does not match input'() { when: - messaging.send(new BookReturned('not_matching'), [wrong: 'header_value'], 'bookStorage') + messaging.send(new BookReturned('not_matching'), [wrong: 'header_value'], 'bookStorage') then: - Message receivedMessage = messaging.receive('returnBook', 100, TimeUnit.MILLISECONDS) + Message receivedMessage = messaging.receive('returnBook', 100, TimeUnit.MILLISECONDS) and: - receivedMessage == null + receivedMessage == null } private boolean assertJsons(Object payload) { @@ -170,52 +170,52 @@ class StreamStubRunnerSpec extends Specification { } Contract dsl = - // tag::sample_dsl[] - Contract.make { - label 'return_book_1' - input { triggeredBy('bookReturnedTriggered()') } - outputMessage { - sentTo('returnBook') - body('''{ "bookName" : "foo" }''') - headers { header('BOOK-NAME', 'foo') } - } - } + // tag::sample_dsl[] + Contract.make { + label 'return_book_1' + input { triggeredBy('bookReturnedTriggered()') } + outputMessage { + sentTo('returnBook') + body('''{ "bookName" : "foo" }''') + headers { header('BOOK-NAME', 'foo') } + } + } // end::sample_dsl[] Contract dsl2 = - // tag::sample_dsl_2[] - Contract.make { - label 'return_book_2' - input { - messageFrom('bookStorage') - messageBody([ - bookName: 'foo' - ]) - messageHeaders { header('sample', 'header') } - } - outputMessage { - sentTo('returnBook') - body([ - bookName: 'foo' - ]) - headers { header('BOOK-NAME', 'foo') } - } - } + // tag::sample_dsl_2[] + Contract.make { + label 'return_book_2' + input { + messageFrom('bookStorage') + messageBody([ + bookName: 'foo' + ]) + messageHeaders { header('sample', 'header') } + } + outputMessage { + sentTo('returnBook') + body([ + bookName: 'foo' + ]) + headers { header('BOOK-NAME', 'foo') } + } + } // end::sample_dsl_2[] Contract dsl3 = - // tag::sample_dsl_3[] - Contract.make { - label 'delete_book' - input { - messageFrom('delete') - messageBody([ - bookName: 'foo' - ]) - messageHeaders { header('sample', 'header') } - assertThat('bookWasDeleted()') - } - } + // tag::sample_dsl_3[] + Contract.make { + label 'delete_book' + input { + messageFrom('delete') + messageBody([ + bookName: 'foo' + ]) + messageHeaders { header('sample', 'header') } + assertThat('bookWasDeleted()') + } + } // end::sample_dsl_3[]