Messaging polyglot support (#1472)
added support for AMQP, KAFKA and standalone options
This commit is contained in:
Marcin Grzejszczak
committed by
GitHub
GitHub
parent
c5d3456d3a
commit
a915cf102b
@@ -0,0 +1,186 @@
|
||||
/*
|
||||
* Copyright 2013-2020 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 contracts;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import io.restassured.RestAssured;
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.TestInfo;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
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.SpringBootTest;
|
||||
import org.springframework.boot.web.client.RestTemplateBuilder;
|
||||
import org.springframework.cloud.contract.verifier.converter.YamlContract;
|
||||
import org.springframework.cloud.contract.verifier.messaging.MessageVerifier;
|
||||
import org.springframework.cloud.contract.verifier.messaging.amqp.AmqpMetadata;
|
||||
import org.springframework.cloud.contract.verifier.messaging.boot.AutoConfigureMessageVerifier;
|
||||
import org.springframework.cloud.contract.verifier.messaging.camel.StandaloneMetadata;
|
||||
import org.springframework.cloud.contract.verifier.messaging.internal.ContractVerifierMessageMetadata;
|
||||
import org.springframework.cloud.contract.verifier.util.ContractVerifierUtil;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.context.annotation.Import;
|
||||
import org.springframework.util.StringUtils;
|
||||
import org.springframework.web.client.RestTemplate;
|
||||
|
||||
/**
|
||||
* @author Marcin Grzejszczak
|
||||
*/
|
||||
@SpringBootTest(classes = ContractTestsBase.Config.class, webEnvironment = SpringBootTest.WebEnvironment.NONE)
|
||||
@AutoConfigureMessageVerifier
|
||||
public abstract class ContractTestsBase {
|
||||
|
||||
private static final Logger log = LoggerFactory.getLogger(ContractTestsBase.class);
|
||||
|
||||
/**
|
||||
* URL at which the application is running.
|
||||
*/
|
||||
@Value("${APPLICATION_BASE_URL}")
|
||||
String url;
|
||||
|
||||
/**
|
||||
* Optional username to access the application.
|
||||
*/
|
||||
@Value("${APPLICATION_USERNAME:}")
|
||||
String username;
|
||||
|
||||
/**
|
||||
* Optional password to access the application.
|
||||
*/
|
||||
@Value("${APPLICATION_PASSWORD:}")
|
||||
String password;
|
||||
|
||||
/**
|
||||
* Timeout to connect to the application to trigger a message.
|
||||
*/
|
||||
@Value("${MESSAGING_TRIGGER_CONNECT_TIMEOUT:5000}")
|
||||
Integer connectTimeout;
|
||||
|
||||
/**
|
||||
* Timeout to read the response from the application to trigger a message.
|
||||
*/
|
||||
@Value("${MESSAGING_TRIGGER_READ_TIMEOUT:5000}")
|
||||
Integer readTimeout;
|
||||
|
||||
/**
|
||||
* Defines the messaging type when dealing with message based contracts.
|
||||
*/
|
||||
@Value("${MESSAGING_TYPE:}")
|
||||
String messagingType;
|
||||
|
||||
@Autowired
|
||||
MessageVerifier messageVerifier;
|
||||
|
||||
@BeforeEach
|
||||
public void setup(TestInfo testInfo) {
|
||||
RestAssured.baseURI = this.url;
|
||||
if (StringUtils.hasText(this.username)) {
|
||||
RestAssured.authentication = RestAssured.basic(this.username, this.password);
|
||||
}
|
||||
setupMessagingFromContract(testInfo);
|
||||
}
|
||||
|
||||
private void setupMessagingFromContract(TestInfo testInfo) {
|
||||
try {
|
||||
YamlContract contract = ContractVerifierUtil.contract(this, testInfo.getDisplayName());
|
||||
setupMessagingIfPresent(contract);
|
||||
} catch (Exception e) {
|
||||
log.warn("An exception occurred while trying to setup messaging from contract", e);
|
||||
}
|
||||
}
|
||||
|
||||
private void setupMessagingIfPresent(YamlContract contract) {
|
||||
if (contract.input == null && contract.outputMessage == null) {
|
||||
return;
|
||||
}
|
||||
setupAmqpIfPresent(contract);
|
||||
setupStandaloneIfPresent(contract);
|
||||
}
|
||||
|
||||
private void setupAmqpIfPresent(YamlContract contract) {
|
||||
AmqpMetadata amqpMetadata = AmqpMetadata.fromMetadata(contract.metadata);
|
||||
if (isMessagingType("rabbit") && hasDeclaredOutputQueue(amqpMetadata) || isMessagingType("kafka")) {
|
||||
log.info("First will try to receive a message to setup the connection with the broker");
|
||||
if (contract.input != null && StringUtils.hasText(contract.input.messageFrom)) {
|
||||
setupConnection(contract.input.messageFrom, contract);
|
||||
}
|
||||
if (contract.outputMessage != null && StringUtils.hasText(contract.outputMessage.sentTo)){
|
||||
setupConnection(contract.outputMessage.sentTo, contract);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private void setupStandaloneIfPresent(YamlContract contract) {
|
||||
StandaloneMetadata metadata = StandaloneMetadata.fromMetadata(contract.metadata);
|
||||
if (StringUtils.hasText(metadata.getSetup().getOptions())) {
|
||||
log.info("First will try to receive a message to setup the connection with the broker");
|
||||
setMessageType(contract, ContractVerifierMessageMetadata.MessageType.SETUP);
|
||||
setupConnection(metadata.getSetup().getOptions(), contract);
|
||||
}
|
||||
}
|
||||
|
||||
private void setMessageType(YamlContract contract,
|
||||
ContractVerifierMessageMetadata.MessageType output) {
|
||||
contract.metadata.put(ContractVerifierMessageMetadata.METADATA_KEY,
|
||||
new ContractVerifierMessageMetadata(output));
|
||||
}
|
||||
|
||||
private void setupConnection(String destination, YamlContract contract) {
|
||||
if (StringUtils.isEmpty(destination)) {
|
||||
return;
|
||||
}
|
||||
log.info("Setting up destination [{}]", destination);
|
||||
this.messageVerifier.receive(destination, 100, TimeUnit.MILLISECONDS, contract);
|
||||
}
|
||||
|
||||
private boolean hasDeclaredOutputQueue(AmqpMetadata amqpMetadata) {
|
||||
return StringUtils.hasText(amqpMetadata.getOutputMessage().getConnectToBroker().getDeclareQueueWithName());
|
||||
}
|
||||
|
||||
private boolean isMessagingType(String rabbit) {
|
||||
return rabbit.equalsIgnoreCase(this.messagingType);
|
||||
}
|
||||
|
||||
public void triggerMessage(String label) {
|
||||
String url = this.url + "/springcloudcontract/" + label;
|
||||
log.info("Will send a request to [{}] in order to trigger a message", url);
|
||||
restTemplate().postForObject(url, "", String.class);
|
||||
}
|
||||
|
||||
private RestTemplate restTemplate() {
|
||||
RestTemplateBuilder builder = new RestTemplateBuilder()
|
||||
.setConnectTimeout(Duration.ofMillis(this.connectTimeout))
|
||||
.setReadTimeout(Duration.ofMillis(this.readTimeout));
|
||||
if (StringUtils.hasText(this.username)) {
|
||||
builder = builder.basicAuthentication(this.username, this.password);
|
||||
}
|
||||
return builder.build();
|
||||
}
|
||||
|
||||
@Configuration
|
||||
@Import(MessagingAutoConfig.class)
|
||||
@EnableAutoConfiguration
|
||||
protected static class Config {
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,200 @@
|
||||
/*
|
||||
* Copyright 2013-2020 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 contracts;
|
||||
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import org.apache.camel.ConsumerTemplate;
|
||||
import org.apache.camel.Exchange;
|
||||
import org.apache.camel.Message;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import org.springframework.amqp.core.MessageProperties;
|
||||
import org.springframework.beans.factory.annotation.Value;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
|
||||
import org.springframework.cloud.contract.verifier.converter.YamlContract;
|
||||
import org.springframework.cloud.contract.verifier.messaging.MessageVerifier;
|
||||
import org.springframework.cloud.contract.verifier.messaging.amqp.AmqpMetadata;
|
||||
import org.springframework.cloud.contract.verifier.messaging.internal.ContractVerifierMessage;
|
||||
import org.springframework.cloud.contract.verifier.messaging.internal.ContractVerifierMessageMetadata;
|
||||
import org.springframework.cloud.contract.verifier.messaging.internal.ContractVerifierMessaging;
|
||||
import org.springframework.cloud.contract.verifier.messaging.kafka.KafkaMetadata;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.context.annotation.Profile;
|
||||
import org.springframework.util.StringUtils;
|
||||
|
||||
/**
|
||||
* @author Marcin Grzejszczak
|
||||
*/
|
||||
@Configuration
|
||||
@ConditionalOnProperty("MESSAGING_TYPE")
|
||||
@Profile("messagingtype")
|
||||
public class MessagingAutoConfig {
|
||||
|
||||
/**
|
||||
* Type of messaging. Can be either [rabbit] or [kafka].
|
||||
*/
|
||||
@Value("${MESSAGING_TYPE:}")
|
||||
String messagingType;
|
||||
|
||||
/**
|
||||
* For RabbitMQ - brokers addresses.
|
||||
*/
|
||||
@Value("${SPRING_RABBITMQ_ADDRESSES:}")
|
||||
String springRabbitmqAddresses;
|
||||
|
||||
/**
|
||||
* For Kafka - brokers addresses.
|
||||
*/
|
||||
@Value("${SPRING_KAFKA_BOOTSTRAP_SERVERS:}")
|
||||
String springKafkaBootstrapServers;
|
||||
|
||||
@Bean
|
||||
public ContractVerifierMessaging<Message> contractVerifierMessaging(
|
||||
MessageVerifier<Message> exchange) {
|
||||
return new ContractVerifierCamelHelper(exchange);
|
||||
}
|
||||
|
||||
@Bean
|
||||
MessageVerifier<Message> manualMessageVerifier(ConsumerTemplate consumerTemplate) {
|
||||
return new MessageVerifier<Message>() {
|
||||
|
||||
private final Logger log = LoggerFactory.getLogger(MessageVerifier.class);
|
||||
|
||||
@Override
|
||||
public Message receive(String destination, long timeout, TimeUnit timeUnit, YamlContract yamlContract) {
|
||||
String uri = messagingType() + "://" + destination + additionalOptions(yamlContract);
|
||||
log.info("Camel URI [{}]", uri);
|
||||
Exchange exchange = consumerTemplate.receive(uri, timeUnit.toMillis(timeout));
|
||||
if (exchange == null) {
|
||||
return null;
|
||||
}
|
||||
return exchange.getMessage();
|
||||
}
|
||||
|
||||
private String messagingType() {
|
||||
if (messagingType.equalsIgnoreCase("kafka")) {
|
||||
return "kafka";
|
||||
}
|
||||
return "rabbitmq";
|
||||
}
|
||||
|
||||
private String additionalOptions(YamlContract contract) {
|
||||
if (contract == null) {
|
||||
return "";
|
||||
}
|
||||
if (messagingType.equalsIgnoreCase("kafka")) {
|
||||
return setKafkaOpts(contract);
|
||||
}
|
||||
return setRabbitOpts(contract);
|
||||
}
|
||||
|
||||
private String setKafkaOpts(YamlContract contract) {
|
||||
String opts = defaultOpts(contract);
|
||||
KafkaMetadata metadata = KafkaMetadata.fromMetadata(contract.metadata);
|
||||
ContractVerifierMessageMetadata messageMetadata = ContractVerifierMessageMetadata.fromMetadata(contract.metadata);
|
||||
if (inputMessage(messageMetadata) && StringUtils.hasText(metadata.getInput().getConnectToBroker().getAdditionalOptions())) {
|
||||
return opts + "&" + metadata.getInput().getConnectToBroker().getAdditionalOptions();
|
||||
}
|
||||
else if (StringUtils.hasText(metadata.getOutputMessage().getConnectToBroker().getAdditionalOptions())) {
|
||||
return opts + "&" + metadata.getOutputMessage().getConnectToBroker().getAdditionalOptions();
|
||||
}
|
||||
return opts;
|
||||
}
|
||||
|
||||
private String defaultOpts(YamlContract contract) {
|
||||
String consumerGroup = sameConsumerGroupForSameContract(contract);
|
||||
return "?brokers=" + getRequiredProperty("SPRING_KAFKA_BOOTSTRAP_SERVERS", springKafkaBootstrapServers) + "&autoOffsetReset=latest&groupId=" + consumerGroup + "&shutdownTimeout=5";
|
||||
}
|
||||
|
||||
private String sameConsumerGroupForSameContract(YamlContract contract) {
|
||||
return contract.input.hashCode() + "_" + contract.outputMessage.hashCode();
|
||||
}
|
||||
|
||||
private String setRabbitOpts(YamlContract contract) {
|
||||
String opts = "?addresses=" + getRequiredProperty("SPRING_RABBITMQ_ADDRESSES", springRabbitmqAddresses);
|
||||
AmqpMetadata metadata = AmqpMetadata.fromMetadata(contract.metadata);
|
||||
ContractVerifierMessageMetadata messageMetadata = ContractVerifierMessageMetadata.fromMetadata(contract.metadata);
|
||||
if (inputMessage(messageMetadata) && StringUtils.hasText(metadata.getInput().getConnectToBroker().getAdditionalOptions())) {
|
||||
return opts + "&" + metadata.getInput().getConnectToBroker().getAdditionalOptions();
|
||||
}
|
||||
else if (StringUtils.hasText(metadata.getOutputMessage().getConnectToBroker().getAdditionalOptions())) {
|
||||
return opts + "&" + metadata.getOutputMessage().getConnectToBroker().getAdditionalOptions();
|
||||
}
|
||||
return defaultOpts(opts, metadata, messageMetadata);
|
||||
}
|
||||
|
||||
private String getRequiredProperty(String name, String value) {
|
||||
if (StringUtils.isEmpty(value)) {
|
||||
throw new IllegalStateException("The property [" + name + "] must not be empty!");
|
||||
}
|
||||
return value;
|
||||
}
|
||||
|
||||
private boolean inputMessage(ContractVerifierMessageMetadata messageMetadata) {
|
||||
return messageMetadata.getMessageType() == ContractVerifierMessageMetadata.MessageType.INPUT;
|
||||
}
|
||||
|
||||
private String defaultOpts(String opts, AmqpMetadata amqpMetadata, ContractVerifierMessageMetadata messageMetadata) {
|
||||
AmqpMetadata.ConnectToBroker connectToBroker = inputMessage(messageMetadata) ? amqpMetadata.getInput().getConnectToBroker() : amqpMetadata.getOutputMessage().getConnectToBroker();
|
||||
MessageProperties messageProperties = inputMessage(messageMetadata) ? amqpMetadata.getInput().getMessageProperties() : amqpMetadata.getOutputMessage().getMessageProperties();
|
||||
if (StringUtils.hasText(connectToBroker.getDeclareQueueWithName())) {
|
||||
opts = opts + "&queue=" + connectToBroker.getDeclareQueueWithName();
|
||||
}
|
||||
if (messageProperties != null && StringUtils.hasText(messageProperties.getReceivedRoutingKey())) {
|
||||
opts = opts + "&routingKey=" + messageProperties.getReceivedRoutingKey();
|
||||
}
|
||||
return opts;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Message receive(String destination, YamlContract yamlContract) {
|
||||
return receive(destination, 5, TimeUnit.SECONDS, yamlContract);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void send(Message message, String destination, YamlContract yamlContract) {
|
||||
throw new UnsupportedOperationException("Currently supports only receiving");
|
||||
}
|
||||
|
||||
@Override
|
||||
public void send(Object payload, Map headers, String destination, YamlContract yamlContract) {
|
||||
throw new UnsupportedOperationException("Currently supports only receiving");
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
class ContractVerifierCamelHelper extends ContractVerifierMessaging<Message> {
|
||||
|
||||
ContractVerifierCamelHelper(MessageVerifier<Message> exchange) {
|
||||
super(exchange);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected ContractVerifierMessage convert(Message receive) {
|
||||
if (receive == null) {
|
||||
return null;
|
||||
}
|
||||
return new ContractVerifierMessage(receive.getBody(), receive.getHeaders());
|
||||
}
|
||||
|
||||
}
|
||||
@@ -1,57 +0,0 @@
|
||||
/*
|
||||
* Copyright 2013-2020 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 contracts;
|
||||
|
||||
import io.restassured.RestAssured;
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Value;
|
||||
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
|
||||
import org.springframework.boot.test.context.SpringBootTest;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.util.StringUtils;
|
||||
|
||||
/**
|
||||
* @author Marcin Grzejszczak
|
||||
*/
|
||||
@SpringBootTest(classes = RestBase.Config.class, webEnvironment = SpringBootTest.WebEnvironment.NONE)
|
||||
public abstract class RestBase {
|
||||
|
||||
@Value("${APPLICATION_BASE_URL}")
|
||||
String url;
|
||||
|
||||
@Value("${APPLICATION_USERNAME:}")
|
||||
String username;
|
||||
|
||||
@Value("${APPLICATION_PASSWORD:}")
|
||||
String password;
|
||||
|
||||
@BeforeEach
|
||||
public void setup() {
|
||||
RestAssured.baseURI = this.url;
|
||||
if (StringUtils.hasText(this.username)) {
|
||||
RestAssured.authentication = RestAssured.basic(this.username, this.password);
|
||||
}
|
||||
}
|
||||
|
||||
@Configuration
|
||||
@EnableAutoConfiguration
|
||||
protected static class Config {
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user