diff --git a/.circleci/config.yml b/.circleci/config.yml index 8252142..cd47046 100644 --- a/.circleci/config.yml +++ b/.circleci/config.yml @@ -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: diff --git a/spring-cloud-bus-tests/pom.xml b/spring-cloud-bus-tests/pom.xml index 899701d..9ebe2ae 100644 --- a/spring-cloud-bus-tests/pom.xml +++ b/spring-cloud-bus-tests/pom.xml @@ -17,12 +17,21 @@ .. + + 1.15.0-rc1 + + org.springframework.boot spring-boot-starter-web test + + org.springframework.boot + spring-boot-starter-webflux + test + org.springframework.boot spring-boot-starter-actuator @@ -30,7 +39,7 @@ org.springframework.cloud - spring-cloud-bus + spring-cloud-starter-bus-amqp test @@ -45,12 +54,19 @@ org.springframework.cloud - spring-cloud-starter-stream-rabbit + spring-cloud-stream-test-support test - org.springframework.cloud - spring-cloud-stream-test-support + org.testcontainers + junit-jupiter + ${testcontainers.version} + test + + + org.testcontainers + rabbitmq + ${testcontainers.version} test diff --git a/spring-cloud-bus-tests/src/test/java/org/springframework/cloud/bus/BusAmqpIntegrationTests.java b/spring-cloud-bus-tests/src/test/java/org/springframework/cloud/bus/BusAmqpIntegrationTests.java new file mode 100644 index 0000000..7bbab10 --- /dev/null +++ b/spring-cloud-bus-tests/src/test/java/org/springframework/cloud/bus/BusAmqpIntegrationTests.java @@ -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 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 { + + CountDownLatch latch = new CountDownLatch(1); + + @Override + public void onApplicationEvent(EnvironmentChangeRemoteApplicationEvent event) { + latch.countDown(); + } + + } + +} diff --git a/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/BusAutoConfiguration.java b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/BusAutoConfiguration.java index f84f159..e80eaaf 100644 --- a/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/BusAutoConfiguration.java +++ b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/BusAutoConfiguration.java @@ -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 diff --git a/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/BusBridge.java b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/BusBridge.java new file mode 100644 index 0000000..be95194 --- /dev/null +++ b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/BusBridge.java @@ -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); + +} diff --git a/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/SpringCloudBusClient.java b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/BusConstants.java similarity index 61% rename from spring-cloud-bus/src/main/java/org/springframework/cloud/bus/SpringCloudBusClient.java rename to spring-cloud-bus/src/main/java/org/springframework/cloud/bus/BusConstants.java index 9558dc5..3f997e1 100644 --- a/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/SpringCloudBusClient.java +++ b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/BusConstants.java @@ -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"; } diff --git a/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/BusConsumer.java b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/BusConsumer.java new file mode 100644 index 0000000..3cd9022 --- /dev/null +++ b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/BusConsumer.java @@ -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 { + + 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())); + } + } + +} diff --git a/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/BusEnvironmentPostProcessor.java b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/BusEnvironmentPostProcessor.java index b41228c..4629aba 100644 --- a/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/BusEnvironmentPostProcessor.java +++ b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/BusEnvironmentPostProcessor.java @@ -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 map = new HashMap(); - 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 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 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 map) { + private void addOrReplace(MutablePropertySources propertySources, Map 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); + } } } diff --git a/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/BusProperties.java b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/BusProperties.java index c1dc203..0daa9a1 100644 --- a/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/BusProperties.java +++ b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/BusProperties.java @@ -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(); + } + } } diff --git a/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/DefaultBusPathMatcher.java b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/DefaultBusPathMatcher.java index 98dc20d..800f990 100644 --- a/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/DefaultBusPathMatcher.java +++ b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/DefaultBusPathMatcher.java @@ -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); } diff --git a/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/RemoteApplicationEventListener.java b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/RemoteApplicationEventListener.java new file mode 100644 index 0000000..d7581f8 --- /dev/null +++ b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/RemoteApplicationEventListener.java @@ -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 { + + 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); + } + } + +} diff --git a/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/ServiceMatcherAutoConfiguration.java b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/ServiceMatcherAutoConfiguration.java index d80b67f..fe090e0 100644 --- a/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/ServiceMatcherAutoConfiguration.java +++ b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/ServiceMatcherAutoConfiguration.java @@ -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); } } diff --git a/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/StreamBusBridge.java b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/StreamBusBridge.java new file mode 100644 index 0000000..ec592a6 --- /dev/null +++ b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/StreamBusBridge.java @@ -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()); + } + +} diff --git a/spring-cloud-bus/src/main/resources/META-INF/additional-spring-configuration-metadata.json b/spring-cloud-bus/src/main/resources/META-INF/additional-spring-configuration-metadata.json new file mode 100644 index 0000000..0671408 --- /dev/null +++ b/spring-cloud-bus/src/main/resources/META-INF/additional-spring-configuration-metadata.json @@ -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 + } + ] +} + diff --git a/spring-cloud-bus/src/test/java/org/springframework/cloud/bus/BusAutoConfigurationTests.java b/spring-cloud-bus/src/test/java/org/springframework/cloud/bus/BusAutoConfigurationTests.java index 058cbbe..93d7d5f 100644 --- a/spring-cloud-bus/src/test/java/org/springframework/cloud/bus/BusAutoConfigurationTests.java +++ b/spring-cloud-bus/src/test/java/org/springframework/cloud/bus/BusAutoConfigurationTests.java @@ -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 properties = new HashMap<>(); - properties.put(SpringCloudBusClient.INPUT, new BindingProperties()); - properties.put(SpringCloudBusClient.OUTPUT, new BindingProperties()); - - testDestinations(properties); - } - - @Test - public void initSetsBindingDestinationIfNotNullDefault() { - HashMap 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 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 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 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 { @@ -358,6 +309,8 @@ public class BusAutoConfigurationTests { @Configuration(proxyBeanMethods = false) protected static class AckMessageConfiguration implements ApplicationListener { + 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(); } } diff --git a/spring-cloud-bus/src/test/java/org/springframework/cloud/bus/BusEnvironmentPostProcessorTests.java b/spring-cloud-bus/src/test/java/org/springframework/cloud/bus/BusEnvironmentPostProcessorTests.java new file mode 100644 index 0000000..9ab7977 --- /dev/null +++ b/spring-cloud-bus/src/test/java/org/springframework/cloud/bus/BusEnvironmentPostProcessorTests.java @@ -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)); + } + +}