From 4d45bdf31d114eb23b0d6c804373e2df2cfcb6a4 Mon Sep 17 00:00:00 2001 From: Marcin Grzejszczak Date: Mon, 16 Sep 2019 22:37:08 +0200 Subject: [PATCH] Added support for Spring Kafka fixes gh-877 --- README.adoc | 2 +- .../asciidoc/_project-features-messaging.adoc | 149 ++++++++- spring-cloud-contract-stub-runner/pom.xml | 10 + .../kafka/StubRunnerKafkaConfiguration.java | 146 ++++++++ .../kafka/StubRunnerKafkaMessageSelector.java | 259 +++++++++++++++ .../kafka/StubRunnerKafkaRouter.java | 121 +++++++ .../kafka/StubRunnerKafkaTransformer.java | 62 ++++ ...ubRunnerWireMockTestExecutionListener.java | 3 +- ...itional-spring-configuration-metadata.json | 18 + spring-cloud-contract-verifier/pom.xml | 10 + .../amqp/SpringAmqpStubMessages.java | 18 +- .../ContractVerifierKafkaConfiguration.java | 82 +++++ ...tVerifierKafkaStubMessagesInitializer.java | 63 ++++ .../messaging/kafka/KafkaStubMessages.java | 173 ++++++++++ .../kafka/KafkaStubMessagesInitializer.java | 44 +++ .../main/resources/META-INF/spring.factories | 1 + tests/pom.xml | 2 + .../pom.xml | 82 +++++ .../messaging/kafka/BookReturned.groovy | 34 ++ .../kafka/KafkaStubRunnerSpec.groovy | 311 ++++++++++++++++++ .../src/test/resources/application.yml | 21 ++ .../test/resources/stubs/bookDeleted.groovy | 13 + .../test/resources/stubs/bookReturned1.groovy | 13 + .../test/resources/stubs/bookReturned2.groovy | 21 ++ 24 files changed, 1639 insertions(+), 19 deletions(-) create mode 100644 spring-cloud-contract-stub-runner/src/main/java/org/springframework/cloud/contract/stubrunner/messaging/kafka/StubRunnerKafkaConfiguration.java create mode 100644 spring-cloud-contract-stub-runner/src/main/java/org/springframework/cloud/contract/stubrunner/messaging/kafka/StubRunnerKafkaMessageSelector.java create mode 100644 spring-cloud-contract-stub-runner/src/main/java/org/springframework/cloud/contract/stubrunner/messaging/kafka/StubRunnerKafkaRouter.java create mode 100644 spring-cloud-contract-stub-runner/src/main/java/org/springframework/cloud/contract/stubrunner/messaging/kafka/StubRunnerKafkaTransformer.java create mode 100644 spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/kafka/ContractVerifierKafkaConfiguration.java create mode 100644 spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/kafka/ContractVerifierKafkaStubMessagesInitializer.java create mode 100644 spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/kafka/KafkaStubMessages.java create mode 100644 spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/kafka/KafkaStubMessagesInitializer.java create mode 100644 tests/spring-cloud-contract-stub-runner-kafka/pom.xml create mode 100644 tests/spring-cloud-contract-stub-runner-kafka/src/test/groovy/org/springframework/cloud/contract/stubrunner/messaging/kafka/BookReturned.groovy create mode 100644 tests/spring-cloud-contract-stub-runner-kafka/src/test/groovy/org/springframework/cloud/contract/stubrunner/messaging/kafka/KafkaStubRunnerSpec.groovy create mode 100644 tests/spring-cloud-contract-stub-runner-kafka/src/test/resources/application.yml create mode 100644 tests/spring-cloud-contract-stub-runner-kafka/src/test/resources/stubs/bookDeleted.groovy create mode 100644 tests/spring-cloud-contract-stub-runner-kafka/src/test/resources/stubs/bookReturned1.groovy create mode 100644 tests/spring-cloud-contract-stub-runner-kafka/src/test/resources/stubs/bookReturned2.groovy diff --git a/README.adoc b/README.adoc index d8f382be32..82cc139b9a 100644 --- a/README.adoc +++ b/README.adoc @@ -198,7 +198,7 @@ image::https://raw.githubusercontent.com/spring-cloud/spring-cloud-build/{spring Go to `File` -> `Settings` -> `Other settings` -> `Checkstyle`. There click on the `+` icon in the `Configuration file` section. There, you'll have to define where the checkstyle rules should be picked from. In the image above, we've picked the rules from the cloned Spring Cloud Build repository. However, you can point to the Spring Cloud Build's GitHub repository (e.g. for the `checkstyle.xml` : `https://raw.githubusercontent.com/spring-cloud/spring-cloud-build/master/spring-cloud-build-tools/src/main/resources/checkstyle.xml`). We need to provide the following variables: -- `checkstyle.header.file` - please point it to the Spring Cloud Build's, `spring-cloud-build-tools/src/main/resources/checkstyle/checkstyle-header.txt` file either in your cloned repo or via the `https://raw.githubusercontent.com/spring-cloud/spring-cloud-build/master/spring-cloud-build-tools/src/main/resources/checkstyle-header.txt` URL. +- `checkstyle.header.file` - please point it to the Spring Cloud Build's, `spring-cloud-build-tools/src/main/resources/checkstyle-header.txt` file either in your cloned repo or via the `https://raw.githubusercontent.com/spring-cloud/spring-cloud-build/master/spring-cloud-build-tools/src/main/resources/checkstyle-header.txt` URL. - `checkstyle.suppressions.file` - default suppressions. Please point it to the Spring Cloud Build's, `spring-cloud-build-tools/src/checkstyle/checkstyle-suppressions.xml` file either in your cloned repo or via the `https://raw.githubusercontent.com/spring-cloud/spring-cloud-build/master/spring-cloud-build-tools/src/checkstyle/checkstyle-suppressions.xml` URL. - `checkstyle.additional.suppressions.file` - this variable corresponds to suppressions in your local project. E.g. you're working on `spring-cloud-contract`. Then point to the `project-root/src/checkstyle/checkstyle-suppressions.xml` folder. Example for `spring-cloud-contract` would be: `/home/username/spring-cloud-contract/src/checkstyle/checkstyle-suppressions.xml`. diff --git a/docs/src/main/asciidoc/_project-features-messaging.adoc b/docs/src/main/asciidoc/_project-features-messaging.adoc index 4c691a58ee..88ed38db41 100644 --- a/docs/src/main/asciidoc/_project-features-messaging.adoc +++ b/docs/src/main/asciidoc/_project-features-messaging.adoc @@ -101,7 +101,8 @@ You can use one of the following four integration configurations: * Spring Integration * Spring Cloud Stream * Spring AMQP -* Spring JMS +* Spring JMS (requires embedded broker) +* Spring Kafka (requires embedded broker) Since we use Spring Boot, if you have added one of these libraries to the classpath, all the messaging configuration is automatically set up. @@ -1150,4 +1151,150 @@ Since the route is set for you, you can send a message to the `{output_name}` de ---- include::{tests_path}/spring-cloud-contract-stub-runner-jms/src/test/groovy/org/springframework/cloud/contract/stubrunner/messaging/jms/JmsStubRunnerSpec.groovy[tags=trigger_no_output,indent=0] ---- +==== + +[[features-messaging-stub-runner-kafka]] +=== Consumer Side Messaging With Spring Kafka + +Spring Cloud Contract Stub Runner's messaging module provides an easy way to +integrate with Spring Kafka. + +The integration assumes that you have a running instance of a embedded Kafka broker (via the `spring-kafka-test` dependency). + +[[features-messaging-stub-runner-kafka-adding]] +==== Adding the Runner to the Project + +You need to have both Spring Kafka, Spring Kafka Test (to run the `@EmbeddedBroker`) and Spring Cloud Contract Stub Runner on the classpath. Remember to annotate your test class +with `@AutoConfigureStubRunner`. + +With Kafka integration, in order to poll for a single message we need to register a consumer upon Spring context startup. That may lead to a situation that, when you're on the consumer side, Stub Runner can register an additional consumer for the same group id and topic. That could lead to a situation that only one of the components would actually poll for the message. Since on the consumer side you have both the Spring Cloud Contract Stub Runner and Spring Cloud Contract Verifier classpath, we need to be able to switch off such behaviour. That's done automatically via the `stubrunner.kafka.initializer.enabled` flag, that will disable the Contact Verifier consumer registration. If your application is both the consumer and the producer of a kafka message, you might need to manually toggle that property to `false` in the base class of your generated tests. + +:input_name: input +:output_name: output + +[[features-messaging-stub-runner-kafka-example]] +==== Examples + +Assume that the stub structure looks as follows: + +==== +[source,bash,indent=0] +---- +├── stubs + ├── bookDeleted.groovy + ├── bookReturned1.groovy + └── bookReturned2.groovy + +---- +==== + +Further assume the following test configuration (notice the `spring.kafka.bootstrap-servers` pointing to the embedded broker's IP via `${spring.embedded.kafka.brokers}`): + +==== +[source,yml,indent=0] +---- +stubrunner: + repository-root: stubs:classpath:/stubs/ + ids: my:stubs + stubs-mode: remote +spring: + kafka: + bootstrap-servers: ${spring.embedded.kafka.brokers} + producer: + properties: + "value.serializer": "org.springframework.kafka.support.serializer.JsonSerializer" + "spring.json.trusted.packages": "*" + consumer: + properties: + "value.deserializer": "org.springframework.kafka.support.serializer.JsonDeserializer" + "value.serializer": "org.springframework.kafka.support.serializer.JsonSerializer" + "spring.json.trusted.packages": "*" + group-id: groupId +---- +==== + +Now consider the following contracts (we number them 1 and 2): + +==== +[source,groovy] +---- +include::{tests_path}/spring-cloud-contract-stub-runner-kafka/src/test/groovy/org/springframework/cloud/contract/stubrunner/messaging/kafka/KafkaStubRunnerSpec.groovy[tags=sample_dsl,indent=0] +---- + +[source,groovy] +---- +include::{tests_path}/spring-cloud-contract-stub-runner-kafka/src/test/groovy/org/springframework/cloud/contract/stubrunner/messaging/kafka/KafkaStubRunnerSpec.groovy[tags=sample_dsl_2,indent=0] +---- +==== + +[[features-messaging-stub-runner-kafka-scenario1]] +===== Scenario 1 (No Input Message) + +To trigger a message from the `return_book_1` label, we use the `StubTigger` interface, as follows: + +==== +[source,groovy] +---- +include::{tests_path}/spring-cloud-contract-stub-runner-kafka/src/test/groovy/org/springframework/cloud/contract/stubrunner/messaging/kafka/KafkaStubRunnerSpec.groovy[tags=client_trigger,indent=0] +---- +==== + +Next, we want to listen to the output of the message sent to `{output_name}`: + +==== +[source,groovy] +---- +include::{tests_path}/spring-cloud-contract-stub-runner-kafka/src/test/groovy/org/springframework/cloud/contract/stubrunner/messaging/kafka/KafkaStubRunnerSpec.groovy[tags=client_trigger_receive,indent=0] +---- +==== + +The received message would then pass the following assertions: + +==== +[source,groovy] +---- +include::{tests_path}/spring-cloud-contract-stub-runner-kafka/src/test/groovy/org/springframework/cloud/contract/stubrunner/messaging/kafka/KafkaStubRunnerSpec.groovy[tags=client_trigger_message,indent=0] +---- +==== + +[[features-messaging-stub-runner-kafka-scenario2]] +===== Scenario 2 (Output Triggered by Input) + +Since the route is set for you, you can send a message to the `{output_name}` destination. + +==== +[source,groovy] +---- +include::{tests_path}/spring-cloud-contract-stub-runner-kafka/src/test/groovy/org/springframework/cloud/contract/stubrunner/messaging/kafka/KafkaStubRunnerSpec.groovy[tags=client_send,indent=0] +---- +==== + +Next, we want to listen to the output of the message sent to `{output_name}`, as follows: + +==== +[source,groovy] +---- +include::{tests_path}/spring-cloud-contract-stub-runner-kafka/src/test/groovy/org/springframework/cloud/contract/stubrunner/messaging/kafka/KafkaStubRunnerSpec.groovy[tags=client_receive,indent=0] +---- +==== + +The received message would pass the following assertions: + +==== +[source,groovy] +---- +include::{tests_path}/spring-cloud-contract-stub-runner-kafka/src/test/groovy/org/springframework/cloud/contract/stubrunner/messaging/kafka/KafkaStubRunnerSpec.groovy[tags=client_receive_message,indent=0] +---- +==== + +[[features-messaging-stub-runner-kafka-scenario3]] +===== Scenario 3 (Input with No Output) + +Since the route is set for you, you can send a message to the `{output_name}` destination, as follows: + +==== +[source,groovy] +---- +include::{tests_path}/spring-cloud-contract-stub-runner-kafka/src/test/groovy/org/springframework/cloud/contract/stubrunner/messaging/kafka/KafkaStubRunnerSpec.groovy[tags=trigger_no_output,indent=0] +---- ==== \ No newline at end of file diff --git a/spring-cloud-contract-stub-runner/pom.xml b/spring-cloud-contract-stub-runner/pom.xml index f4b26eaa55..3100e10a90 100644 --- a/spring-cloud-contract-stub-runner/pom.xml +++ b/spring-cloud-contract-stub-runner/pom.xml @@ -70,6 +70,16 @@ spring-boot-starter-activemq true + + org.springframework.kafka + spring-kafka + true + + + org.springframework.kafka + spring-kafka-test + true + org.apache.camel camel-spring-boot-starter diff --git a/spring-cloud-contract-stub-runner/src/main/java/org/springframework/cloud/contract/stubrunner/messaging/kafka/StubRunnerKafkaConfiguration.java b/spring-cloud-contract-stub-runner/src/main/java/org/springframework/cloud/contract/stubrunner/messaging/kafka/StubRunnerKafkaConfiguration.java new file mode 100644 index 0000000000..47e76d9675 --- /dev/null +++ b/spring-cloud-contract-stub-runner/src/main/java/org/springframework/cloud/contract/stubrunner/messaging/kafka/StubRunnerKafkaConfiguration.java @@ -0,0 +1,146 @@ +/* + * Copyright 2013-2019 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 + * + * https://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.contract.stubrunner.messaging.kafka; + +import java.util.Collection; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Map.Entry; + +import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; +import org.springframework.boot.autoconfigure.AutoConfigureBefore; +import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; +import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; +import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.cloud.contract.spec.Contract; +import org.springframework.cloud.contract.stubrunner.BatchStubRunner; +import org.springframework.cloud.contract.stubrunner.StubConfiguration; +import org.springframework.cloud.contract.verifier.messaging.kafka.ContractVerifierKafkaConfiguration; +import org.springframework.cloud.contract.verifier.messaging.kafka.KafkaStubMessagesInitializer; +import org.springframework.cloud.contract.verifier.util.MapConverter; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.kafka.core.ConsumerFactory; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.listener.ContainerProperties; +import org.springframework.kafka.listener.GenericMessageListener; +import org.springframework.kafka.listener.KafkaMessageListenerContainer; +import org.springframework.kafka.test.EmbeddedKafkaBroker; +import org.springframework.util.LinkedMultiValueMap; +import org.springframework.util.MultiValueMap; +import org.springframework.util.StringUtils; + +/** + * Spring Integration configuration that iterates over the downloaded Groovy DSLs and + * registers a flow for each DSL. + * + * @author Marcin Grzejszczak + */ +@Configuration +@ConditionalOnClass({ KafkaTemplate.class, EmbeddedKafkaBroker.class }) +@ConditionalOnProperty(name = "stubrunner.kafka.enabled", havingValue = "true", + matchIfMissing = true) +@ConditionalOnBean(EmbeddedKafkaBroker.class) +@AutoConfigureBefore(ContractVerifierKafkaConfiguration.class) +public class StubRunnerKafkaConfiguration { + + @Bean + @ConditionalOnMissingBean + @ConditionalOnProperty(name = "stubrunner.kafka.initializer.enabled", + havingValue = "true", matchIfMissing = true) + KafkaStubMessagesInitializer stubRunnerKafkaStubMessagesInitializer() { + return (broker, kafkaProperties) -> new HashMap<>(); + } + + @Bean + @ConditionalOnMissingBean(name = "stubFlowRegistrar") + public FlowRegistrar stubFlowRegistrar(ConfigurableListableBeanFactory beanFactory, + BatchStubRunner batchStubRunner) { + Map> contracts = batchStubRunner + .getContracts(); + for (Entry> entry : contracts + .entrySet()) { + StubConfiguration key = entry.getKey(); + Collection value = entry.getValue(); + String name = key.getGroupId() + "_" + key.getArtifactId(); + MultiValueMap map = new LinkedMultiValueMap<>(); + for (Contract dsl : value) { + if (dsl == null) { + continue; + } + if (dsl.getInput() != null && dsl.getInput().getMessageFrom() != null + && StringUtils.hasText( + dsl.getInput().getMessageFrom().getClientValue())) { + String from = dsl.getInput().getMessageFrom().getClientValue(); + map.add(from, dsl); + } + } + for (Entry> entries : map.entrySet()) { + List matchingContracts = entries.getValue(); + final String flowName = name + "_" + entries.getKey() + "_" + + Math.abs(matchingContracts.hashCode()); + // listener + StubRunnerKafkaRouter router = new StubRunnerKafkaRouter( + matchingContracts, beanFactory); + StubRunnerKafkaRouter listener = (StubRunnerKafkaRouter) beanFactory + .initializeBean(router, flowName); + beanFactory.registerSingleton(flowName, listener); + registerContainers(beanFactory, matchingContracts, flowName, listener); + } + + } + return new FlowRegistrar(); + } + + private void registerContainers(ConfigurableListableBeanFactory beanFactory, + List matchingContracts, String flowName, + StubRunnerKafkaRouter listener) { + // listener's container + ConsumerFactory consumerFactory = beanFactory.getBean(ConsumerFactory.class); + for (Contract matchingContract : matchingContracts) { + if (matchingContract.getInput() == null) { + continue; + } + String destination = MapConverter.getStubSideValuesForNonBody( + matchingContract.getInput().getMessageFrom()).toString(); + ContainerProperties containerProperties = new ContainerProperties( + destination); + KafkaMessageListenerContainer container = listenerContainer(consumerFactory, + containerProperties, listener); + String containerName = flowName + ".container"; + Object initializedContainer = beanFactory.initializeBean(container, + containerName); + beanFactory.registerSingleton(containerName, initializedContainer); + } + } + + private KafkaMessageListenerContainer listenerContainer( + ConsumerFactory consumerFactory, ContainerProperties containerProperties, + GenericMessageListener listener) { + KafkaMessageListenerContainer container = new KafkaMessageListenerContainer( + consumerFactory, containerProperties); + container.setupMessageListener(listener); + return container; + } + + static class FlowRegistrar { + + } + +} diff --git a/spring-cloud-contract-stub-runner/src/main/java/org/springframework/cloud/contract/stubrunner/messaging/kafka/StubRunnerKafkaMessageSelector.java b/spring-cloud-contract-stub-runner/src/main/java/org/springframework/cloud/contract/stubrunner/messaging/kafka/StubRunnerKafkaMessageSelector.java new file mode 100644 index 0000000000..f60ef40592 --- /dev/null +++ b/spring-cloud-contract-stub-runner/src/main/java/org/springframework/cloud/contract/stubrunner/messaging/kafka/StubRunnerKafkaMessageSelector.java @@ -0,0 +1,259 @@ +/* + * Copyright 2013-2019 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 + * + * https://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.contract.stubrunner.messaging.kafka; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.WeakHashMap; +import java.util.regex.Pattern; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.jayway.jsonpath.DocumentContext; +import com.jayway.jsonpath.JsonPath; +import com.toomuchcoding.jsonassert.JsonAssertion; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.cloud.contract.spec.Contract; +import org.springframework.cloud.contract.spec.internal.BodyMatcher; +import org.springframework.cloud.contract.spec.internal.BodyMatchers; +import org.springframework.cloud.contract.spec.internal.FromFileProperty; +import org.springframework.cloud.contract.spec.internal.Header; +import org.springframework.cloud.contract.spec.internal.RegexProperty; +import org.springframework.cloud.contract.verifier.messaging.internal.ContractVerifierObjectMapper; +import org.springframework.cloud.contract.verifier.util.ContentType; +import org.springframework.cloud.contract.verifier.util.ContentUtils; +import org.springframework.cloud.contract.verifier.util.JsonPaths; +import org.springframework.cloud.contract.verifier.util.JsonToJsonPathsConverter; +import org.springframework.cloud.contract.verifier.util.MapConverter; +import org.springframework.cloud.contract.verifier.util.MethodBufferingJsonVerifiable; +import org.springframework.messaging.Message; + +/** + * Passes through a message that matches the one defined in the DSL. + * + * @author Marcin Grzejszczak + */ +class StubRunnerKafkaMessageSelector { + + private static final Map, Contract> CACHE = Collections + .synchronizedMap(new WeakHashMap<>()); + + private static final Log log = LogFactory + .getLog(StubRunnerKafkaMessageSelector.class); + + private final List groovyDsls; + + private final ContractVerifierObjectMapper objectMapper = new ContractVerifierObjectMapper(); + + StubRunnerKafkaMessageSelector(List groovyDsls) { + this.groovyDsls = groovyDsls; + } + + Contract matchingContract(Message message) { + if (CACHE.containsKey(message)) { + return CACHE.get(message); + } + Contract contract = getContract(message); + if (contract != null) { + CACHE.put(message, contract); + } + return contract; + } + + void updateCache(Message message, Contract contract) { + CACHE.put(message, contract); + } + + private Contract getContract(Message message) { + for (Contract groovyDsl : this.groovyDsls) { + Contract contract = matchContract(message, groovyDsl); + if (contract != null) { + return contract; + } + } + return null; + } + + private Contract matchContract(Message message, Contract groovyDsl) { + List unmatchedHeaders = headersMatch(message, groovyDsl); + if (!unmatchedHeaders.isEmpty()) { + if (log.isDebugEnabled()) { + log.debug("Contract [" + groovyDsl + + "] hasn't matched the following headers " + unmatchedHeaders); + } + return null; + } + Object inputMessage = message.getPayload(); + Object dslBody = MapConverter + .getStubSideValues(groovyDsl.getInput().getMessageBody()); + if (dslBody instanceof FromFileProperty) { + if (log.isDebugEnabled()) { + log.debug("Will compare file content"); + } + FromFileProperty property = (FromFileProperty) dslBody; + if (property.isString()) { + // continue processing as if body was pure string + dslBody = property.asString(); + } + else if (!(inputMessage instanceof byte[])) { + if (log.isDebugEnabled()) { + log.debug( + "Contract provided byte comparison, but the input message is of type [" + + inputMessage.getClass() + + "]. Can't compare the two."); + } + return null; + } + else { + boolean matches = Arrays.equals(property.asBytes(), + (byte[]) inputMessage); + if (log.isDebugEnabled() && !matches) { + log.debug( + "Contract provided byte comparison, but the byte arrays don't match"); + } + return matches ? groovyDsl : null; + } + } + if (matchViaContent(groovyDsl, inputMessage, dslBody)) { + return groovyDsl; + } + return null; + } + + private boolean matchViaContent(Contract groovyDsl, Object inputMessage, + Object dslBody) { + boolean matches; + ContentType type = ContentUtils.getClientContentType(inputMessage, + groovyDsl.getInput().getMessageHeaders()); + if (type == ContentType.JSON) { + BodyMatchers matchers = groovyDsl.getInput().getBodyMatchers(); + matches = matchesForJsonPayload(groovyDsl, inputMessage, matchers, dslBody); + } + else if (dslBody instanceof RegexProperty && inputMessage instanceof String) { + Pattern pattern = ((RegexProperty) dslBody).getPattern(); + matches = pattern.matcher((String) inputMessage).matches(); + bodyUnmatchedLog(dslBody, matches, pattern); + } + else { + matches = dslBody.equals(inputMessage); + bodyUnmatchedLog(dslBody, matches, inputMessage); + } + return matches; + } + + private void bodyUnmatchedLog(Object dslBody, boolean matches, Object pattern) { + if (log.isDebugEnabled() && !matches) { + log.debug("Body was supposed to " + unmatchedText(pattern) + + " but the value is [" + dslBody.toString() + "]"); + } + } + + private boolean matchesForJsonPayload(Contract groovyDsl, Object inputMessage, + BodyMatchers matchers, Object dslBody) { + Object matchingInputMessage = JsonToJsonPathsConverter + .removeMatchingJsonPaths(dslBody, matchers); + JsonPaths jsonPaths = JsonToJsonPathsConverter + .transformToJsonPathWithStubsSideValuesAndNoArraySizeCheck( + matchingInputMessage); + DocumentContext parsedJson; + try { + parsedJson = JsonPath + .parse(this.objectMapper.writeValueAsString(inputMessage)); + } + catch (JsonProcessingException e) { + throw new IllegalStateException("Cannot serialize to JSON", e); + } + List unmatchedJsonPath = new ArrayList<>(); + boolean matches = true; + for (MethodBufferingJsonVerifiable path : jsonPaths) { + matches &= matchesJsonPath(unmatchedJsonPath, parsedJson, path.jsonPath()); + } + if (matchers != null && matchers.hasMatchers()) { + for (BodyMatcher matcher : matchers.matchers()) { + String jsonPath = JsonToJsonPathsConverter + .convertJsonPathAndRegexToAJsonPath(matcher, dslBody); + matches &= matchesJsonPath(unmatchedJsonPath, parsedJson, jsonPath); + } + } + if (!unmatchedJsonPath.isEmpty()) { + if (log.isDebugEnabled()) { + log.debug("Contract [" + groovyDsl + "] didn't match the body due to " + + unmatchedJsonPath); + } + } + return matches; + } + + private boolean matchesJsonPath(List unmatchedJsonPath, + DocumentContext parsedJson, String jsonPath) { + try { + JsonAssertion.assertThat(parsedJson).matchesJsonPath(jsonPath); + return true; + } + catch (Exception e) { + unmatchedJsonPath.add(e.getLocalizedMessage()); + return false; + } + } + + private List headersMatch(Message message, Contract groovyDsl) { + List unmatchedHeaders = new ArrayList<>(); + Map headers = message.getHeaders(); + for (Header it : groovyDsl.getInput().getMessageHeaders().getEntries()) { + String name = it.getName(); + Object value = it.getClientValue(); + Object valueInHeader = headers.get(name); + valueInHeader = valueInHeader instanceof byte[] + ? fromByte((byte[]) valueInHeader) : valueInHeader; + boolean matches; + if (value instanceof RegexProperty) { + Pattern pattern = ((RegexProperty) value).getPattern(); + matches = pattern.matcher(valueInHeader.toString()).matches(); + } + else { + matches = valueInHeader != null + && valueInHeader.toString().equals(value.toString()); + } + if (!matches) { + unmatchedHeaders.add("Header with name [" + name + "] was supposed to " + + unmatchedText(value) + " but the value is [" + + (valueInHeader != null ? valueInHeader.toString() : "null") + + "]"); + } + } + return unmatchedHeaders; + } + + private String fromByte(byte[] valueInHeader) { + String string = new String(valueInHeader); + if (string.startsWith("\"") && string.endsWith("\"")) { + return string.substring(1, string.length() - 1); + } + return string; + } + + private String unmatchedText(Object expectedValue) { + return expectedValue instanceof RegexProperty + ? "match pattern [" + ((RegexProperty) expectedValue).pattern() + "]" + : "be equal to [" + expectedValue + "]"; + } + +} diff --git a/spring-cloud-contract-stub-runner/src/main/java/org/springframework/cloud/contract/stubrunner/messaging/kafka/StubRunnerKafkaRouter.java b/spring-cloud-contract-stub-runner/src/main/java/org/springframework/cloud/contract/stubrunner/messaging/kafka/StubRunnerKafkaRouter.java new file mode 100644 index 0000000000..014d26b306 --- /dev/null +++ b/spring-cloud-contract-stub-runner/src/main/java/org/springframework/cloud/contract/stubrunner/messaging/kafka/StubRunnerKafkaRouter.java @@ -0,0 +1,121 @@ +/* + * Copyright 2013-2019 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 + * + * https://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.contract.stubrunner.messaging.kafka; + +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.common.header.Header; +import org.apache.kafka.common.header.Headers; + +import org.springframework.beans.factory.BeanFactory; +import org.springframework.cloud.contract.spec.Contract; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.listener.MessageListener; +import org.springframework.kafka.support.Acknowledgment; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHeaders; +import org.springframework.messaging.support.MessageBuilder; + +/** + * @author Marcin Grzejszczak + */ +class StubRunnerKafkaRouter implements MessageListener { + + private static final Log log = LogFactory.getLog(StubRunnerKafkaRouter.class); + + private final StubRunnerKafkaMessageSelector selector; + + private final BeanFactory beanFactory; + + private final List contracts; + + private KafkaTemplate kafkaTemplate; + + StubRunnerKafkaRouter(List groovyDsls, BeanFactory beanFactory) { + this.selector = new StubRunnerKafkaMessageSelector(groovyDsls); + this.beanFactory = beanFactory; + this.contracts = groovyDsls; + } + + private KafkaTemplate kafkaTemplate() { + if (this.kafkaTemplate == null) { + this.kafkaTemplate = this.beanFactory.getBean(KafkaTemplate.class); + } + return this.kafkaTemplate; + } + + @Override + public void onMessage(ConsumerRecord data) { + if (log.isDebugEnabled()) { + log.debug("Received message [" + data + "]"); + } + Message message = MessageBuilder.createMessage(data.value(), + headers(data.headers())); + Contract dsl = this.selector.matchingContract(message); + if (dsl != null && dsl.getOutputMessage() != null + && dsl.getOutputMessage().getSentTo() != null) { + String destination = dsl.getOutputMessage().getSentTo().getClientValue(); + if (log.isDebugEnabled()) { + log.debug( + "Found a matching contract with an output message. Will send it to the [" + + destination + "] destination"); + } + Message transform = new StubRunnerKafkaTransformer(this.contracts) + .transform(dsl); + String defaultTopic = kafkaTemplate().getDefaultTopic(); + try { + kafkaTemplate().setDefaultTopic(destination); + kafkaTemplate().send(transform); + } + finally { + kafkaTemplate().setDefaultTopic(defaultTopic); + } + } + } + + private MessageHeaders headers(Headers headers) { + Map map = new HashMap<>(); + for (Header header : headers) { + map.put(header.key(), header.value()); + } + return new MessageHeaders(map); + } + + @Override + public void onMessage(ConsumerRecord data, + Acknowledgment acknowledgment) { + onMessage(data); + } + + @Override + public void onMessage(ConsumerRecord data, Consumer consumer) { + onMessage(data); + } + + @Override + public void onMessage(ConsumerRecord data, + Acknowledgment acknowledgment, Consumer consumer) { + onMessage(data); + } + +} diff --git a/spring-cloud-contract-stub-runner/src/main/java/org/springframework/cloud/contract/stubrunner/messaging/kafka/StubRunnerKafkaTransformer.java b/spring-cloud-contract-stub-runner/src/main/java/org/springframework/cloud/contract/stubrunner/messaging/kafka/StubRunnerKafkaTransformer.java new file mode 100644 index 0000000000..9260e1c686 --- /dev/null +++ b/spring-cloud-contract-stub-runner/src/main/java/org/springframework/cloud/contract/stubrunner/messaging/kafka/StubRunnerKafkaTransformer.java @@ -0,0 +1,62 @@ +/* + * Copyright 2013-2019 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 + * + * https://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.contract.stubrunner.messaging.kafka; + +import java.util.List; +import java.util.Map; + +import org.springframework.cloud.contract.spec.Contract; +import org.springframework.cloud.contract.spec.internal.FromFileProperty; +import org.springframework.cloud.contract.verifier.util.BodyExtractor; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHeaders; +import org.springframework.messaging.support.MessageBuilder; + +/** + * Sends forward a message defined in the DSL. + * + * @author Marcin Grzejszczak + */ +class StubRunnerKafkaTransformer { + + private final StubRunnerKafkaMessageSelector selector; + + StubRunnerKafkaTransformer(List groovyDsls) { + this.selector = new StubRunnerKafkaMessageSelector(groovyDsls); + } + + public Message transform(Contract groovyDsl) { + Object outputBody = outputBody(groovyDsl); + Map headers = groovyDsl.getOutputMessage().getHeaders() + .asStubSideMap(); + Message newMessage = MessageBuilder.createMessage(outputBody, + new MessageHeaders(headers)); + this.selector.updateCache(newMessage, groovyDsl); + return newMessage; + } + + private Object outputBody(Contract groovyDsl) { + Object outputBody = BodyExtractor + .extractClientValueFromBody(groovyDsl.getOutputMessage().getBody()); + if (outputBody instanceof FromFileProperty) { + FromFileProperty property = (FromFileProperty) outputBody; + return property.asBytes(); + } + return BodyExtractor.extractStubValueFrom(outputBody); + } + +} diff --git a/spring-cloud-contract-stub-runner/src/main/java/org/springframework/cloud/contract/stubrunner/provider/wiremock/StubRunnerWireMockTestExecutionListener.java b/spring-cloud-contract-stub-runner/src/main/java/org/springframework/cloud/contract/stubrunner/provider/wiremock/StubRunnerWireMockTestExecutionListener.java index c47a1dc17d..88c91b1f15 100644 --- a/spring-cloud-contract-stub-runner/src/main/java/org/springframework/cloud/contract/stubrunner/provider/wiremock/StubRunnerWireMockTestExecutionListener.java +++ b/spring-cloud-contract-stub-runner/src/main/java/org/springframework/cloud/contract/stubrunner/provider/wiremock/StubRunnerWireMockTestExecutionListener.java @@ -46,7 +46,8 @@ public final class StubRunnerWireMockTestExecutionListener } return; } - if (WireMockHttpServerStub.SERVERS.values().stream().noneMatch(p -> p.random)) { + if (!WireMockHttpServerStub.SERVERS.isEmpty() && WireMockHttpServerStub.SERVERS + .values().stream().noneMatch(p -> p.random)) { if (log.isWarnEnabled()) { log.warn("You've used fixed ports for WireMock setup - " + "will mark context as dirty. Please use random ports, as much " diff --git a/spring-cloud-contract-stub-runner/src/main/resources/META-INF/additional-spring-configuration-metadata.json b/spring-cloud-contract-stub-runner/src/main/resources/META-INF/additional-spring-configuration-metadata.json index 188fd5c1c5..f7cda2fca5 100644 --- a/spring-cloud-contract-stub-runner/src/main/resources/META-INF/additional-spring-configuration-metadata.json +++ b/spring-cloud-contract-stub-runner/src/main/resources/META-INF/additional-spring-configuration-metadata.json @@ -65,6 +65,24 @@ "type": "java.lang.Boolean", "description": "Whether to enable Stub Runner integration with Spring Cloud Stream.", "defaultValue": true + }, + { + "name": "stubrunner.jms.enabled", + "type": "java.lang.Boolean", + "description": "Whether to enable Stub Runner integration with Spring JMS.", + "defaultValue": true + }, + { + "name": "stubrunner.kafka.enabled", + "type": "java.lang.Boolean", + "description": "Whether to enable Stub Runner integration with Spring Kafka.", + "defaultValue": true + }, + { + "name": "stubrunner.kafka.initializer.enabled", + "type": "java.lang.Boolean", + "description": "Whether to allow Stub Runner to take care of polling for messages instead of the KafkaStubMessages component. The latter should be used only on the producer side.", + "defaultValue": true } ] } diff --git a/spring-cloud-contract-verifier/pom.xml b/spring-cloud-contract-verifier/pom.xml index 1ca7d39637..a3f8b826e8 100644 --- a/spring-cloud-contract-verifier/pom.xml +++ b/spring-cloud-contract-verifier/pom.xml @@ -36,6 +36,16 @@ spring-jms true + + org.springframework.kafka + spring-kafka + true + + + org.springframework.kafka + spring-kafka-test + true + jakarta.jms jakarta.jms-api 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 5e0f48632a..8e6c45be99 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 @@ -75,14 +75,7 @@ public class SpringAmqpStubMessages implements MessageVerifier { Assert.isTrue( mockingDetails(rabbitTemplate).isSpy() || mockingDetails(rabbitTemplate).isMock(), - "StubRunner AMQP will work only if RabbiTemplate is a spy"); // we get - // send - // messages - // by - // capturing - // arguments - // on the - // spy + "StubRunner AMQP will work only if RabbiTemplate is a spy"); this.rabbitTemplate = rabbitTemplate; this.messageListenerAccessor = messageListenerAccessor; } @@ -95,14 +88,7 @@ public class SpringAmqpStubMessages implements MessageVerifier { Assert.isTrue( mockingDetails(rabbitTemplate).isSpy() || mockingDetails(rabbitTemplate).isMock(), - "StubRunner AMQP will work only if RabbiTemplate is a spy"); // we get - // send - // messages - // by - // capturing - // arguments - // on the - // spy + "StubRunner AMQP will work only if RabbiTemplate is a spy"); this.rabbitTemplate = rabbitTemplate; this.messageListenerAccessor = messageListenerAccessor; this.rabbitProperties = rabbitProperties; diff --git a/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/kafka/ContractVerifierKafkaConfiguration.java b/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/kafka/ContractVerifierKafkaConfiguration.java new file mode 100644 index 0000000000..500fd43a5a --- /dev/null +++ b/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/kafka/ContractVerifierKafkaConfiguration.java @@ -0,0 +1,82 @@ +/* + * Copyright 2013-2019 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 + * + * https://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.contract.verifier.messaging.kafka; + +import org.springframework.boot.autoconfigure.AutoConfigureBefore; +import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; +import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; +import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.boot.autoconfigure.kafka.KafkaProperties; +import org.springframework.cloud.contract.verifier.messaging.MessageVerifier; +import org.springframework.cloud.contract.verifier.messaging.integration.ContractVerifierIntegrationConfiguration; +import org.springframework.cloud.contract.verifier.messaging.internal.ContractVerifierMessage; +import org.springframework.cloud.contract.verifier.messaging.internal.ContractVerifierMessaging; +import org.springframework.cloud.contract.verifier.messaging.noop.NoOpContractVerifierAutoConfiguration; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.test.EmbeddedKafkaBroker; +import org.springframework.messaging.Message; + +/** + * @author Marcin Grzejszczak + */ +@Configuration +@ConditionalOnClass({ KafkaTemplate.class, EmbeddedKafkaBroker.class }) +@ConditionalOnProperty(name = "stubrunner.kafka.enabled", havingValue = "true", + matchIfMissing = true) +@AutoConfigureBefore({ ContractVerifierIntegrationConfiguration.class, + NoOpContractVerifierAutoConfiguration.class }) +@ConditionalOnBean(EmbeddedKafkaBroker.class) +public class ContractVerifierKafkaConfiguration { + + @Bean + @ConditionalOnMissingBean + MessageVerifier> contractVerifierKafkaMessageExchange( + KafkaTemplate kafkaTemplate, EmbeddedKafkaBroker broker, + KafkaProperties kafkaProperties, KafkaStubMessagesInitializer initializer) { + return new KafkaStubMessages(kafkaTemplate, broker, kafkaProperties, initializer); + } + + @Bean + @ConditionalOnMissingBean + KafkaStubMessagesInitializer contractVerifierKafkaStubMessagesInitializer() { + return new ContractVerifierKafkaStubMessagesInitializer(); + } + + @Bean + @ConditionalOnMissingBean + ContractVerifierMessaging> contractVerifierKafkaMessaging( + MessageVerifier> exchange) { + return new ContractVerifierKafkaHelper(exchange); + } + +} + +class ContractVerifierKafkaHelper extends ContractVerifierMessaging> { + + ContractVerifierKafkaHelper(MessageVerifier> exchange) { + super(exchange); + } + + @Override + protected ContractVerifierMessage convert(Message message) { + return new ContractVerifierMessage(message.getPayload(), message.getHeaders()); + } + +} diff --git a/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/kafka/ContractVerifierKafkaStubMessagesInitializer.java b/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/kafka/ContractVerifierKafkaStubMessagesInitializer.java new file mode 100644 index 0000000000..481ae016aa --- /dev/null +++ b/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/kafka/ContractVerifierKafkaStubMessagesInitializer.java @@ -0,0 +1,63 @@ +/* + * Copyright 2013-2019 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 + * + * https://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.contract.verifier.messaging.kafka; + +import java.util.HashMap; +import java.util.Map; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerConfig; + +import org.springframework.boot.autoconfigure.kafka.KafkaProperties; +import org.springframework.kafka.core.DefaultKafkaConsumerFactory; +import org.springframework.kafka.test.EmbeddedKafkaBroker; +import org.springframework.kafka.test.utils.KafkaTestUtils; + +class ContractVerifierKafkaStubMessagesInitializer + implements KafkaStubMessagesInitializer { + + private static final Log log = LogFactory + .getLog(ContractVerifierKafkaStubMessagesInitializer.class); + + @Override + public Map initialize(EmbeddedKafkaBroker broker, + KafkaProperties kafkaProperties) { + Map map = new HashMap<>(); + for (String topic : broker.getTopics()) { + map.put(topic, prepareListener(broker, topic, kafkaProperties)); + } + return map; + } + + private Consumer prepareListener(EmbeddedKafkaBroker broker, String destination, + KafkaProperties kafkaProperties) { + Map consumerProperties = KafkaTestUtils.consumerProps( + kafkaProperties.getConsumer().getGroupId(), "false", broker); + consumerProperties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + DefaultKafkaConsumerFactory consumerFactory = new DefaultKafkaConsumerFactory<>( + consumerProperties); + Consumer consumer = consumerFactory.createConsumer(); + broker.consumeFromAnEmbeddedTopic(consumer, destination); + if (log.isDebugEnabled()) { + log.debug("Prepared consumer for destination [" + destination + "]"); + } + return consumer; + } + +} diff --git a/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/kafka/KafkaStubMessages.java b/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/kafka/KafkaStubMessages.java new file mode 100644 index 0000000000..190ea4f309 --- /dev/null +++ b/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/kafka/KafkaStubMessages.java @@ -0,0 +1,173 @@ +/* + * Copyright 2013-2019 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 + * + * https://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.contract.verifier.messaging.kafka; + +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.TimeUnit; + +import net.minidev.json.JSONObject; +import net.minidev.json.parser.JSONParser; +import net.minidev.json.parser.ParseException; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.common.header.Header; +import org.apache.kafka.common.header.Headers; + +import org.springframework.boot.autoconfigure.kafka.KafkaProperties; +import org.springframework.cloud.contract.verifier.messaging.MessageVerifier; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.test.EmbeddedKafkaBroker; +import org.springframework.kafka.test.utils.KafkaTestUtils; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHeaders; +import org.springframework.messaging.support.MessageBuilder; + +class KafkaStubMessages implements MessageVerifier> { + + private static final Log log = LogFactory.getLog(KafkaStubMessages.class); + + private final KafkaTemplate kafkaTemplate; + + private final Receiver receiver; + + KafkaStubMessages(KafkaTemplate kafkaTemplate, EmbeddedKafkaBroker broker, + KafkaProperties kafkaProperties, KafkaStubMessagesInitializer initializer) { + this.kafkaTemplate = kafkaTemplate; + Map topicToConsumer = initializer.initialize(broker, + kafkaProperties); + this.receiver = new Receiver(topicToConsumer); + } + + @Override + public void send(Message message, String destination) { + String defaultTopic = this.kafkaTemplate.getDefaultTopic(); + try { + this.kafkaTemplate.setDefaultTopic(destination); + if (log.isDebugEnabled()) { + log.debug("Will send a message [" + message + "] to destination [" + + destination + "]"); + } + this.kafkaTemplate.send(message).get(5, TimeUnit.SECONDS); + this.kafkaTemplate.flush(); + } + catch (Exception ex) { + throw new IllegalStateException(ex); + } + finally { + this.kafkaTemplate.setDefaultTopic(defaultTopic); + } + } + + @Override + public Message receive(String destination, long timeout, TimeUnit timeUnit) { + return this.receiver.receive(destination, timeout, timeUnit); + } + + @Override + public Message receive(String destination) { + return receive(destination, 5, TimeUnit.SECONDS); + } + + @Override + public void send(Object payload, Map headers, String destination) { + Message message = MessageBuilder.createMessage(payload, + new MessageHeaders(headers)); + send(message, destination); + } + +} + +class Receiver { + + private final Map consumers; + + private static final Log log = LogFactory.getLog(Receiver.class); + + Receiver(Map consumers) { + this.consumers = consumers; + } + + Message receive(String topic, long timeout, TimeUnit timeUnit) { + Consumer consumer = this.consumers.get(topic); + if (consumer == null) { + throw new IllegalStateException( + "No consumer set up for topic [" + topic + "]"); + } + ConsumerRecord record = KafkaTestUtils.getSingleRecord(consumer, + topic, timeUnit.toMillis(timeout)); + if (log.isDebugEnabled()) { + log.debug("Got a single record for destination [" + topic + "]"); + } + return new Record(record).toMessage(); + } + +} + +class Record { + + private final ConsumerRecord record; + + Record(ConsumerRecord record) { + this.record = record; + } + + private Map toMap(Headers headers) { + Map map = new HashMap<>(); + for (Header header : headers) { + map.put(header.key(), header.value()); + } + return map; + } + + Message toMessage() { + Object textPayload = record.value(); + // sometimes it's a message sometimes just payload + MessageHeaders headers = new MessageHeaders(toMap(record.headers())); + if (textPayload instanceof String && ((String) textPayload).contains("payload") + && ((String) textPayload).contains("headers")) { + try { + Object object = new JSONParser(JSONParser.DEFAULT_PERMISSIVE_MODE) + .parse((String) textPayload); + JSONObject jo = (JSONObject) object; + String payload = (String) jo.get("payload"); + JSONObject headersInJson = (JSONObject) jo.get("headers"); + Map newHeaders = new HashMap(headers); + newHeaders.putAll(headersInJson); + return MessageBuilder.createMessage(unquoted(payload), + new MessageHeaders(newHeaders)); + } + catch (ParseException ex) { + throw new IllegalStateException(ex); + } + } + return MessageBuilder.createMessage(unquoted(textPayload), headers); + } + + private Object unquoted(Object value) { + String textPayload = value instanceof byte[] ? new String((byte[]) value) + : value.toString(); + if (textPayload.startsWith("\"") && textPayload.endsWith("\"")) { + return textPayload.substring(1, textPayload.length() - 1).replace("\\\"", + "\""); + } + return textPayload; + } + +} diff --git a/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/kafka/KafkaStubMessagesInitializer.java b/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/kafka/KafkaStubMessagesInitializer.java new file mode 100644 index 0000000000..f50978dfd3 --- /dev/null +++ b/spring-cloud-contract-verifier/src/main/java/org/springframework/cloud/contract/verifier/messaging/kafka/KafkaStubMessagesInitializer.java @@ -0,0 +1,44 @@ +/* + * Copyright 2013-2019 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 + * + * https://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.contract.verifier.messaging.kafka; + +import java.util.Map; + +import org.apache.kafka.clients.consumer.Consumer; + +import org.springframework.boot.autoconfigure.kafka.KafkaProperties; +import org.springframework.kafka.test.EmbeddedKafkaBroker; + +/** + * Logic used to initialize {@link KafkaStubMessages}. This interface might have a + * different implementation for the producer side and for the consumer side. That's + * because you can't poll for a single message by different consumers. + * + * @author Marcin Grzejszczak + * @since 2.2.0 + */ +public interface KafkaStubMessagesInitializer { + + /** + * @param broker - embedded Kafka broker + * @param kafkaProperties - kafka properties + * @return topic to initialized consumer mapping + */ + Map initialize(EmbeddedKafkaBroker broker, + KafkaProperties kafkaProperties); + +} diff --git a/spring-cloud-contract-verifier/src/main/resources/META-INF/spring.factories b/spring-cloud-contract-verifier/src/main/resources/META-INF/spring.factories index 275d99c1ae..478246fbde 100644 --- a/spring-cloud-contract-verifier/src/main/resources/META-INF/spring.factories +++ b/spring-cloud-contract-verifier/src/main/resources/META-INF/spring.factories @@ -6,4 +6,5 @@ org.springframework.cloud.contract.verifier.messaging.amqp.ContractVerifierAmqpA org.springframework.cloud.contract.verifier.messaging.amqp.RabbitMockConnectionFactoryAutoConfiguration,\ org.springframework.cloud.contract.verifier.messaging.camel.ContractVerifierCamelConfiguration,\ org.springframework.cloud.contract.verifier.messaging.jms.ContractVerifierJmsConfiguration,\ +org.springframework.cloud.contract.verifier.messaging.kafka.ContractVerifierKafkaConfiguration,\ org.springframework.cloud.contract.verifier.messaging.noop.NoOpContractVerifierAutoConfiguration diff --git a/tests/pom.xml b/tests/pom.xml index b4442fc187..998cd5111f 100644 --- a/tests/pom.xml +++ b/tests/pom.xml @@ -58,6 +58,7 @@ spring-cloud-contract-stub-runner-stream spring-cloud-contract-stub-runner-amqp spring-cloud-contract-stub-runner-jms + spring-cloud-contract-stub-runner-kafka @@ -83,6 +84,7 @@ spring-cloud-contract-stub-runner-stream spring-cloud-contract-stub-runner-amqp spring-cloud-contract-stub-runner-jms + spring-cloud-contract-stub-runner-kafka diff --git a/tests/spring-cloud-contract-stub-runner-kafka/pom.xml b/tests/spring-cloud-contract-stub-runner-kafka/pom.xml new file mode 100644 index 0000000000..f5579cf799 --- /dev/null +++ b/tests/spring-cloud-contract-stub-runner-kafka/pom.xml @@ -0,0 +1,82 @@ + + + 4.0.0 + + org.springframework.cloud + spring-cloud-contract-tests + 2.2.0.BUILD-SNAPSHOT + .. + + spring-cloud-contract-stub-runner-kafka + jar + Spring Cloud Contract Stub Runner JMS + Spring Cloud Contract Stub Runner JMS + + + org.springframework.cloud + spring-cloud-contract-stub-runner + + + org.springframework.cloud + spring-cloud-starter-contract-stub-runner-jetty + test + + + org.springframework.kafka + spring-kafka + + + org.springframework.kafka + spring-kafka-test + test + + + junit + junit + test + + + org.codehaus.groovy + groovy + + + org.spockframework + spock-core + test + + + org.spockframework + spock-spring + test + + + org.springframework.boot + spring-boot-starter-test + test + + + org.springframework.boot + spring-boot-starter-web + test + + + info.solidsoft.spock + spock-global-unroll + test + + + + + + org.codehaus.gmavenplus + gmavenplus-plugin + + + org.apache.maven.plugins + maven-surefire-plugin + + + + diff --git a/tests/spring-cloud-contract-stub-runner-kafka/src/test/groovy/org/springframework/cloud/contract/stubrunner/messaging/kafka/BookReturned.groovy b/tests/spring-cloud-contract-stub-runner-kafka/src/test/groovy/org/springframework/cloud/contract/stubrunner/messaging/kafka/BookReturned.groovy new file mode 100644 index 0000000000..3e76443661 --- /dev/null +++ b/tests/spring-cloud-contract-stub-runner-kafka/src/test/groovy/org/springframework/cloud/contract/stubrunner/messaging/kafka/BookReturned.groovy @@ -0,0 +1,34 @@ +/* + * Copyright 2013-2019 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 + * + * https://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.contract.stubrunner.messaging.kafka + + +import groovy.transform.CompileStatic +import groovy.transform.EqualsAndHashCode + +@CompileStatic +@EqualsAndHashCode +class BookReturned implements Serializable { + String bookName + + BookReturned(String bookName) { + this.bookName = bookName + } + + BookReturned() { + } +} diff --git a/tests/spring-cloud-contract-stub-runner-kafka/src/test/groovy/org/springframework/cloud/contract/stubrunner/messaging/kafka/KafkaStubRunnerSpec.groovy b/tests/spring-cloud-contract-stub-runner-kafka/src/test/groovy/org/springframework/cloud/contract/stubrunner/messaging/kafka/KafkaStubRunnerSpec.groovy new file mode 100644 index 0000000000..34bdee670c --- /dev/null +++ b/tests/spring-cloud-contract-stub-runner-kafka/src/test/groovy/org/springframework/cloud/contract/stubrunner/messaging/kafka/KafkaStubRunnerSpec.groovy @@ -0,0 +1,311 @@ +/* + * Copyright 2013-2019 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 + * + * https://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.contract.stubrunner.messaging.kafka + +import java.util.concurrent.CountDownLatch +import java.util.concurrent.TimeUnit + +import groovy.json.JsonOutput +import groovy.json.JsonSlurper +import spock.lang.IgnoreIf +import spock.lang.Specification +import spock.util.concurrent.PollingConditions + +import org.springframework.beans.factory.annotation.Autowired +import org.springframework.beans.factory.annotation.Value +import org.springframework.boot.autoconfigure.EnableAutoConfiguration +import org.springframework.boot.test.context.SpringBootContextLoader +import org.springframework.boot.test.context.SpringBootTest +import org.springframework.cloud.contract.spec.Contract +import org.springframework.cloud.contract.stubrunner.StubFinder +import org.springframework.cloud.contract.stubrunner.spring.AutoConfigureStubRunner +import org.springframework.context.annotation.Bean +import org.springframework.context.annotation.ComponentScan +import org.springframework.context.annotation.Configuration +import org.springframework.kafka.annotation.EnableKafka +import org.springframework.kafka.annotation.KafkaListener +import org.springframework.kafka.core.KafkaTemplate +import org.springframework.kafka.support.DefaultKafkaHeaderMapper +import org.springframework.kafka.test.EmbeddedKafkaBroker +import org.springframework.kafka.test.context.EmbeddedKafka +import org.springframework.messaging.Message +import org.springframework.messaging.MessageHeaders +import org.springframework.messaging.support.MessageBuilder +import org.springframework.test.context.ContextConfiguration +/** + * @author Marcin Grzejszczak + */ +@ContextConfiguration(classes = Config, loader = SpringBootContextLoader) +@SpringBootTest(properties = ["debug=true"]) +@AutoConfigureStubRunner +@IgnoreIf({ os.windows }) +@EmbeddedKafka(topics = ["input", "output", "delete"]) +class KafkaStubRunnerSpec extends Specification { + + @Autowired + StubFinder stubFinder + @Autowired + KafkaTemplate kafkaTemplate + @Autowired + EmbeddedKafkaBroker broker + @Value('${spring.embedded.kafka.brokers}') + String brokers + @Autowired + MyMessageListener myMessageListener + PollingConditions await = new PollingConditions() + + def setup() { + this.myMessageListener.clear() + } + + private Message receiveFromOutput() { + this.myMessageListener.await() + return this.myMessageListener.output + } + + def 'should download the stub and register a route for it'() { + when: + // tag::client_send[] + Message message = MessageBuilder.createMessage(new BookReturned('foo'), new MessageHeaders([sample: "header",])) + kafkaTemplate.setDefaultTopic('input') + kafkaTemplate.send(message) + // end::client_send[] + then: + // tag::client_receive[] + Message receivedMessage = receiveFromOutput() + // end::client_receive[] + and: + await.eventually { + // tag::client_receive_message[] + assert receivedMessage != null + assert assertThatBodyContainsBookNameFoo(receivedMessage.getPayload()) + assert receivedMessage.getHeaders().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[] + then: + // tag::client_trigger_receive[] + Message receivedMessage = receiveFromOutput() + // end::client_trigger_receive[] + and: + await.eventually { + // tag::client_trigger_message[] + assert receivedMessage != null + assert assertThatBodyContainsBookNameFoo(receivedMessage.getPayload()) + assert receivedMessage.getHeaders().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('my:stubs', 'return_book_1') + // end::trigger_group_artifact[] + then: + Message receivedMessage = receiveFromOutput() + and: + await.eventually { + assert receivedMessage != null + assert assertThatBodyContainsBookNameFoo(receivedMessage.getPayload()) + assert receivedMessage.getHeaders().get('BOOK-NAME') == 'foo' + } + } + + def 'should trigger a label for the existing artifactId'() { + when: + // tag::trigger_artifact[] + stubFinder.trigger('stubs', 'return_book_1') + // end::trigger_artifact[] + then: + Message receivedMessage = receiveFromOutput() + and: + await.eventually { + assert receivedMessage != null + assert assertThatBodyContainsBookNameFoo(receivedMessage.getPayload()) + assert receivedMessage.getHeaders().get('BOOK-NAME') == 'foo' + } + } + + def 'should throw an exception when missing label is passed'() { + when: + stubFinder.trigger('missing label') + then: + thrown(IllegalArgumentException) + } + + def 'should throw an exception when missing label and artifactid is passed'() { + when: + stubFinder.trigger('some:service', 'return_book_1') + then: + thrown(IllegalArgumentException) + } + + def 'should trigger messages by running all triggers'() { + when: + // tag::trigger_all[] + stubFinder.trigger() + // end::trigger_all[] + then: + Message receivedMessage = receiveFromOutput() + and: + await.eventually { + assert receivedMessage != null + assert assertThatBodyContainsBookNameFoo(receivedMessage.getPayload()) + assert receivedMessage.getHeaders().get('BOOK-NAME') == 'foo' + } + } + + def 'should trigger a label with no output message'() { + when: + // tag::trigger_no_output[] + Message message = MessageBuilder.createMessage(new BookReturned('foo'), new MessageHeaders([sample: "header",])) + kafkaTemplate.setDefaultTopic('delete') + kafkaTemplate.send(message) + // end::trigger_no_output[] + then: + noExceptionThrown() + } + + def 'should not trigger a message that does not match input'() { + when: + Message message = MessageBuilder.createMessage(new BookReturned('notmatching'), new MessageHeaders([wrong: "header",])) + kafkaTemplate.setDefaultTopic('input') + kafkaTemplate.send(message) + then: + Message receivedMessage = receiveFromOutput() + and: + receivedMessage == null + } + + private boolean assertThatBodyContainsBookNameFoo(Object payload) { + println "Got payload [" + payload + "]" + String objectAsString = payload instanceof String ? payload : + JsonOutput.toJson(payload) + def json = new JsonSlurper().parseText(objectAsString) + if (objectAsString.contains("payload")) { + return json.payload.bookName == 'foo' + } + return json.bookName == 'foo' + } + + @Configuration + @ComponentScan + @EnableAutoConfiguration + @EnableKafka + static class Config { + + @Bean + DefaultKafkaHeaderMapper headerMapper() { + return new DefaultKafkaHeaderMapper(); + } + + @Bean + MyMessageListener myMessageListener() { + return new MyMessageListener(); + } + } + + static class MyMessageListener { + + CountDownLatch latch = new CountDownLatch(1) + + Message output + + @KafkaListener(topics = ["output"]) + void output(Message message) { + println "I got the message [${message}]" + this.output = message + this.latch.countDown() + } + + void clear() { + this.output = null + this.latch = new CountDownLatch(1) + } + + void await() { + this.latch.await(5, TimeUnit.SECONDS) + } + } + + Contract dsl = + // tag::sample_dsl[] + Contract.make { + label 'return_book_1' + input { + triggeredBy('bookReturnedTriggered()') + } + outputMessage { + sentTo('output') + 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('input') + messageBody([ + bookName: 'foo' + ]) + messageHeaders { + header('sample', 'header') + } + } + outputMessage { + sentTo('output') + 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()') + } + } + // end::sample_dsl_3[] +} \ No newline at end of file diff --git a/tests/spring-cloud-contract-stub-runner-kafka/src/test/resources/application.yml b/tests/spring-cloud-contract-stub-runner-kafka/src/test/resources/application.yml new file mode 100644 index 0000000000..46bd19ecaa --- /dev/null +++ b/tests/spring-cloud-contract-stub-runner-kafka/src/test/resources/application.yml @@ -0,0 +1,21 @@ +stubrunner: + repository-root: stubs:classpath:/stubs/ + ids: my:stubs + stubs-mode: remote +spring: + kafka: + bootstrap-servers: ${spring.embedded.kafka.brokers} + producer: + properties: + "value.serializer": "org.springframework.kafka.support.serializer.JsonSerializer" + "spring.json.trusted.packages": "*" + consumer: + properties: + "value.deserializer": "org.springframework.kafka.support.serializer.JsonDeserializer" + "value.serializer": "org.springframework.kafka.support.serializer.JsonSerializer" + "spring.json.trusted.packages": "*" + group-id: groupId +server: + port: 0 +debug: true +logging.level.org.springframework.cloud.contract: debug \ No newline at end of file diff --git a/tests/spring-cloud-contract-stub-runner-kafka/src/test/resources/stubs/bookDeleted.groovy b/tests/spring-cloud-contract-stub-runner-kafka/src/test/resources/stubs/bookDeleted.groovy new file mode 100644 index 0000000000..b013bba223 --- /dev/null +++ b/tests/spring-cloud-contract-stub-runner-kafka/src/test/resources/stubs/bookDeleted.groovy @@ -0,0 +1,13 @@ +org.springframework.cloud.contract.spec.Contract.make { + label 'delete_book' + input { + messageFrom('delete') + messageBody([ + bookName: 'foo' + ]) + messageHeaders { + header('sample', 'header') + } + assertThat('bookWasDeleted()') + } +} \ No newline at end of file diff --git a/tests/spring-cloud-contract-stub-runner-kafka/src/test/resources/stubs/bookReturned1.groovy b/tests/spring-cloud-contract-stub-runner-kafka/src/test/resources/stubs/bookReturned1.groovy new file mode 100644 index 0000000000..0094bc7b1d --- /dev/null +++ b/tests/spring-cloud-contract-stub-runner-kafka/src/test/resources/stubs/bookReturned1.groovy @@ -0,0 +1,13 @@ +org.springframework.cloud.contract.spec.Contract.make { + label 'return_book_1' + input { + triggeredBy('bookReturnedTriggered()') + } + outputMessage { + sentTo('output') + body('''{ "bookName" : "foo" }''') + headers { + header('BOOK-NAME', 'foo') + } + } +} \ No newline at end of file diff --git a/tests/spring-cloud-contract-stub-runner-kafka/src/test/resources/stubs/bookReturned2.groovy b/tests/spring-cloud-contract-stub-runner-kafka/src/test/resources/stubs/bookReturned2.groovy new file mode 100644 index 0000000000..e0d53098d5 --- /dev/null +++ b/tests/spring-cloud-contract-stub-runner-kafka/src/test/resources/stubs/bookReturned2.groovy @@ -0,0 +1,21 @@ +org.springframework.cloud.contract.spec.Contract.make { + label 'return_book_2' + input { + messageFrom('input') + messageBody([ + bookName: 'foo' + ]) + messageHeaders { + header('sample', 'header') + } + } + outputMessage { + sentTo('output') + body([ + bookName: 'foo' + ]) + headers { + header('BOOK-NAME', 'foo') + } + } +} \ No newline at end of file