SpringCloud/spring-cloud-bus
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Refactors bus to use spring cloud function. (#236)

Browse Source 
Rather than the annotation model. The monolithic `BusAutoConfiguration` was split.

BusBridge is now an interface with a send() method. In the future, there will be an RSocketBusBridge, the default is `StreamBusBridge`.

StreamBusBridge uses StreamBridge to send messages.

RemoteApplicationEventListener listens for remote events sends them to bus if needed.

Adds an amqp integration test.

fixes gh-227
This commit is contained in:
Spencer Gibb
2020-09-25 12:12:16 -04:00
committed by GitHub
parent aa817ea36c
commit 17c16bc345
16 changed files with 608 additions and 323 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
.circleci/config.yml
View File

@@ -1,24 +1,24 @@
version: 2
jobs:
build:
docker:
- image: springcloud/pipeline-base
user: appuser
machine:
image: ubuntu-1604:202007-01
environment:
_JAVA_OPTIONS: "-Xms1024m -Xmx2048m"
TERM: dumb
CACHE_VERSION: 2
branches:
ignore:
- gh-pages # list of branches to ignore
steps:
- checkout
- restore_cache:
key: sc-bus-{{ .Branch }}
key: sc-bus-{{ .Environment.CACHE_VERSION }}-{{ .Branch }}
- run:
name: "Download dependencies"
command: ./mvnw -s .settings.xml -U --fail-never dependency:go-offline || true
- save_cache:
key: sc-bus-{{ .Branch }}
key: sc-bus-{{ .Environment.CACHE_VERSION }}-{{ .Branch }}
paths:
- ~/.m2
- run:

24
spring-cloud-bus-tests/pom.xml
View File

@@ -17,12 +17,21 @@
<relativePath>..</relativePath> <!-- lookup parent from repository -->
</parent>
<properties>
<testcontainers.version>1.15.0-rc1</testcontainers.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-webflux</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-actuator</artifactId>
@@ -30,7 +39,7 @@
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-bus</artifactId>
<artifactId>spring-cloud-starter-bus-amqp</artifactId>
<scope>test</scope>
</dependency>
<dependency>
@@ -45,12 +54,19 @@
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-stream-rabbit</artifactId>
<artifactId>spring-cloud-stream-test-support</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-test-support</artifactId>
<groupId>org.testcontainers</groupId>
<artifactId>junit-jupiter</artifactId>
<version>${testcontainers.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>rabbitmq</artifactId>
<version>${testcontainers.version}</version>
<scope>test</scope>
</dependency>
</dependencies>

105
spring-cloud-bus-tests/src/test/java/org/springframework/cloud/bus/BusAmqpIntegrationTests.java Normal file
View File

@@ -0,0 +1,105 @@
/*
* Copyright 2015-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 org.springframework.cloud.bus;
import java.util.HashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.testcontainers.containers.RabbitMQContainer;
import org.testcontainers.junit.jupiter.Container;
import org.testcontainers.junit.jupiter.Testcontainers;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringBootConfiguration;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.cloud.bus.event.EnvironmentChangeRemoteApplicationEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.test.context.DynamicPropertyRegistry;
import org.springframework.test.context.DynamicPropertySource;
import org.springframework.test.web.reactive.server.WebTestClient;
import static org.assertj.core.api.Assertions.assertThat;
import static org.springframework.boot.test.context.SpringBootTest.WebEnvironment.RANDOM_PORT;
@SpringBootTest(webEnvironment = RANDOM_PORT, properties = { "management.endpoints.web.exposure.include=*",
"logging.level.org.springframework.cloud.bus=TRACE", "spring.cloud.bus.id=app:1",
"spring.autoconfigure.exclude=org.springframework.cloud.stream.test.binder.TestSupportBinderAutoConfiguration" })
@Testcontainers
public class BusAmqpIntegrationTests {
@Container
private static final RabbitMQContainer rabbitMQContainer = new RabbitMQContainer();
private static ConfigurableApplicationContext context;
@DynamicPropertySource
static void properties(DynamicPropertyRegistry registry) {
registry.add("spring.rabbitmq.host", rabbitMQContainer::getHost);
registry.add("spring.rabbitmq.port", rabbitMQContainer::getAmqpPort);
}
@BeforeAll
static void before() {
context = new SpringApplicationBuilder(TestConfig.class).properties("server.port=0",
"spring.rabbitmq.host=" + rabbitMQContainer.getHost(),
"spring.rabbitmq.port=" + rabbitMQContainer.getAmqpPort(),
"management.endpoints.web.exposure.include=*", "spring.cloud.bus.id=app:2",
"spring.autoconfigure.exclude=org.springframework.cloud.stream.test.binder.TestSupportBinderAutoConfiguration")
.run();
}
@AfterAll
static void after() {
if (context != null) {
context.close();
}
}
@Test
void remoteEventsAreSentViaAmqp(@Autowired WebTestClient client, @Autowired TestConfig testConfig)
throws InterruptedException {
assertThat(rabbitMQContainer.isRunning());
HashMap<String, String> map = new HashMap<>();
map.put("name", "foo");
map.put("value", "bar");
client.post().uri("/actuator/busenv").bodyValue(map).exchange().expectStatus().is2xxSuccessful();
TestConfig remoteTestConfig = context.getBean(TestConfig.class);
assertThat(remoteTestConfig.latch.await(5, TimeUnit.SECONDS)).isTrue();
assertThat(testConfig.latch.await(5, TimeUnit.SECONDS)).isTrue();
}
@SpringBootConfiguration
@EnableAutoConfiguration
static class TestConfig implements ApplicationListener<EnvironmentChangeRemoteApplicationEvent> {
CountDownLatch latch = new CountDownLatch(1);
@Override
public void onApplicationEvent(EnvironmentChangeRemoteApplicationEvent event) {
latch.countDown();
}
}
}

138
spring-cloud-bus/src/main/java/org/springframework/cloud/bus/BusAutoConfiguration.java
View File

@@ -16,12 +16,6 @@
package org.springframework.cloud.bus;
import javax.annotation.PostConstruct;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.actuate.autoconfigure.endpoint.condition.ConditionalOnAvailableEndpoint;
import org.springframework.boot.actuate.endpoint.annotation.Endpoint;
import org.springframework.boot.actuate.trace.http.HttpTraceRepository;
@@ -34,26 +28,17 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.cloud.autoconfigure.LifecycleMvcEndpointAutoConfiguration;
import org.springframework.cloud.bus.endpoint.EnvironmentBusEndpoint;
import org.springframework.cloud.bus.event.AckRemoteApplicationEvent;
import org.springframework.cloud.bus.event.EnvironmentChangeListener;
import org.springframework.cloud.bus.event.RemoteApplicationEvent;
import org.springframework.cloud.bus.event.SentApplicationEvent;
import org.springframework.cloud.bus.event.TraceListener;
import org.springframework.cloud.context.environment.EnvironmentManager;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.config.BindingProperties;
import org.springframework.cloud.stream.config.BindingServiceConfiguration;
import org.springframework.cloud.stream.config.BindingServiceProperties;
import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.ApplicationEventPublisherAware;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.event.EventListener;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.MessageBuilder;
import static org.springframework.cloud.bus.BusConstants.BUS_CONSUMER;
/**
* @author Spencer Gibb
@@ -61,122 +46,37 @@ import org.springframework.messaging.support.MessageBuilder;
*/
@Configuration(proxyBeanMethods = false)
@ConditionalOnBusEnabled
@EnableBinding(SpringCloudBusClient.class)
@EnableConfigurationProperties(BusProperties.class)
@AutoConfigureBefore(BindingServiceConfiguration.class)
// so stream bindings work properly
@AutoConfigureAfter({ LifecycleMvcEndpointAutoConfiguration.class, ServiceMatcherAutoConfiguration.class })
// so actuator endpoints have needed dependencies
public class BusAutoConfiguration implements ApplicationEventPublisherAware {
public class BusAutoConfiguration {
private static final Log log = LogFactory.getLog(BusAutoConfiguration.class);
/**
* Name of the Bus path matcher.
*/
public static final String BUS_PATH_MATCHER_NAME = "busPathMatcher";
/**
* Name of the Spring Cloud Config property.
*/
public static final String CLOUD_CONFIG_NAME_PROPERTY = "spring.cloud.config.name";
private final ServiceMatcher serviceMatcher;
private final BindingServiceProperties bindings;
private final BusProperties bus;
private MessageChannel cloudBusOutboundChannel;
private ApplicationEventPublisher applicationEventPublisher;
public BusAutoConfiguration(ServiceMatcher serviceMatcher, BindingServiceProperties bindings, BusProperties bus) {
this.serviceMatcher = serviceMatcher;
this.bindings = bindings;
this.bus = bus;
@Bean
@ConditionalOnMissingBean(BusBridge.class)
public StreamBusBridge streamBusBridge(StreamBridge streamBridge, BusProperties properties) {
return new StreamBusBridge(streamBridge, properties);
}
@PostConstruct
public void init() {
BindingProperties inputBinding = this.bindings.getBindings().get(SpringCloudBusClient.INPUT);
if (inputBinding == null) {
this.bindings.getBindings().put(SpringCloudBusClient.INPUT, new BindingProperties());
}
BindingProperties input = this.bindings.getBindings().get(SpringCloudBusClient.INPUT);
if (input.getDestination() == null || input.getDestination().equals(SpringCloudBusClient.INPUT)) {
input.setDestination(this.bus.getDestination());
}
BindingProperties outputBinding = this.bindings.getBindings().get(SpringCloudBusClient.OUTPUT);
if (outputBinding == null) {
this.bindings.getBindings().put(SpringCloudBusClient.OUTPUT, new BindingProperties());
}
BindingProperties output = this.bindings.getBindings().get(SpringCloudBusClient.OUTPUT);
if (output.getDestination() == null || output.getDestination().equals(SpringCloudBusClient.OUTPUT)) {
output.setDestination(this.bus.getDestination());
}
@Bean
@ConditionalOnMissingBean
public RemoteApplicationEventListener busRemoteApplicationEventListener(ServiceMatcher serviceMatcher,
BusBridge busBridge) {
return new RemoteApplicationEventListener(serviceMatcher, busBridge);
}
@Override
public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
this.applicationEventPublisher = applicationEventPublisher;
}
@Autowired
@Output(SpringCloudBusClient.OUTPUT)
public void setCloudBusOutboundChannel(MessageChannel cloudBusOutboundChannel) {
this.cloudBusOutboundChannel = cloudBusOutboundChannel;
}
@EventListener(classes = RemoteApplicationEvent.class)
public void acceptLocal(RemoteApplicationEvent event) {
if (this.serviceMatcher.isFromSelf(event) && !(event instanceof AckRemoteApplicationEvent)) {
if (log.isDebugEnabled()) {
log.debug("Sending remote event on bus: " + event);
}
this.cloudBusOutboundChannel.send(MessageBuilder.withPayload(event).build());
}
}
@StreamListener(SpringCloudBusClient.INPUT)
public void acceptRemote(RemoteApplicationEvent event) {
if (event instanceof AckRemoteApplicationEvent) {
if (this.bus.getTrace().isEnabled() && !this.serviceMatcher.isFromSelf(event)
&& this.applicationEventPublisher != null) {
this.applicationEventPublisher.publishEvent(event);
}
// If it's an ACK we are finished processing at this point
return;
}
if (log.isDebugEnabled()) {
log.debug("Received remote event from bus: " + event);
}
if (this.serviceMatcher.isForSelf(event) && this.applicationEventPublisher != null) {
if (!this.serviceMatcher.isFromSelf(event)) {
this.applicationEventPublisher.publishEvent(event);
}
if (this.bus.getAck().isEnabled()) {
AckRemoteApplicationEvent ack = new AckRemoteApplicationEvent(this, this.serviceMatcher.getServiceId(),
this.bus.getAck().getDestinationService(), event.getDestinationService(), event.getId(),
event.getClass());
this.cloudBusOutboundChannel.send(MessageBuilder.withPayload(ack).build());
this.applicationEventPublisher.publishEvent(ack);
}
}
if (this.bus.getTrace().isEnabled() && this.applicationEventPublisher != null) {
// We are set to register sent events so publish it for local consumption,
// irrespective of the origin
this.applicationEventPublisher.publishEvent(new SentApplicationEvent(this, event.getOriginService(),
event.getDestinationService(), event.getId(), event.getClass()));
}
@Bean
@ConditionalOnMissingBean(name = BUS_CONSUMER)
public BusConsumer busConsumer(ApplicationEventPublisher applicationEventPublisher, ServiceMatcher serviceMatcher,
BusBridge busBridge, BusProperties properties) {
return new BusConsumer(applicationEventPublisher, serviceMatcher, busBridge, properties);
}
@Configuration(proxyBeanMethods = false)
@ConditionalOnClass({ Endpoint.class })
@ConditionalOnBean(HttpTraceRepository.class)
@ConditionalOnProperty(value = "spring.cloud.bus.trace.enabled", matchIfMissing = false)
@ConditionalOnProperty(BusProperties.PREFIX + ".trace.enabled")
protected static class BusAckTraceConfiguration {
@Bean

25
spring-cloud-bus/src/main/java/org/springframework/cloud/bus/BusBridge.java Normal file
View File

@@ -0,0 +1,25 @@
/*
* Copyright 2015-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 org.springframework.cloud.bus;
import org.springframework.cloud.bus.event.RemoteApplicationEvent;
public interface BusBridge {
void send(RemoteApplicationEvent event);
}

23
spring-cloud-bus/src/main/java/org/springframework/cloud/bus/SpringCloudBusClient.java → spring-cloud-bus/src/main/java/org/springframework/cloud/bus/BusConstants.java
View File

@@ -16,31 +16,30 @@
package org.springframework.cloud.bus;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.SubscribableChannel;
/**
* @author Dave Syer
*
*/
public interface SpringCloudBusClient {
abstract class BusConstants {
/**
* Name of the input channel for Spring Cloud Bus.
*/
String INPUT = "springCloudBusInput";
public static final String INPUT = "springCloudBusInput";
/**
* Name of the output channel for Spring Cloud Bus.
*/
String OUTPUT = "springCloudBusOutput";
public static final String OUTPUT = "springCloudBusOutput";
@Output(SpringCloudBusClient.OUTPUT)
MessageChannel springCloudBusOutput();
/**
* Name of the output channel for Spring Cloud Bus.
*/
public static final String DESTINATION = "springCloudBus";
@Input(SpringCloudBusClient.INPUT)
SubscribableChannel springCloudBusInput();
/**
* Name of the Spring Cloud Bus function.
*/
public static final String BUS_CONSUMER = "busConsumer";
}

84
spring-cloud-bus/src/main/java/org/springframework/cloud/bus/BusConsumer.java Normal file
View File

@@ -0,0 +1,84 @@
/*
* Copyright 2015-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 org.springframework.cloud.bus;
import java.util.function.Consumer;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.cloud.bus.event.AckRemoteApplicationEvent;
import org.springframework.cloud.bus.event.RemoteApplicationEvent;
import org.springframework.cloud.bus.event.SentApplicationEvent;
import org.springframework.context.ApplicationEventPublisher;
public class BusConsumer implements Consumer<RemoteApplicationEvent> {
private final Log log = LogFactory.getLog(getClass());
private final ApplicationEventPublisher publisher;
private final ServiceMatcher serviceMatcher;
private final BusBridge busBridge;
private final BusProperties properties;
public BusConsumer(ApplicationEventPublisher publisher, ServiceMatcher serviceMatcher, BusBridge busBridge,
BusProperties properties) {
this.publisher = publisher;
this.serviceMatcher = serviceMatcher;
this.busBridge = busBridge;
this.properties = properties;
}
@Override
public void accept(RemoteApplicationEvent event) {
if (event instanceof AckRemoteApplicationEvent) {
if (this.properties.getTrace().isEnabled() && !this.serviceMatcher.isFromSelf(event)
&& this.publisher != null) {
this.publisher.publishEvent(event);
}
// If it's an ACK we are finished processing at this point
return;
}
if (log.isDebugEnabled()) {
log.debug("Received remote event from bus: " + event);
}
if (this.serviceMatcher.isForSelf(event) && this.publisher != null) {
if (!this.serviceMatcher.isFromSelf(event)) {
this.publisher.publishEvent(event);
}
if (this.properties.getAck().isEnabled()) {
AckRemoteApplicationEvent ack = new AckRemoteApplicationEvent(this, this.serviceMatcher.getServiceId(),
this.properties.getAck().getDestinationService(), event.getDestinationService(), event.getId(),
event.getClass());
this.busBridge.send(ack);
this.publisher.publishEvent(ack);
}
}
if (this.properties.getTrace().isEnabled() && this.publisher != null) {
// We are set to register sent events so publish it for local consumption,
// irrespective of the origin
this.publisher.publishEvent(new SentApplicationEvent(this, event.getOriginService(),
event.getDestinationService(), event.getId(), event.getClass()));
}
}
}

54
spring-cloud-bus/src/main/java/org/springframework/cloud/bus/BusEnvironmentPostProcessor.java
View File

@@ -22,11 +22,14 @@ import java.util.Map;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.env.EnvironmentPostProcessor;
import org.springframework.cloud.commons.util.IdUtils;
import org.springframework.cloud.function.context.FunctionProperties;
import org.springframework.core.env.ConfigurableEnvironment;
import org.springframework.core.env.MapPropertySource;
import org.springframework.core.env.MutablePropertySources;
import org.springframework.core.env.PropertySource;
import static org.springframework.cloud.bus.BusProperties.PREFIX;
/**
* {@link EnvironmentPostProcessor} that sets the default properties for the Bus.
*
@@ -35,21 +38,43 @@ import org.springframework.core.env.PropertySource;
*/
public class BusEnvironmentPostProcessor implements EnvironmentPostProcessor {
private static final String PROPERTY_SOURCE_NAME = "defaultProperties";
static final String DEFAULTS_PROPERTY_SOURCE_NAME = "springCloudBusDefaultProperties";
static final String OVERRIDES_PROPERTY_SOURCE_NAME = "springCloudBusOverridesProperties";
private static final String FN_DEF_PROP = FunctionProperties.PREFIX + ".definition";
@Override
public void postProcessEnvironment(ConfigurableEnvironment environment, SpringApplication application) {
Map<String, Object> map = new HashMap<String, Object>();
map.put("spring.cloud.stream.bindings." + SpringCloudBusClient.OUTPUT + ".content-type",
environment.getProperty("spring.cloud.bus.content-type", "application/json"));
map.put("spring.cloud.bus.id", IdUtils.getUnresolvedServiceId());
addOrReplace(environment.getPropertySources(), map);
Map<String, Object> overrides = new HashMap<>();
String definition = BusConstants.BUS_CONSUMER;
if (environment.containsProperty(FN_DEF_PROP)) {
String property = environment.getProperty(FN_DEF_PROP);
if (property != null && property.contains(BusConstants.BUS_CONSUMER)) {
// in the case that EnvironmentPostProcessor are run more than once.
return;
}
definition = property + ";" + definition;
}
overrides.put(FN_DEF_PROP, definition);
addOrReplace(environment.getPropertySources(), overrides, OVERRIDES_PROPERTY_SOURCE_NAME, true);
Map<String, Object> defaults = new HashMap<>();
defaults.put("spring.cloud.stream.function.bindings." + BusConstants.BUS_CONSUMER + "-in-0",
BusConstants.INPUT);
String destination = environment.getProperty(PREFIX + ".destination", BusConstants.DESTINATION);
defaults.put("spring.cloud.stream.bindings." + BusConstants.INPUT + ".destination", destination);
if (!environment.containsProperty(PREFIX + ".id")) {
defaults.put(PREFIX + ".id", IdUtils.getUnresolvedServiceId());
}
addOrReplace(environment.getPropertySources(), defaults, DEFAULTS_PROPERTY_SOURCE_NAME, false);
}
private void addOrReplace(MutablePropertySources propertySources, Map<String, Object> map) {
private void addOrReplace(MutablePropertySources propertySources, Map<String, Object> map,
String propertySourceName, boolean first) {
MapPropertySource target = null;
if (propertySources.contains(PROPERTY_SOURCE_NAME)) {
PropertySource<?> source = propertySources.get(PROPERTY_SOURCE_NAME);
if (propertySources.contains(propertySourceName)) {
PropertySource<?> source = propertySources.get(propertySourceName);
if (source instanceof MapPropertySource) {
target = (MapPropertySource) source;
for (String key : map.keySet()) {
@@ -60,10 +85,15 @@ public class BusEnvironmentPostProcessor implements EnvironmentPostProcessor {
}
}
if (target == null) {
target = new MapPropertySource(PROPERTY_SOURCE_NAME, map);
target = new MapPropertySource(propertySourceName, map);
}
if (!propertySources.contains(PROPERTY_SOURCE_NAME)) {
propertySources.addLast(target);
if (!propertySources.contains(propertySourceName)) {
if (first) {
propertySources.addFirst(target);
}
else {
propertySources.addLast(target);
}
}
}

87
spring-cloud-bus/src/main/java/org/springframework/cloud/bus/BusProperties.java
View File

@@ -17,57 +17,52 @@
package org.springframework.cloud.bus;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.core.style.ToStringCreator;
import org.springframework.util.MimeType;
import org.springframework.util.MimeTypeUtils;
/**
* @author Dave Syer
*
*/
@ConfigurationProperties("spring.cloud.bus")
@ConfigurationProperties(BusProperties.PREFIX)
public class BusProperties {
/**
* Environment change event related properties.
* Configuration prefix for spring cloud bus.
*/
private Env env = new Env();
/**
* Refresh event related properties.
*/
private Refresh refresh = new Refresh();
public static final String PREFIX = "spring.cloud.bus";
/**
* Properties related to acks.
*/
private Ack ack = new Ack();
private final Ack ack = new Ack();
/**
* Properties related to tracing of acks.
*/
private Trace trace = new Trace();
private final Trace trace = new Trace();
/**
* Name of Spring Cloud Stream destination for messages.
*/
private String destination = "springCloudBus";
private String destination = BusConstants.DESTINATION;
/**
* The identifier for this application instance.
*/
private String id = "application";
/**
* The bus mime-type.
*/
private MimeType contentType = MimeTypeUtils.APPLICATION_JSON;
/**
* Flag to indicate that the bus is enabled.
*/
private boolean enabled = true;
public Env getEnv() {
return this.env;
}
public Refresh getRefresh() {
return this.refresh;
}
public Ack getAck() {
return this.ack;
}
@@ -100,43 +95,18 @@ public class BusProperties {
this.id = id;
}
/**
* Spring Cloud Bus environment related properties.
*/
public static class Env {
/**
* Flag to switch off environment change events (default on).
*/
private boolean enabled = true;
public boolean isEnabled() {
return this.enabled;
}
public void setEnabled(boolean enabled) {
this.enabled = enabled;
}
public MimeType getContentType() {
return this.contentType;
}
/**
* Spring Cloud Bus properties related to refreshing.
*/
public static class Refresh {
public void setContentType(MimeType contentType) {
this.contentType = contentType;
}
/**
* Flag to switch off refresh events (default on).
*/
private boolean enabled = true;
public boolean isEnabled() {
return this.enabled;
}
public void setEnabled(boolean enabled) {
this.enabled = enabled;
}
@Override
public String toString() {
return new ToStringCreator(this).append("ack", ack).append("trace", trace).append("destination", destination)
.append("id", id).append("contentType", contentType).append("enabled", enabled).toString();
}
@@ -171,6 +141,12 @@ public class BusProperties {
this.destinationService = destinationService;
}
@Override
public String toString() {
return new ToStringCreator(this).append("enabled", enabled).append("destinationService", destinationService)
.toString();
}
}
/**
@@ -191,6 +167,11 @@ public class BusProperties {
this.enabled = enabled;
}
@Override
public String toString() {
return new ToStringCreator(this).append("enabled", enabled).toString();
}
}
}

16
spring-cloud-bus/src/main/java/org/springframework/cloud/bus/DefaultBusPathMatcher.java
View File

@@ -49,7 +49,9 @@ public class DefaultBusPathMatcher implements PathMatcher {
protected boolean matchMultiProfile(String pattern, String idToMatch) {
log.debug("matchMultiProfile : " + pattern + ", " + idToMatch);
if (log.isDebugEnabled()) {
log.debug("matchMultiProfile : " + pattern + ", " + idToMatch);
}
// parse the id
String[] tokens = tokenizeToStringArray(idToMatch, ":");
@@ -82,12 +84,16 @@ public class DefaultBusPathMatcher implements PathMatcher {
for (String id : idsWithSingleProfile) {
if (this.delagateMatcher.match(pattern, id)) {
log.debug("matched true");
if (log.isDebugEnabled()) {
log.debug("matched true");
}
return true;
}
}
log.debug("matched false");
if (log.isDebugEnabled()) {
log.debug("matched false");
}
return false;
}
@@ -98,7 +104,9 @@ public class DefaultBusPathMatcher implements PathMatcher {
@Override
public boolean match(String pattern, String path) {
log.debug("In match: " + pattern + ", " + path);
if (log.isDebugEnabled()) {
log.debug("In match: " + pattern + ", " + path);
}
if (!this.delagateMatcher.match(pattern, path)) {
return matchMultiProfile(pattern, path);
}

50
spring-cloud-bus/src/main/java/org/springframework/cloud/bus/RemoteApplicationEventListener.java Normal file
View File

@@ -0,0 +1,50 @@
/*
* Copyright 2015-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 org.springframework.cloud.bus;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.cloud.bus.event.AckRemoteApplicationEvent;
import org.springframework.cloud.bus.event.RemoteApplicationEvent;
import org.springframework.context.ApplicationListener;
public class RemoteApplicationEventListener implements ApplicationListener<RemoteApplicationEvent> {
private final Log log = LogFactory.getLog(getClass());
private final ServiceMatcher serviceMatcher;
private final BusBridge busBridge;
public RemoteApplicationEventListener(ServiceMatcher serviceMatcher, BusBridge busBridge) {
this.serviceMatcher = serviceMatcher;
this.busBridge = busBridge;
}
@Override
public void onApplicationEvent(RemoteApplicationEvent event) {
if (this.serviceMatcher.isFromSelf(event) && !(event instanceof AckRemoteApplicationEvent)) {
if (log.isDebugEnabled()) {
log.debug("Sending remote event on bus: " + event);
}
// TODO: configurable mimetype?
this.busBridge.send(event);
}
}
}

19
spring-cloud-bus/src/main/java/org/springframework/cloud/bus/ServiceMatcherAutoConfiguration.java
View File

@@ -24,8 +24,6 @@ import org.springframework.core.env.Environment;
import org.springframework.util.AntPathMatcher;
import org.springframework.util.PathMatcher;
import static org.springframework.cloud.bus.BusAutoConfiguration.CLOUD_CONFIG_NAME_PROPERTY;
/**
* @author Ryan Baxter
*/
@@ -34,10 +32,20 @@ import static org.springframework.cloud.bus.BusAutoConfiguration.CLOUD_CONFIG_NA
@EnableConfigurationProperties(BusProperties.class)
public class ServiceMatcherAutoConfiguration {
/**
* Name of the Bus path matcher.
*/
public static final String BUS_PATH_MATCHER_NAME = "busPathMatcher";
/**
* Name of the Spring Cloud Config property.
*/
public static final String CLOUD_CONFIG_NAME_PROPERTY = "spring.cloud.config.name";
@BusPathMatcher
// There is a @Bean of type PathMatcher coming from Spring MVC
@ConditionalOnMissingBean(name = BusAutoConfiguration.BUS_PATH_MATCHER_NAME)
@Bean(name = BusAutoConfiguration.BUS_PATH_MATCHER_NAME)
@ConditionalOnMissingBean(name = BUS_PATH_MATCHER_NAME)
@Bean(name = BUS_PATH_MATCHER_NAME)
public PathMatcher busPathMatcher() {
return new DefaultBusPathMatcher(new AntPathMatcher(":"));
}
@@ -46,8 +54,7 @@ public class ServiceMatcherAutoConfiguration {
public ServiceMatcher serviceMatcher(@BusPathMatcher PathMatcher pathMatcher, BusProperties properties,
Environment environment) {
String[] configNames = environment.getProperty(CLOUD_CONFIG_NAME_PROPERTY, String[].class, new String[] {});
ServiceMatcher serviceMatcher = new ServiceMatcher(pathMatcher, properties.getId(), configNames);
return serviceMatcher;
return new ServiceMatcher(pathMatcher, properties.getId(), configNames);
}
}

39
spring-cloud-bus/src/main/java/org/springframework/cloud/bus/StreamBusBridge.java Normal file
View File

@@ -0,0 +1,39 @@
/*
* Copyright 2015-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 org.springframework.cloud.bus;
import org.springframework.cloud.bus.event.RemoteApplicationEvent;
import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.messaging.support.MessageBuilder;
public class StreamBusBridge implements BusBridge {
private final StreamBridge streamBridge;
private final BusProperties properties;
public StreamBusBridge(StreamBridge streamBridge, BusProperties properties) {
this.streamBridge = streamBridge;
this.properties = properties;
}
public void send(RemoteApplicationEvent event) {
// TODO: configurable mimetype?
this.streamBridge.send(properties.getDestination(), MessageBuilder.withPayload(event).build());
}
}

23
spring-cloud-bus/src/main/resources/META-INF/additional-spring-configuration-metadata.json Normal file
View File

@@ -0,0 +1,23 @@
{
"properties": [
{
"name": "spring.cloud.bus.env.enabled",
"type": "java.lang.Boolean",
"description": "Flag to switch off environment change events (default on).",
"defaultValue": true
},
{
"name": "spring.cloud.bus.refresh.enabled",
"type": "java.lang.Boolean",
"description": "Flag to switch off refresh events (default on).",
"defaultValue": true
},
{
"name": "spring.cloud.bus.trace.enabled",
"type": "java.lang.Boolean",
"description": "Flag to switch on tracing of acks (default off).",
"defaultValue": false
}
]
}

170
spring-cloud-bus/src/test/java/org/springframework/cloud/bus/BusAutoConfigurationTests.java
View File

@@ -20,36 +20,29 @@ import java.util.HashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import javax.annotation.PostConstruct;
import org.junit.After;
import org.junit.Ignore;
import org.junit.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.autoconfigure.ImportAutoConfiguration;
import org.springframework.boot.autoconfigure.context.PropertyPlaceholderAutoConfiguration;
import org.springframework.cloud.bus.event.AckRemoteApplicationEvent;
import org.springframework.cloud.bus.event.RefreshRemoteApplicationEvent;
import org.springframework.cloud.bus.event.RemoteApplicationEvent;
import org.springframework.cloud.bus.event.SentApplicationEvent;
import org.springframework.cloud.bus.event.UnknownRemoteApplicationEvent;
import org.springframework.cloud.context.refresh.ContextRefresher;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.config.BindingProperties;
import org.springframework.cloud.stream.config.BindingServiceProperties;
import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.cloud.stream.test.binder.TestSupportBinderAutoConfiguration;
import org.springframework.context.ApplicationListener;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.integration.annotation.MessageEndpoint;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.messaging.Message;
import org.springframework.context.annotation.Primary;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.ChannelInterceptor;
import org.springframework.messaging.support.ChannelInterceptorAdapter;
import org.springframework.messaging.support.GenericMessage;
import static org.assertj.core.api.Assertions.assertThat;
@@ -78,7 +71,7 @@ public class BusAutoConfigurationTests {
public void inboundNotForSelf() {
this.context = SpringApplication.run(InboundMessageHandlerConfiguration.class, "--spring.cloud.bus.id=foo",
"--server.port=0");
this.context.getBean(SpringCloudBusClient.INPUT, MessageChannel.class)
this.context.getBean(BusConstants.INPUT, MessageChannel.class)
.send(new GenericMessage<>(new RefreshRemoteApplicationEvent(this, "bar", "bar")));
assertThat(this.context.getBean(InboundMessageHandlerConfiguration.class).refresh).isNull();
}
@@ -87,7 +80,7 @@ public class BusAutoConfigurationTests {
public void inboundFromSelf() {
this.context = SpringApplication.run(InboundMessageHandlerConfiguration.class, "--spring.cloud.bus.id=foo",
"--server.port=0");
this.context.getBean(SpringCloudBusClient.INPUT, MessageChannel.class)
this.context.getBean(BusConstants.INPUT, MessageChannel.class)
.send(new GenericMessage<>(new RefreshRemoteApplicationEvent(this, "foo", null)));
assertThat(this.context.getBean(InboundMessageHandlerConfiguration.class).refresh).isNull();
}
@@ -96,7 +89,7 @@ public class BusAutoConfigurationTests {
public void inboundNotFromSelf() {
this.context = SpringApplication.run(InboundMessageHandlerConfiguration.class, "--spring.cloud.bus.id=bar",
"--server.port=0");
this.context.getBean(SpringCloudBusClient.INPUT, MessageChannel.class)
this.context.getBean(BusConstants.INPUT, MessageChannel.class)
.send(new GenericMessage<>(new RefreshRemoteApplicationEvent(this, "foo", null)));
assertThat(this.context.getBean(InboundMessageHandlerConfiguration.class).refresh).isNotNull();
}
@@ -106,26 +99,27 @@ public class BusAutoConfigurationTests {
this.context = SpringApplication.run(
new Class[] { InboundMessageHandlerConfiguration.class, OutboundMessageHandlerConfiguration.class,
SentMessageConfiguration.class },
new String[] { "--spring.cloud.bus.id=bar", "--server.port=0" });
this.context.getBean(SpringCloudBusClient.INPUT, MessageChannel.class)
new String[] { "--spring.cloud.bus.id=bar", "--server.port=0",
"--spring.main.allow-bean-definition-overriding=true" });
this.context.getBean(BusConstants.INPUT, MessageChannel.class)
.send(new GenericMessage<>(new RefreshRemoteApplicationEvent(this, "foo", null)));
RefreshRemoteApplicationEvent refresh = this.context.getBean(InboundMessageHandlerConfiguration.class).refresh;
assertThat(refresh).isNotNull();
OutboundMessageHandlerConfiguration outbound = this.context.getBean(OutboundMessageHandlerConfiguration.class);
outbound.latch.await(2000L, TimeUnit.MILLISECONDS);
String message = (String) outbound.message.getPayload();
assertThat(message.contains("\"ackId\":\"" + refresh.getId())).as("Wrong ackId: " + message).isTrue();
TestStreamBusBridge busBridge = this.context.getBean(TestStreamBusBridge.class);
busBridge.latch.await(200, TimeUnit.SECONDS);
assertThat(busBridge.message).isInstanceOf(AckRemoteApplicationEvent.class);
AckRemoteApplicationEvent message = (AckRemoteApplicationEvent) busBridge.message;
assertThat(message.getAckId()).as("Wrong ackId: %s", message).isEqualTo(refresh.getId());
}
@Test
public void inboundNotFromSelfWithTrace() throws Exception {
public void inboundNotFromSelfWithTrace() {
this.context = SpringApplication.run(
new Class[] { InboundMessageHandlerConfiguration.class, OutboundMessageHandlerConfiguration.class,
SentMessageConfiguration.class },
new String[] { "--spring.cloud.bus.trace.enabled=true", "--spring.cloud.bus.id=bar",
"--server.port=0" });
this.context.getBean(SpringCloudBusClient.INPUT, MessageChannel.class)
.send(new GenericMessage<>(new RefreshRemoteApplicationEvent(this, "foo", null)));
this.context.getBean(BusConsumer.class).accept(new RefreshRemoteApplicationEvent(this, "foo", null));
RefreshRemoteApplicationEvent refresh = this.context.getBean(InboundMessageHandlerConfiguration.class).refresh;
assertThat(refresh).isNotNull();
SentMessageConfiguration sent = this.context.getBean(SentMessageConfiguration.class);
@@ -134,18 +128,18 @@ public class BusAutoConfigurationTests {
}
@Test
public void inboundAckWithTrace() throws Exception {
public void inboundAckWithTrace() throws InterruptedException {
this.context = SpringApplication.run(
new Class[] { InboundMessageHandlerConfiguration.class, OutboundMessageHandlerConfiguration.class,
AckMessageConfiguration.class },
new String[] { "--spring.cloud.bus.trace.enabled=true", "--spring.cloud.bus.id=bar",
"--server.port=0" });
this.context.getBean(BusProperties.class).setId("bar");
this.context.getBean(SpringCloudBusClient.INPUT, MessageChannel.class).send(new GenericMessage<>(
new AckRemoteApplicationEvent(this, "foo", null, "ID", "bar", RefreshRemoteApplicationEvent.class)));
AckMessageConfiguration sent = this.context.getBean(AckMessageConfiguration.class);
assertThat(sent.event).isNotNull();
assertThat(sent.count).isEqualTo(1);
this.context.getBean(BusConsumer.class).accept(
new AckRemoteApplicationEvent(this, "foo", null, "ID", "bar", RefreshRemoteApplicationEvent.class));
AckMessageConfiguration ack = this.context.getBean(AckMessageConfiguration.class);
assertThat(ack.latch.await(5, TimeUnit.SECONDS)).isTrue();
assertThat(ack.event).isNotNull();
assertThat(ack.count).isEqualTo(1);
}
@Test
@@ -153,9 +147,9 @@ public class BusAutoConfigurationTests {
this.context = SpringApplication.run(OutboundMessageHandlerConfiguration.class, "--debug=true",
"--spring.cloud.bus.id=foo", "--server.port=0");
this.context.publishEvent(new RefreshRemoteApplicationEvent(this, "foo", null));
OutboundMessageHandlerConfiguration outbound = this.context.getBean(OutboundMessageHandlerConfiguration.class);
outbound.latch.await(2000L, TimeUnit.MILLISECONDS);
assertThat(outbound.message).as("message was null").isNotNull();
TestStreamBusBridge busBridge = this.context.getBean(TestStreamBusBridge.class);
busBridge.latch.await(2, TimeUnit.SECONDS);
assertThat(busBridge.message).as("message was null").isNotNull();
}
@Test
@@ -163,14 +157,14 @@ public class BusAutoConfigurationTests {
this.context = SpringApplication.run(OutboundMessageHandlerConfiguration.class, "--spring.cloud.bus.id=bar",
"--server.port=0");
this.context.publishEvent(new RefreshRemoteApplicationEvent(this, "foo", null));
assertThat(this.context.getBean(OutboundMessageHandlerConfiguration.class).message).isNull();
assertThat(this.context.getBean(TestStreamBusBridge.class).message).isNull();
}
@Test
public void inboundNotFromSelfPathPattern() {
this.context = SpringApplication.run(InboundMessageHandlerConfiguration.class, "--spring.cloud.bus.id=bar:1000",
"--server.port=0");
this.context.getBean(SpringCloudBusClient.INPUT, MessageChannel.class)
this.context.getBean(BusConstants.INPUT, MessageChannel.class)
.send(new GenericMessage<>(new RefreshRemoteApplicationEvent(this, "foo", "bar:*")));
assertThat(this.context.getBean(InboundMessageHandlerConfiguration.class).refresh).isNotNull();
}
@@ -179,7 +173,7 @@ public class BusAutoConfigurationTests {
public void inboundNotFromSelfDeepPathPattern() {
this.context = SpringApplication.run(InboundMessageHandlerConfiguration.class,
"--spring.cloud.bus.id=bar:test:1000", "--server.port=0");
this.context.getBean(SpringCloudBusClient.INPUT, MessageChannel.class)
this.context.getBean(BusConstants.INPUT, MessageChannel.class)
.send(new GenericMessage<>(new RefreshRemoteApplicationEvent(this, "foo", "bar:**")));
assertThat(this.context.getBean(InboundMessageHandlerConfiguration.class).refresh).isNotNull();
}
@@ -188,7 +182,7 @@ public class BusAutoConfigurationTests {
public void inboundNotFromSelfFlatPattern() {
this.context = SpringApplication.run(InboundMessageHandlerConfiguration.class, "--spring.cloud.bus.id=bar",
"--server.port=0");
this.context.getBean(SpringCloudBusClient.INPUT, MessageChannel.class)
this.context.getBean(BusConstants.INPUT, MessageChannel.class)
.send(new GenericMessage<>(new RefreshRemoteApplicationEvent(this, "foo", "bar*")));
assertThat(this.context.getBean(InboundMessageHandlerConfiguration.class).refresh).isNotNull();
}
@@ -198,79 +192,44 @@ public class BusAutoConfigurationTests {
public void inboundNotFromSelfUnknown() {
this.context = SpringApplication.run(InboundMessageHandlerConfiguration.class, "--spring.cloud.bus.id=bar",
"--server.port=0");
this.context.getBean(SpringCloudBusClient.INPUT, MessageChannel.class)
this.context.getBean(BusConstants.INPUT, MessageChannel.class)
.send(new GenericMessage<>(new UnknownRemoteApplicationEvent(this, "UnknownEvent", "yada".getBytes())));
// No Exception expected
}
@Test
public void initSetsBindingDestinationIfNullDefault() {
HashMap<String, BindingProperties> properties = new HashMap<>();
properties.put(SpringCloudBusClient.INPUT, new BindingProperties());
properties.put(SpringCloudBusClient.OUTPUT, new BindingProperties());
testDestinations(properties);
}
@Test
public void initSetsBindingDestinationIfNotNullDefault() {
HashMap<String, BindingProperties> properties = new HashMap<>();
BindingProperties input = new BindingProperties();
input.setDestination(SpringCloudBusClient.INPUT);
properties.put(SpringCloudBusClient.INPUT, input);
BindingProperties output = new BindingProperties();
output.setDestination(SpringCloudBusClient.OUTPUT);
properties.put(SpringCloudBusClient.OUTPUT, output);
testDestinations(properties);
}
@Test
public void initDoesNotOverrideCustomDestination() {
HashMap<String, BindingProperties> properties = new HashMap<>();
BindingProperties input = new BindingProperties();
input.setDestination("mydestination");
properties.put(SpringCloudBusClient.INPUT, input);
properties.put(BusConstants.INPUT, input);
BindingProperties output = new BindingProperties();
output.setDestination("mydestination");
properties.put(SpringCloudBusClient.OUTPUT, output);
properties.put(BusConstants.OUTPUT, output);
setupBusAutoConfig(properties);
BindingProperties inputProps = properties.get(SpringCloudBusClient.INPUT);
BindingProperties inputProps = properties.get(BusConstants.INPUT);
assertThat(inputProps.getDestination()).isEqualTo("mydestination");
BindingProperties outputProps = properties.get(SpringCloudBusClient.OUTPUT);
BindingProperties outputProps = properties.get(BusConstants.OUTPUT);
assertThat(outputProps.getDestination()).isEqualTo("mydestination");
}
private void testDestinations(HashMap<String, BindingProperties> properties) {
BusProperties bus = setupBusAutoConfig(properties);
BindingProperties input = properties.get(SpringCloudBusClient.INPUT);
assertThat(input.getDestination()).isEqualTo(bus.getDestination());
BindingProperties output = properties.get(SpringCloudBusClient.OUTPUT);
assertThat(output.getDestination()).isEqualTo(bus.getDestination());
}
private BusProperties setupBusAutoConfig(HashMap<String, BindingProperties> properties) {
BindingServiceProperties serviceProperties = mock(BindingServiceProperties.class);
when(serviceProperties.getBindings()).thenReturn(properties);
BusProperties bus = new BusProperties();
BusAutoConfiguration configuration = new BusAutoConfiguration(mock(ServiceMatcher.class), serviceProperties,
bus);
configuration.init();
BusAutoConfiguration configuration = new BusAutoConfiguration();
return bus;
}
// see https://github.com/spring-cloud/spring-cloud-bus/issues/101
@Test
@Ignore // TODO: replicate problem
public void serviceMatcherIdIsConstantAfterRefresh() {
this.context = SpringApplication.run(new Class[] { RefreshConfig.class, },
new String[] { "--spring.main.allow-bean-definition-overriding=true" });
new String[] { "--server.port=0", "--spring.main.allow-bean-definition-overriding=true" });
String originalServiceId = this.context.getBean(ServiceMatcher.class).getServiceId();
this.context.getBean(ContextRefresher.class).refresh();
String newServiceId = this.context.getBean(ServiceMatcher.class).getServiceId();
@@ -285,48 +244,40 @@ public class BusAutoConfigurationTests {
@Configuration(proxyBeanMethods = false)
@EnableAutoConfiguration
@Import({ MessageConsumer.class, BusAutoConfiguration.class, TestSupportBinderAutoConfiguration.class,
@ImportAutoConfiguration({ BusAutoConfiguration.class, TestSupportBinderAutoConfiguration.class,
PropertyPlaceholderAutoConfiguration.class })
protected static class OutboundMessageHandlerConfiguration {
@Autowired
@Output(SpringCloudBusClient.OUTPUT)
private MessageChannel cloudBusOutboundChannel;
private CountDownLatch latch = new CountDownLatch(1);
private Message<?> message;
@PostConstruct
public void init() {
((DirectChannel) this.cloudBusOutboundChannel).addInterceptor(interceptor());
}
private ChannelInterceptor interceptor() {
return new ChannelInterceptorAdapter() {
@Override
public void postSend(Message<?> message, MessageChannel channel, boolean sent) {
OutboundMessageHandlerConfiguration.this.message = message;
OutboundMessageHandlerConfiguration.this.latch.countDown();
}
};
@Bean
@Primary
StreamBusBridge testStreamBusBridge(StreamBridge streamBridge, BusProperties properties) {
return new TestStreamBusBridge(streamBridge, properties);
}
}
@Configuration(proxyBeanMethods = false)
@MessageEndpoint
protected static class MessageConsumer {
protected static class TestStreamBusBridge extends StreamBusBridge {
@ServiceActivator(inputChannel = SpringCloudBusClient.OUTPUT)
public void handle(Message<?> msg) {
private CountDownLatch latch = new CountDownLatch(1);
private RemoteApplicationEvent message;
public TestStreamBusBridge(StreamBridge streamBridge, BusProperties properties) {
super(streamBridge, properties);
}
@Override
public void send(RemoteApplicationEvent event) {
latch.countDown();
message = event;
super.send(event);
}
}
@Configuration(proxyBeanMethods = false)
@EnableAutoConfiguration
@Import({ MessageConsumer.class, BusAutoConfiguration.class, TestSupportBinderAutoConfiguration.class,
@ImportAutoConfiguration({ BusAutoConfiguration.class, TestSupportBinderAutoConfiguration.class,
PropertyPlaceholderAutoConfiguration.class })
protected static class InboundMessageHandlerConfiguration
implements ApplicationListener<RefreshRemoteApplicationEvent> {
@@ -358,6 +309,8 @@ public class BusAutoConfigurationTests {
@Configuration(proxyBeanMethods = false)
protected static class AckMessageConfiguration implements ApplicationListener<AckRemoteApplicationEvent> {
private CountDownLatch latch = new CountDownLatch(1);
private AckRemoteApplicationEvent event;
private int count;
@@ -366,6 +319,7 @@ public class BusAutoConfigurationTests {
public void onApplicationEvent(AckRemoteApplicationEvent event) {
this.event = event;
this.count++;
latch.countDown();
}
}

64
spring-cloud-bus/src/test/java/org/springframework/cloud/bus/BusEnvironmentPostProcessorTests.java Normal file
View File

@@ -0,0 +1,64 @@
/*
* Copyright 2015-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 org.springframework.cloud.bus;
import org.junit.jupiter.api.Test;
import org.springframework.boot.SpringApplication;
import org.springframework.cloud.function.context.FunctionProperties;
import org.springframework.mock.env.MockEnvironment;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.mock;
import static org.springframework.cloud.bus.BusConstants.BUS_CONSUMER;
import static org.springframework.cloud.bus.BusConstants.DESTINATION;
import static org.springframework.cloud.bus.BusConstants.INPUT;
import static org.springframework.cloud.bus.BusEnvironmentPostProcessor.DEFAULTS_PROPERTY_SOURCE_NAME;
import static org.springframework.cloud.bus.BusEnvironmentPostProcessor.OVERRIDES_PROPERTY_SOURCE_NAME;
public class BusEnvironmentPostProcessorTests {
@Test
void testDefaults() {
MockEnvironment env = new MockEnvironment().withProperty("cachedrandom.application.value", "123");
new BusEnvironmentPostProcessor().postProcessEnvironment(env, mock(SpringApplication.class));
assertThat(env.getProperty(FunctionProperties.PREFIX + ".definition")).isEqualTo(BUS_CONSUMER);
assertThat(env.getProperty("spring.cloud.stream.function.bindings." + BUS_CONSUMER + "-in-0")).isEqualTo(INPUT);
assertThat(env.getProperty("spring.cloud.stream.bindings." + INPUT + ".destination")).isEqualTo(DESTINATION);
assertThat(env.getProperty(BusProperties.PREFIX + ".id")).isNotBlank();
assertThat(env.getPropertySources().contains(OVERRIDES_PROPERTY_SOURCE_NAME));
assertThat(env.getPropertySources().contains(DEFAULTS_PROPERTY_SOURCE_NAME));
}
@Test
void testOverrides() {
String fnDefKey = FunctionProperties.PREFIX + ".definition";
String idKey = BusProperties.PREFIX + ".id";
MockEnvironment env = new MockEnvironment().withProperty("cachedrandom.application.value", "123")
.withProperty(BusProperties.PREFIX + ".destination", "mydestination").withProperty(idKey, "app:1")
.withProperty(fnDefKey, "uppercase");
new BusEnvironmentPostProcessor().postProcessEnvironment(env, mock(SpringApplication.class));
assertThat(env.getProperty(fnDefKey)).isEqualTo("uppercase;" + BUS_CONSUMER);
assertThat(env.getProperty("spring.cloud.stream.function.bindings." + BUS_CONSUMER + "-in-0")).isEqualTo(INPUT);
assertThat(env.getProperty("spring.cloud.stream.bindings." + INPUT + ".destination"))
.isEqualTo("mydestination");
assertThat(env.getProperty(idKey)).isEqualTo("app:1");
assertThat(env.getPropertySources().contains(OVERRIDES_PROPERTY_SOURCE_NAME));
assertThat(env.getPropertySources().contains(DEFAULTS_PROPERTY_SOURCE_NAME));
}
}