diff --git a/.mvn/maven.config b/.mvn/maven.config index 3b8cf46..381df16 100644 --- a/.mvn/maven.config +++ b/.mvn/maven.config @@ -1 +1 @@ --DaltSnapshotDeploymentRepository=repo.spring.io::default::https://repo.spring.io/libs-snapshot-local -P spring +-DaltSnapshotDeploymentRepository=repo.spring.io::default::https://repo.spring.io/snapshot -P spring diff --git a/.settings.xml b/.settings.xml index 03645e8..a50de34 100644 --- a/.settings.xml +++ b/.settings.xml @@ -23,7 +23,7 @@ spring-snapshots Spring Snapshots - https://repo.spring.io/libs-snapshot-local + https://repo.spring.io/snapshot true @@ -31,7 +31,7 @@ spring-milestones Spring Milestones - https://repo.spring.io/libs-milestone-local + https://repo.spring.io/milestone false @@ -49,7 +49,7 @@ spring-snapshots Spring Snapshots - https://repo.spring.io/libs-snapshot-local + https://repo.spring.io/snapshot true @@ -57,7 +57,7 @@ spring-milestones Spring Milestones - https://repo.spring.io/libs-milestone-local + https://repo.spring.io/milestone false diff --git a/pom.xml b/pom.xml index d969548..e129267 100644 --- a/pom.xml +++ b/pom.xml @@ -21,15 +21,18 @@ spring-cloud-bus-dependencies spring-cloud-bus + spring-cloud-bus-rsocket spring-cloud-bus-tests spring-cloud-starter-bus-amqp spring-cloud-starter-bus-kafka + spring-cloud-starter-bus-stream docs 3.0.0-SNAPSHOT 3.1.0-SNAPSHOT + 3.1.0-SNAPSHOT bus @@ -75,7 +78,7 @@ spring-snapshots Spring Snapshots - https://repo.spring.io/libs-snapshot-local + https://repo.spring.io/snapshot true @@ -86,7 +89,7 @@ spring-milestones Spring Milestones - https://repo.spring.io/libs-milestone-local + https://repo.spring.io/milestone false @@ -104,7 +107,7 @@ spring-snapshots Spring Snapshots - https://repo.spring.io/libs-snapshot-local + https://repo.spring.io/snapshot true @@ -115,7 +118,7 @@ spring-milestones Spring Milestones - https://repo.spring.io/libs-milestone-local + https://repo.spring.io/milestone false @@ -123,7 +126,7 @@ spring-releases Spring Releases - https://repo.spring.io/libs-release-local + https://repo.spring.io/release false @@ -159,6 +162,18 @@ pom import + + org.springframework.cloud + spring-cloud-function-dependencies + ${spring-cloud-function.version} + pom + import + + + org.springframework.cloud + spring-cloud-function-rsocket + ${spring-cloud-function.version} + diff --git a/spring-cloud-bus-dependencies/pom.xml b/spring-cloud-bus-dependencies/pom.xml index 7cd28f6..f466cf2 100644 --- a/spring-cloud-bus-dependencies/pom.xml +++ b/spring-cloud-bus-dependencies/pom.xml @@ -26,11 +26,21 @@ spring-cloud-starter-bus-kafka ${project.version} + + org.springframework.cloud + spring-cloud-starter-bus-stream + ${project.version} + org.springframework.cloud spring-cloud-bus ${project.version} + + org.springframework.cloud + spring-cloud-bus-rsocket + ${project.version} + @@ -40,7 +50,7 @@ spring-snapshots Spring Snapshots - https://repo.spring.io/libs-snapshot-local + https://repo.spring.io/snapshot true @@ -51,7 +61,7 @@ spring-milestones Spring Milestones - https://repo.spring.io/libs-milestone-local + https://repo.spring.io/milestone false @@ -69,7 +79,7 @@ spring-snapshots Spring Snapshots - https://repo.spring.io/libs-snapshot-local + https://repo.spring.io/snapshot true @@ -80,7 +90,7 @@ spring-milestones Spring Milestones - https://repo.spring.io/libs-milestone-local + https://repo.spring.io/milestone false diff --git a/spring-cloud-bus-rsocket/pom.xml b/spring-cloud-bus-rsocket/pom.xml new file mode 100644 index 0000000..04a66ef --- /dev/null +++ b/spring-cloud-bus-rsocket/pom.xml @@ -0,0 +1,75 @@ + + + 4.0.0 + + spring-cloud-bus-rsocket + jar + + spring-cloud-bus-rsocket + Spring Cloud Bus + + + org.springframework.cloud + spring-cloud-bus-parent + 3.0.0-SNAPSHOT + .. + + + + 0.2.0-SNAPSHOT + + + + + org.springframework.boot + spring-boot-configuration-processor + true + + + org.springframework.boot + spring-boot-starter-actuator + true + + + org.springframework.cloud + spring-cloud-bus + + + org.springframework.cloud + spring-cloud-function-rsocket + + + com.fasterxml.jackson.dataformat + jackson-dataformat-cbor + + + io.rsocket.routing + rsocket-routing-client-spring + ${rsocket-routing.version} + true + + + org.springframework.boot + spring-boot-autoconfigure-processor + true + + + org.springframework.boot + spring-boot-starter-test + test + + + org.junit.vintage + junit-vintage-engine + test + + + org.springframework.cloud + spring-cloud-test-support + test + + + + diff --git a/spring-cloud-bus-rsocket/src/main/java/org/springframework/cloud/bus/rsocket/BusRSocketAutoConfiguration.java b/spring-cloud-bus-rsocket/src/main/java/org/springframework/cloud/bus/rsocket/BusRSocketAutoConfiguration.java new file mode 100644 index 0000000..264173a --- /dev/null +++ b/spring-cloud-bus-rsocket/src/main/java/org/springframework/cloud/bus/rsocket/BusRSocketAutoConfiguration.java @@ -0,0 +1,61 @@ +/* + * 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.rsocket; + +import io.rsocket.RSocket; +import io.rsocket.routing.client.spring.RoutingClientProperties; +import io.rsocket.routing.client.spring.RoutingRSocketRequester; + +import org.springframework.boot.autoconfigure.AutoConfigureBefore; +import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.cloud.bus.BusAutoConfiguration; +import org.springframework.cloud.bus.BusProperties; +import org.springframework.cloud.bus.BusRefreshAutoConfiguration; +import org.springframework.cloud.bus.ConditionalOnBusEnabled; +import org.springframework.cloud.bus.PathServiceMatcherAutoConfiguration; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +/** + * @author Spencer Gibb + */ +@Configuration(proxyBeanMethods = false) +@ConditionalOnBusEnabled +@EnableConfigurationProperties(BusRSocketProperties.class) +@ConditionalOnClass({ RSocket.class, RoutingRSocketRequester.class }) +@AutoConfigureBefore({ BusAutoConfiguration.class, BusRefreshAutoConfiguration.class, + PathServiceMatcherAutoConfiguration.class }) +public class BusRSocketAutoConfiguration { + + @Bean + public RoutingClientDestinationFactory routingClientDestinationFactory(BusRSocketProperties properties) { + return new RoutingClientDestinationFactory(properties); + } + + @Bean + public RSocketRequesterBusBridge rSocketRequesterBusBridge(RoutingRSocketRequester requester) { + return new RSocketRequesterBusBridge(requester); + } + + @Bean + public RSocketServiceMatcher rSocketServiceMatcher(BusProperties properties, + RoutingClientProperties routingClientProperties) { + return new RSocketServiceMatcher(properties.getId(), routingClientProperties); + } + +} diff --git a/spring-cloud-bus-rsocket/src/main/java/org/springframework/cloud/bus/rsocket/BusRSocketEnvironmentPostProcessor.java b/spring-cloud-bus-rsocket/src/main/java/org/springframework/cloud/bus/rsocket/BusRSocketEnvironmentPostProcessor.java new file mode 100644 index 0000000..51af6f8 --- /dev/null +++ b/spring-cloud-bus-rsocket/src/main/java/org/springframework/cloud/bus/rsocket/BusRSocketEnvironmentPostProcessor.java @@ -0,0 +1,45 @@ +/* + * Copyright 2012-2019 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.bus.rsocket; + +import java.util.HashMap; +import java.util.Map; + +import io.rsocket.routing.client.spring.RoutingClientProperties; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.env.EnvironmentPostProcessor; +import org.springframework.core.env.ConfigurableEnvironment; + +import static org.springframework.cloud.bus.BusEnvironmentPostProcessor.addOrReplace; + +/** + * {@link EnvironmentPostProcessor} that sets the default properties for the RSocket + * Routing Client. + */ +public class BusRSocketEnvironmentPostProcessor implements EnvironmentPostProcessor { + + static final String DEFAULTS_PROPERTY_SOURCE_NAME = "springCloudBusRSocketDefaultProperties"; + + @Override + public void postProcessEnvironment(ConfigurableEnvironment environment, SpringApplication application) { + Map defaults = new HashMap<>(); + defaults.put(RoutingClientProperties.CONFIG_PREFIX + ".tags.bus", true); + addOrReplace(environment.getPropertySources(), defaults, DEFAULTS_PROPERTY_SOURCE_NAME, false); + } + +} diff --git a/spring-cloud-bus-rsocket/src/main/java/org/springframework/cloud/bus/rsocket/BusRSocketProperties.java b/spring-cloud-bus-rsocket/src/main/java/org/springframework/cloud/bus/rsocket/BusRSocketProperties.java new file mode 100644 index 0000000..b183430 --- /dev/null +++ b/spring-cloud-bus-rsocket/src/main/java/org/springframework/cloud/bus/rsocket/BusRSocketProperties.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.rsocket; + +import java.util.LinkedHashMap; +import java.util.Map; + +import io.rsocket.routing.common.MutableKey; + +import org.springframework.beans.factory.InitializingBean; +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.core.style.ToStringCreator; +import org.springframework.util.CollectionUtils; + +@ConfigurationProperties("spring.cloud.bus.rsocket") +public class BusRSocketProperties implements InitializingBean { + + private final Map defaultTags = new LinkedHashMap<>(); + + @Override + public void afterPropertiesSet() { + if (CollectionUtils.isEmpty(defaultTags)) { + defaultTags.put(MutableKey.of("bus"), "true"); + } + } + + public Map getDefaultTags() { + return this.defaultTags; + } + + @Override + public String toString() { + return new ToStringCreator(this).append("defaultTags", defaultTags).toString(); + } + +} diff --git a/spring-cloud-bus-rsocket/src/main/java/org/springframework/cloud/bus/rsocket/RSocketRequesterBusBridge.java b/spring-cloud-bus-rsocket/src/main/java/org/springframework/cloud/bus/rsocket/RSocketRequesterBusBridge.java new file mode 100644 index 0000000..e9404d0 --- /dev/null +++ b/spring-cloud-bus-rsocket/src/main/java/org/springframework/cloud/bus/rsocket/RSocketRequesterBusBridge.java @@ -0,0 +1,87 @@ +/* + * 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.rsocket; + +import java.util.HashMap; +import java.util.Map; + +import io.rsocket.routing.client.spring.RoutingRSocketRequester; +import io.rsocket.routing.common.Key; +import io.rsocket.routing.common.WellKnownKey; +import io.rsocket.routing.frames.RoutingType; + +import org.springframework.cloud.bus.BusBridge; +import org.springframework.cloud.bus.BusConstants; +import org.springframework.cloud.bus.event.RemoteApplicationEvent; +import org.springframework.util.StringUtils; + +public class RSocketRequesterBusBridge implements BusBridge { + + public RSocketRequesterBusBridge(RoutingRSocketRequester requester) { + this.requester = requester; + } + + private final RoutingRSocketRequester requester; + + @Override + public void send(RemoteApplicationEvent event) { + requester.route(BusConstants.BUS_CONSUMER).address(builder -> { + builder.routingType(RoutingType.MULTICAST); + // get tags out of destination + getTagsFromDestination(event.getDestinationService()).forEach(builder::with); + }).data(event).send().subscribe(); + } + + static Map getTagsFromDestination(String delimitedProperties) { + String[] properties = StringUtils.tokenizeToStringArray(delimitedProperties, ":"); + Map map = new HashMap<>(); + for (String property : properties) { + int index = lowestIndexOf(property, "="); + String key = (index > 0) ? property.substring(0, index) : property; + String value = (index > 0) ? property.substring(index + 1) : null; + + try { + WellKnownKey wellKnownKey = WellKnownKey.valueOf(key); + map.put(Key.of(wellKnownKey), value); + } + catch (IllegalArgumentException e) { + try { + WellKnownKey wellKnownKey = WellKnownKey.valueOf(key.toUpperCase()); + map.put(Key.of(wellKnownKey), value); + } + catch (IllegalArgumentException e2) { + // not a WellKnownKey, use string + map.put(Key.of(key), value); + } + } + + } + return map; + } + + private static int lowestIndexOf(String property, String... candidates) { + int index = -1; + for (String candidate : candidates) { + int candidateIndex = property.indexOf(candidate); + if (candidateIndex > 0) { + index = (index != -1) ? Math.min(index, candidateIndex) : candidateIndex; + } + } + return index; + } + +} diff --git a/spring-cloud-bus-rsocket/src/main/java/org/springframework/cloud/bus/rsocket/RSocketServiceMatcher.java b/spring-cloud-bus-rsocket/src/main/java/org/springframework/cloud/bus/rsocket/RSocketServiceMatcher.java new file mode 100644 index 0000000..8944f37 --- /dev/null +++ b/spring-cloud-bus-rsocket/src/main/java/org/springframework/cloud/bus/rsocket/RSocketServiceMatcher.java @@ -0,0 +1,88 @@ +/* + * 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.rsocket; + +import java.util.HashMap; +import java.util.Map; + +import io.rsocket.routing.client.spring.RoutingClientProperties; +import io.rsocket.routing.common.Key; +import io.rsocket.routing.common.WellKnownKey; + +import org.springframework.cloud.bus.ServiceMatcher; +import org.springframework.cloud.bus.event.RemoteApplicationEvent; +import org.springframework.util.AntPathMatcher; + +import static org.springframework.cloud.bus.rsocket.RSocketRequesterBusBridge.getTagsFromDestination; + +/** + * A pass thru patcher that allows the RSocket Routing broker to determine which instances + * to send to. + */ +public class RSocketServiceMatcher implements ServiceMatcher { + + private final String busId; + + private final RoutingClientProperties properties; + + private final AntPathMatcher antPathMatcher = new AntPathMatcher(); + + private final Map localTags = new HashMap<>(); + + public RSocketServiceMatcher(String busId, RoutingClientProperties properties) { + this.busId = busId; + this.properties = properties; + convertLocalTags(properties); + } + + @Override + public boolean isFromSelf(RemoteApplicationEvent event) { + String originService = event.getOriginService(); + String serviceId = getBusId(); + return antPathMatcher.match(originService, serviceId); + } + + @Override + public boolean isForSelf(RemoteApplicationEvent event) { + Map tags = getTagsFromDestination(event.getDestinationService()); + for (Map.Entry entry : tags.entrySet()) { + String existingValue = localTags.get(entry.getKey()); + if (existingValue == null || !existingValue.equals(entry.getValue())) { + return false; + } + } + return true; + } + + private void convertLocalTags(RoutingClientProperties properties) { + properties.getTags().forEach((key, value) -> { + if (key.getWellKnownKey() != null) { + localTags.put(Key.of(key.getWellKnownKey()), value); + } + else { + localTags.put(Key.of(key.getKey()), value); + } + }); + localTags.put(Key.of(WellKnownKey.SERVICE_NAME), properties.getServiceName()); + } + + @Override + public String getBusId() { + return busId; + } + +} diff --git a/spring-cloud-bus-rsocket/src/main/java/org/springframework/cloud/bus/rsocket/RoutingClientDestinationFactory.java b/spring-cloud-bus-rsocket/src/main/java/org/springframework/cloud/bus/rsocket/RoutingClientDestinationFactory.java new file mode 100644 index 0000000..b00cee0 --- /dev/null +++ b/spring-cloud-bus-rsocket/src/main/java/org/springframework/cloud/bus/rsocket/RoutingClientDestinationFactory.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.rsocket; + +import java.util.ArrayList; + +import org.springframework.cloud.bus.event.Destination; +import org.springframework.util.StringUtils; + +public class RoutingClientDestinationFactory implements Destination.Factory { + + private final BusRSocketProperties properties; + + public RoutingClientDestinationFactory(BusRSocketProperties properties) { + this.properties = properties; + } + + @Override + public Destination getDestination(String originalDestination) { + ArrayList entries = new ArrayList<>(); + properties.getDefaultTags().forEach((key, s) -> { + String keyStr = (key.getWellKnownKey() != null) ? key.getWellKnownKey().name() : key.getKey(); + entries.add(keyStr + "=" + s); + }); + String defaultTags = StringUtils.collectionToDelimitedString(entries, ":"); + return () -> { + String destination = (StringUtils.isEmpty(originalDestination)) ? defaultTags + : defaultTags + ":" + originalDestination; + if (StringUtils.isEmpty(destination)) { + throw new IllegalArgumentException("destination may not be empty"); + } + return destination; + }; + } + +} diff --git a/spring-cloud-bus-rsocket/src/main/resources/META-INF/spring.factories b/spring-cloud-bus-rsocket/src/main/resources/META-INF/spring.factories new file mode 100644 index 0000000..8b84263 --- /dev/null +++ b/spring-cloud-bus-rsocket/src/main/resources/META-INF/spring.factories @@ -0,0 +1,6 @@ +org.springframework.boot.autoconfigure.EnableAutoConfiguration=\ +org.springframework.cloud.bus.rsocket.BusRSocketAutoConfiguration + +# Environment Post Processor +org.springframework.boot.env.EnvironmentPostProcessor=\ +org.springframework.cloud.bus.rsocket.BusRSocketEnvironmentPostProcessor diff --git a/spring-cloud-bus-tests/src/test/java/org/springframework/cloud/bus/jackson/BusJacksonIntegrationTests.java b/spring-cloud-bus-tests/src/test/java/org/springframework/cloud/bus/jackson/BusJacksonIntegrationTests.java index 1304c02..ea8d911 100644 --- a/spring-cloud-bus-tests/src/test/java/org/springframework/cloud/bus/jackson/BusJacksonIntegrationTests.java +++ b/spring-cloud-bus-tests/src/test/java/org/springframework/cloud/bus/jackson/BusJacksonIntegrationTests.java @@ -130,7 +130,7 @@ public class BusJacksonIntegrationTests { @PutMapping("/names/{name}") public void sayName(@PathVariable String name) { this.names.add(name); - this.publisher.publishEvent(new NameEvent(this, this.busServiceMatcher.getServiceId(), name)); + this.publisher.publishEvent(new NameEvent(this, this.busServiceMatcher.getBusId(), name)); } @GetMapping("/date") diff --git a/spring-cloud-bus/pom.xml b/spring-cloud-bus/pom.xml index 8c11e3b..a8b2710 100644 --- a/spring-cloud-bus/pom.xml +++ b/spring-cloud-bus/pom.xml @@ -40,11 +40,17 @@ org.springframework.cloud spring-cloud-stream + true org.springframework.integration spring-integration-core + + com.fasterxml.jackson.dataformat + jackson-dataformat-cbor + true + org.springframework.boot spring-boot-autoconfigure-processor 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 e80eaaf..a5e9258 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,24 +16,21 @@ package org.springframework.cloud.bus; +import org.springframework.beans.factory.ObjectProvider; 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; -import org.springframework.boot.autoconfigure.AutoConfigureAfter; -import org.springframework.boot.autoconfigure.AutoConfigureBefore; import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.boot.context.properties.EnableConfigurationProperties; -import org.springframework.cloud.autoconfigure.LifecycleMvcEndpointAutoConfiguration; import org.springframework.cloud.bus.endpoint.EnvironmentBusEndpoint; +import org.springframework.cloud.bus.event.Destination; import org.springframework.cloud.bus.event.EnvironmentChangeListener; +import org.springframework.cloud.bus.event.PathDestinationFactory; import org.springframework.cloud.bus.event.TraceListener; import org.springframework.cloud.context.environment.EnvironmentManager; -import org.springframework.cloud.stream.config.BindingServiceConfiguration; -import org.springframework.cloud.stream.function.StreamBridge; -import org.springframework.context.ApplicationContext; import org.springframework.context.ApplicationEventPublisher; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @@ -47,16 +44,12 @@ import static org.springframework.cloud.bus.BusConstants.BUS_CONSUMER; @Configuration(proxyBeanMethods = false) @ConditionalOnBusEnabled @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 { @Bean - @ConditionalOnMissingBean(BusBridge.class) - public StreamBusBridge streamBusBridge(StreamBridge streamBridge, BusProperties properties) { - return new StreamBusBridge(streamBridge, properties); + @ConditionalOnMissingBean(Destination.Factory.class) + public PathDestinationFactory pathDestinationFactory() { + return new PathDestinationFactory(); } @Bean @@ -69,8 +62,8 @@ public class BusAutoConfiguration { @Bean @ConditionalOnMissingBean(name = BUS_CONSUMER) public BusConsumer busConsumer(ApplicationEventPublisher applicationEventPublisher, ServiceMatcher serviceMatcher, - BusBridge busBridge, BusProperties properties) { - return new BusConsumer(applicationEventPublisher, serviceMatcher, busBridge, properties); + ObjectProvider busBridge, BusProperties properties, Destination.Factory destinationFactory) { + return new BusConsumer(applicationEventPublisher, serviceMatcher, busBridge, properties, destinationFactory); } @Configuration(proxyBeanMethods = false) @@ -104,8 +97,9 @@ public class BusAutoConfiguration { @Bean @ConditionalOnAvailableEndpoint - public EnvironmentBusEndpoint environmentBusEndpoint(ApplicationContext context, BusProperties bus) { - return new EnvironmentBusEndpoint(context, bus.getId()); + public EnvironmentBusEndpoint environmentBusEndpoint(ApplicationEventPublisher publisher, BusProperties bus, + Destination.Factory destinationFactory) { + return new EnvironmentBusEndpoint(publisher, bus.getId(), destinationFactory); } } diff --git a/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/BusConstants.java b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/BusConstants.java index 3f997e1..d0d9150 100644 --- a/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/BusConstants.java +++ b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/BusConstants.java @@ -20,7 +20,7 @@ package org.springframework.cloud.bus; * @author Dave Syer * */ -abstract class BusConstants { +public abstract class BusConstants { /** * Name of the input channel for Spring Cloud Bus. 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 index 3cd9022..dccf274 100644 --- 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 @@ -21,7 +21,9 @@ import java.util.function.Consumer; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.springframework.beans.factory.ObjectProvider; import org.springframework.cloud.bus.event.AckRemoteApplicationEvent; +import org.springframework.cloud.bus.event.Destination; import org.springframework.cloud.bus.event.RemoteApplicationEvent; import org.springframework.cloud.bus.event.SentApplicationEvent; import org.springframework.context.ApplicationEventPublisher; @@ -34,16 +36,19 @@ public class BusConsumer implements Consumer { private final ServiceMatcher serviceMatcher; - private final BusBridge busBridge; + private final ObjectProvider busBridge; private final BusProperties properties; - public BusConsumer(ApplicationEventPublisher publisher, ServiceMatcher serviceMatcher, BusBridge busBridge, - BusProperties properties) { + private final Destination.Factory destinationFactory; + + public BusConsumer(ApplicationEventPublisher publisher, ServiceMatcher serviceMatcher, + ObjectProvider busBridge, BusProperties properties, Destination.Factory destinationFactory) { this.publisher = publisher; this.serviceMatcher = serviceMatcher; this.busBridge = busBridge; this.properties = properties; + this.destinationFactory = destinationFactory; } @Override @@ -66,10 +71,10 @@ public class BusConsumer implements Consumer { 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); + AckRemoteApplicationEvent ack = new AckRemoteApplicationEvent(this, this.serviceMatcher.getBusId(), + destinationFactory.getDestination(this.properties.getAck().getDestinationService()), + event.getDestinationService(), event.getId(), event.getClass()); + this.busBridge.ifAvailable(bridge -> bridge.send(ack)); this.publisher.publishEvent(ack); } } 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 8b21982..d65f168 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 @@ -76,7 +76,7 @@ public class BusEnvironmentPostProcessor implements EnvironmentPostProcessor { addOrReplace(environment.getPropertySources(), defaults, DEFAULTS_PROPERTY_SOURCE_NAME, false); } - private void addOrReplace(MutablePropertySources propertySources, Map map, + public static void addOrReplace(MutablePropertySources propertySources, Map map, String propertySourceName, boolean first) { MapPropertySource target = null; if (propertySources.contains(propertySourceName)) { diff --git a/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/BusRefreshAutoConfiguration.java b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/BusRefreshAutoConfiguration.java index 7f2b4a2..de936dc 100644 --- a/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/BusRefreshAutoConfiguration.java +++ b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/BusRefreshAutoConfiguration.java @@ -22,9 +22,10 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.cloud.bus.endpoint.RefreshBusEndpoint; +import org.springframework.cloud.bus.event.Destination; import org.springframework.cloud.bus.event.RefreshListener; import org.springframework.cloud.context.refresh.ContextRefresher; -import org.springframework.context.ApplicationContext; +import org.springframework.context.ApplicationEventPublisher; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @@ -50,8 +51,9 @@ public class BusRefreshAutoConfiguration { @Bean @ConditionalOnAvailableEndpoint - public RefreshBusEndpoint refreshBusEndpoint(ApplicationContext context, BusProperties bus) { - return new RefreshBusEndpoint(context, bus.getId()); + public RefreshBusEndpoint refreshBusEndpoint(ApplicationEventPublisher publisher, BusProperties bus, + Destination.Factory destinationFactory) { + return new RefreshBusEndpoint(publisher, bus.getId(), destinationFactory); } } diff --git a/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/BusStreamAutoConfiguration.java b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/BusStreamAutoConfiguration.java new file mode 100644 index 0000000..19ada03 --- /dev/null +++ b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/BusStreamAutoConfiguration.java @@ -0,0 +1,45 @@ +/* + * 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.boot.autoconfigure.AutoConfigureAfter; +import org.springframework.boot.autoconfigure.AutoConfigureBefore; +import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; +import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.cloud.autoconfigure.LifecycleMvcEndpointAutoConfiguration; +import org.springframework.cloud.stream.config.BindingServiceConfiguration; +import org.springframework.cloud.stream.function.StreamBridge; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +@Configuration(proxyBeanMethods = false) +@ConditionalOnBusEnabled +@ConditionalOnClass({ StreamBridge.class, BindingServiceConfiguration.class }) +@EnableConfigurationProperties(BusProperties.class) +@AutoConfigureBefore({ BindingServiceConfiguration.class, BusAutoConfiguration.class }) +// so stream bindings work properly +@AutoConfigureAfter({ LifecycleMvcEndpointAutoConfiguration.class, PathServiceMatcherAutoConfiguration.class }) +public class BusStreamAutoConfiguration { + + @Bean + @ConditionalOnMissingBean(BusBridge.class) + public StreamBusBridge streamBusBridge(StreamBridge streamBridge, BusProperties properties) { + return new StreamBusBridge(streamBridge, properties); + } + +} diff --git a/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/PathServiceMatcher.java b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/PathServiceMatcher.java new file mode 100644 index 0000000..b0e2534 --- /dev/null +++ b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/PathServiceMatcher.java @@ -0,0 +1,79 @@ +/* + * Copyright 2012-2019 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.bus; + +import org.springframework.cloud.bus.event.RemoteApplicationEvent; +import org.springframework.util.PathMatcher; + +/** + * @author Spencer Gibb + */ +public class PathServiceMatcher implements ServiceMatcher { + + private final PathMatcher matcher; + + private final String id; + + private String[] configNames = new String[] {}; + + public PathServiceMatcher(PathMatcher matcher, String id) { + this.matcher = matcher; + this.id = id; + } + + public PathServiceMatcher(PathMatcher matcher, String id, String[] configNames) { + this(matcher, id); + + int colonIndex = id.indexOf(":"); + if (colonIndex >= 0) { + // if the id contains profiles and port, append them to the config names + String profilesAndPort = id.substring(colonIndex); + for (int i = 0; i < configNames.length; i++) { + configNames[i] = configNames[i] + profilesAndPort; + } + } + this.configNames = configNames; + } + + public boolean isFromSelf(RemoteApplicationEvent event) { + String originService = event.getOriginService(); + String serviceId = getBusId(); + return this.matcher.match(originService, serviceId); + } + + public boolean isForSelf(RemoteApplicationEvent event) { + String destinationService = event.getDestinationService(); + if (destinationService == null || destinationService.trim().isEmpty() + || this.matcher.match(destinationService, getBusId())) { + return true; + } + + // Check all potential config names instead of service name + for (String configName : this.configNames) { + if (this.matcher.match(destinationService, configName)) { + return true; + } + } + + return false; + } + + public String getBusId() { + return this.id; + } + +} 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/PathServiceMatcherAutoConfiguration.java similarity index 86% rename from spring-cloud-bus/src/main/java/org/springframework/cloud/bus/ServiceMatcherAutoConfiguration.java rename to spring-cloud-bus/src/main/java/org/springframework/cloud/bus/PathServiceMatcherAutoConfiguration.java index fe090e0..fa745b8 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/PathServiceMatcherAutoConfiguration.java @@ -30,7 +30,7 @@ import org.springframework.util.PathMatcher; @Configuration(proxyBeanMethods = false) @ConditionalOnBusEnabled @EnableConfigurationProperties(BusProperties.class) -public class ServiceMatcherAutoConfiguration { +public class PathServiceMatcherAutoConfiguration { /** * Name of the Bus path matcher. @@ -51,10 +51,11 @@ public class ServiceMatcherAutoConfiguration { } @Bean - public ServiceMatcher serviceMatcher(@BusPathMatcher PathMatcher pathMatcher, BusProperties properties, + @ConditionalOnMissingBean(ServiceMatcher.class) + public PathServiceMatcher pathServiceMatcher(@BusPathMatcher PathMatcher pathMatcher, BusProperties properties, Environment environment) { String[] configNames = environment.getProperty(CLOUD_CONFIG_NAME_PROPERTY, String[].class, new String[] {}); - return new ServiceMatcher(pathMatcher, properties.getId(), configNames); + return new PathServiceMatcher(pathMatcher, properties.getId(), configNames); } } diff --git a/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/ServiceMatcher.java b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/ServiceMatcher.java index 8725c59..e7d4c91 100644 --- a/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/ServiceMatcher.java +++ b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/ServiceMatcher.java @@ -17,63 +17,16 @@ package org.springframework.cloud.bus; import org.springframework.cloud.bus.event.RemoteApplicationEvent; -import org.springframework.util.PathMatcher; /** * @author Spencer Gibb */ -public class ServiceMatcher { +public interface ServiceMatcher { - private final PathMatcher matcher; + boolean isFromSelf(RemoteApplicationEvent event); - private final String id; + boolean isForSelf(RemoteApplicationEvent event); - private String[] configNames = new String[] {}; - - public ServiceMatcher(PathMatcher matcher, String id) { - this.matcher = matcher; - this.id = id; - } - - public ServiceMatcher(PathMatcher matcher, String id, String[] configNames) { - this(matcher, id); - - int colonIndex = id.indexOf(":"); - if (colonIndex >= 0) { - // if the id contains profiles and port, append them to the config names - String profilesAndPort = id.substring(colonIndex); - for (int i = 0; i < configNames.length; i++) { - configNames[i] = configNames[i] + profilesAndPort; - } - } - this.configNames = configNames; - } - - public boolean isFromSelf(RemoteApplicationEvent event) { - String originService = event.getOriginService(); - String serviceId = getServiceId(); - return this.matcher.match(originService, serviceId); - } - - public boolean isForSelf(RemoteApplicationEvent event) { - String destinationService = event.getDestinationService(); - if (destinationService == null || destinationService.trim().isEmpty() - || this.matcher.match(destinationService, getServiceId())) { - return true; - } - - // Check all potential config names instead of service name - for (String configName : this.configNames) { - if (this.matcher.match(destinationService, configName)) { - return true; - } - } - - return false; - } - - public String getServiceId() { - return this.id; - } + String getBusId(); } diff --git a/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/endpoint/AbstractBusEndpoint.java b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/endpoint/AbstractBusEndpoint.java index 728b68e..c48ab78 100644 --- a/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/endpoint/AbstractBusEndpoint.java +++ b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/endpoint/AbstractBusEndpoint.java @@ -16,6 +16,7 @@ package org.springframework.cloud.bus.endpoint; +import org.springframework.cloud.bus.event.Destination; import org.springframework.context.ApplicationEvent; import org.springframework.context.ApplicationEventPublisher; @@ -24,21 +25,33 @@ import org.springframework.context.ApplicationEventPublisher; */ public class AbstractBusEndpoint { - private ApplicationEventPublisher context; + private ApplicationEventPublisher publisher; private String appId; - public AbstractBusEndpoint(ApplicationEventPublisher context, String appId) { - this.context = context; + private final Destination.Factory destinationFactory; + + public AbstractBusEndpoint(ApplicationEventPublisher publisher, String appId, + Destination.Factory destinationFactory) { + this.publisher = publisher; this.appId = appId; + this.destinationFactory = destinationFactory; } protected String getInstanceId() { return this.appId; } + protected Destination.Factory getDestinationFactory() { + return this.destinationFactory; + } + + protected Destination getDestination(String original) { + return destinationFactory.getDestination(original); + } + protected void publish(ApplicationEvent event) { - this.context.publishEvent(event); + this.publisher.publishEvent(event); } } diff --git a/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/endpoint/EnvironmentBusEndpoint.java b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/endpoint/EnvironmentBusEndpoint.java index 1c89242..b2f93e3 100644 --- a/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/endpoint/EnvironmentBusEndpoint.java +++ b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/endpoint/EnvironmentBusEndpoint.java @@ -21,9 +21,12 @@ import java.util.Map; import org.springframework.boot.actuate.endpoint.annotation.Endpoint; import org.springframework.boot.actuate.endpoint.annotation.Selector; +import org.springframework.boot.actuate.endpoint.annotation.Selector.Match; import org.springframework.boot.actuate.endpoint.annotation.WriteOperation; +import org.springframework.cloud.bus.event.Destination; import org.springframework.cloud.bus.event.EnvironmentChangeRemoteApplicationEvent; import org.springframework.context.ApplicationEventPublisher; +import org.springframework.util.StringUtils; /** * @author Spencer Gibb @@ -31,21 +34,25 @@ import org.springframework.context.ApplicationEventPublisher; @Endpoint(id = "busenv") // TODO: document public class EnvironmentBusEndpoint extends AbstractBusEndpoint { - public EnvironmentBusEndpoint(ApplicationEventPublisher context, String id) { - super(context, id); + public EnvironmentBusEndpoint(ApplicationEventPublisher publisher, String id, + Destination.Factory destinationFactory) { + super(publisher, id, destinationFactory); } @WriteOperation // TODO: document params - public void busEnvWithDestination(String name, String value, @Selector String destination) { + public void busEnvWithDestination(String name, String value, + @Selector(match = Match.ALL_REMAINING) String[] destinations) { Map params = Collections.singletonMap(name, value); - publish(new EnvironmentChangeRemoteApplicationEvent(this, getInstanceId(), destination, params)); + String destination = StringUtils.arrayToDelimitedString(destinations, ":"); + publish(new EnvironmentChangeRemoteApplicationEvent(this, getInstanceId(), getDestination(destination), + params)); } @WriteOperation public void busEnv(String name, String value) { // TODO: document params Map params = Collections.singletonMap(name, value); - publish(new EnvironmentChangeRemoteApplicationEvent(this, getInstanceId(), null, params)); + publish(new EnvironmentChangeRemoteApplicationEvent(this, getInstanceId(), getDestination(null), params)); } } diff --git a/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/endpoint/RefreshBusEndpoint.java b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/endpoint/RefreshBusEndpoint.java index 7d10064..31b2297 100644 --- a/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/endpoint/RefreshBusEndpoint.java +++ b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/endpoint/RefreshBusEndpoint.java @@ -18,9 +18,12 @@ package org.springframework.cloud.bus.endpoint; import org.springframework.boot.actuate.endpoint.annotation.Endpoint; import org.springframework.boot.actuate.endpoint.annotation.Selector; +import org.springframework.boot.actuate.endpoint.annotation.Selector.Match; import org.springframework.boot.actuate.endpoint.annotation.WriteOperation; +import org.springframework.cloud.bus.event.Destination; import org.springframework.cloud.bus.event.RefreshRemoteApplicationEvent; import org.springframework.context.ApplicationEventPublisher; +import org.springframework.util.StringUtils; /** * @author Spencer Gibb @@ -28,18 +31,19 @@ import org.springframework.context.ApplicationEventPublisher; @Endpoint(id = "busrefresh") // TODO: document new id public class RefreshBusEndpoint extends AbstractBusEndpoint { - public RefreshBusEndpoint(ApplicationEventPublisher context, String id) { - super(context, id); + public RefreshBusEndpoint(ApplicationEventPublisher publisher, String id, Destination.Factory destinationFactory) { + super(publisher, id, destinationFactory); } @WriteOperation - public void busRefreshWithDestination(@Selector String destination) { - publish(new RefreshRemoteApplicationEvent(this, getInstanceId(), destination)); + public void busRefreshWithDestination(@Selector(match = Match.ALL_REMAINING) String[] destinations) { + String destination = StringUtils.arrayToDelimitedString(destinations, ":"); + publish(new RefreshRemoteApplicationEvent(this, getInstanceId(), getDestination(destination))); } @WriteOperation public void busRefresh() { - publish(new RefreshRemoteApplicationEvent(this, getInstanceId(), null)); + publish(new RefreshRemoteApplicationEvent(this, getInstanceId(), getDestination(null))); } } diff --git a/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/event/AckRemoteApplicationEvent.java b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/event/AckRemoteApplicationEvent.java index c7a57c6..61c3c29 100644 --- a/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/event/AckRemoteApplicationEvent.java +++ b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/event/AckRemoteApplicationEvent.java @@ -44,9 +44,9 @@ public class AckRemoteApplicationEvent extends RemoteApplicationEvent { this.event = null; } - public AckRemoteApplicationEvent(Object source, String originService, String destinationService, + public AckRemoteApplicationEvent(Object source, String originService, Destination destination, String ackDestinationService, String ackId, Class type) { - super(source, originService, destinationService); + super(source, originService, destination); this.ackDestinationService = ackDestinationService; this.ackId = ackId; this.event = type; @@ -71,6 +71,7 @@ public class AckRemoteApplicationEvent extends RemoteApplicationEvent { * @param eventName the fq class name of the event implementation, not null */ @JsonProperty("event") + @SuppressWarnings("unchecked") public void setEventName(String eventName) { try { this.event = (Class) Class.forName(eventName); diff --git a/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/BusPropertiesAutoConfiguration.java b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/event/Destination.java similarity index 57% rename from spring-cloud-bus/src/main/java/org/springframework/cloud/bus/BusPropertiesAutoConfiguration.java rename to spring-cloud-bus/src/main/java/org/springframework/cloud/bus/event/Destination.java index eb226e6..efd0595 100644 --- a/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/BusPropertiesAutoConfiguration.java +++ b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/event/Destination.java @@ -1,5 +1,5 @@ /* - * Copyright 2012-2019 the original author or authors. + * 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. @@ -14,17 +14,17 @@ * limitations under the License. */ -package org.springframework.cloud.bus; +package org.springframework.cloud.bus.event; -import org.springframework.boot.context.properties.EnableConfigurationProperties; -import org.springframework.context.annotation.Configuration; +public interface Destination { -/** - * @author Spencer Gibb - * @author Dave Syer - */ -@Configuration(proxyBeanMethods = false) -@EnableConfigurationProperties(BusProperties.class) -public class BusPropertiesAutoConfiguration { + String getDestinationAsString(); + + @FunctionalInterface + interface Factory { + + Destination getDestination(String originalDestination); + + } } diff --git a/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/event/EnvironmentChangeRemoteApplicationEvent.java b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/event/EnvironmentChangeRemoteApplicationEvent.java index 0d31587..d0ef03f 100644 --- a/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/event/EnvironmentChangeRemoteApplicationEvent.java +++ b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/event/EnvironmentChangeRemoteApplicationEvent.java @@ -19,6 +19,7 @@ package org.springframework.cloud.bus.event; import java.util.Map; import org.springframework.core.style.ToStringCreator; +import org.springframework.util.Assert; /** * @author Spencer Gibb @@ -34,9 +35,16 @@ public class EnvironmentChangeRemoteApplicationEvent extends RemoteApplicationEv this.values = null; } + @Deprecated public EnvironmentChangeRemoteApplicationEvent(Object source, String originService, String destinationService, Map values) { - super(source, originService, destinationService); + this(source, originService, new PathDestinationFactory().getDestination(destinationService), values); + } + + public EnvironmentChangeRemoteApplicationEvent(Object source, String originService, Destination destination, + Map values) { + super(source, originService, destination); + Assert.notNull(values, "values may not be null"); this.values = values; } diff --git a/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/event/PathDestinationFactory.java b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/event/PathDestinationFactory.java new file mode 100644 index 0000000..d63edd3 --- /dev/null +++ b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/event/PathDestinationFactory.java @@ -0,0 +1,42 @@ +/* + * 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.event; + +import org.springframework.util.StringUtils; + +public class PathDestinationFactory implements Destination.Factory { + + public Destination getDestination(String originalDestination) { + String path = originalDestination; + if (path == null) { + path = "**"; + } + // If the path is not already a wildcard, match everything that + // follows if there at most two path elements, and last element is not a global + // wildcard already + if (!"**".equals(path)) { + if (StringUtils.countOccurrencesOf(path, ":") <= 1 && !StringUtils.endsWithIgnoreCase(path, ":**")) { + // All instances of the destination unless specifically requested + path = path + ":**"; + } + } + + final String finalPath = path; + return () -> finalPath; + } + +} diff --git a/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/event/RefreshListener.java b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/event/RefreshListener.java index e60adff..af4998d 100644 --- a/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/event/RefreshListener.java +++ b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/event/RefreshListener.java @@ -50,7 +50,7 @@ public class RefreshListener implements ApplicationListener "unknown"); this.typeInfo = typeInfo; this.payload = payload; } diff --git a/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/jackson/BusJacksonAutoConfiguration.java b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/jackson/BusJacksonAutoConfiguration.java index bc8d176..3cc4ab9 100644 --- a/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/jackson/BusJacksonAutoConfiguration.java +++ b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/jackson/BusJacksonAutoConfiguration.java @@ -23,6 +23,7 @@ import java.util.Set; import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.exc.InvalidTypeIdException; +import com.fasterxml.jackson.dataformat.cbor.CBORFactory; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -47,6 +48,7 @@ import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.converter.AbstractMessageConverter; import org.springframework.util.ClassUtils; +import org.springframework.util.MimeType; import org.springframework.util.MimeTypeUtils; /** @@ -70,6 +72,18 @@ public class BusJacksonAutoConfiguration { return new BusJacksonMessageConverter(objectMapper); } + @Configuration(proxyBeanMethods = false) + @ConditionalOnClass(CBORFactory.class) + protected static class CborConfiguration { + + @Bean + public AbstractMessageConverter busCborConverter() { + return new BusJacksonMessageConverter(new MimeType("application", "cbor"), + new ObjectMapper(new CBORFactory())); + } + + } + } class BusJacksonMessageConverter extends AbstractMessageConverter implements InitializingBean { @@ -90,7 +104,12 @@ class BusJacksonMessageConverter extends AbstractMessageConverter implements Ini @Autowired(required = false) BusJacksonMessageConverter(@Nullable ObjectMapper objectMapper) { - super(MimeTypeUtils.APPLICATION_JSON); + this(MimeTypeUtils.APPLICATION_JSON, objectMapper); + } + + @Autowired(required = false) + BusJacksonMessageConverter(MimeType mimeType, @Nullable ObjectMapper objectMapper) { + super(mimeType); if (objectMapper != null) { this.mapper = objectMapper; diff --git a/spring-cloud-bus/src/main/resources/META-INF/spring.factories b/spring-cloud-bus/src/main/resources/META-INF/spring.factories index 874dec2..08fc39d 100644 --- a/spring-cloud-bus/src/main/resources/META-INF/spring.factories +++ b/spring-cloud-bus/src/main/resources/META-INF/spring.factories @@ -1,8 +1,8 @@ org.springframework.boot.autoconfigure.EnableAutoConfiguration=\ -org.springframework.cloud.bus.BusPropertiesAutoConfiguration,\ -org.springframework.cloud.bus.ServiceMatcherAutoConfiguration,\ +org.springframework.cloud.bus.PathServiceMatcherAutoConfiguration,\ org.springframework.cloud.bus.BusAutoConfiguration,\ org.springframework.cloud.bus.BusRefreshAutoConfiguration,\ +org.springframework.cloud.bus.BusStreamAutoConfiguration,\ org.springframework.cloud.bus.jackson.BusJacksonAutoConfiguration # Environment Post Processor org.springframework.boot.env.EnvironmentPostProcessor=\ diff --git a/spring-cloud-bus/src/test/java/org/springframework/cloud/bus/BusAutoConfigurationClassPathTests.java b/spring-cloud-bus/src/test/java/org/springframework/cloud/bus/BusAutoConfigurationClassPathTests.java index 0172c63..b6d234e 100644 --- a/spring-cloud-bus/src/test/java/org/springframework/cloud/bus/BusAutoConfigurationClassPathTests.java +++ b/spring-cloud-bus/src/test/java/org/springframework/cloud/bus/BusAutoConfigurationClassPathTests.java @@ -37,7 +37,7 @@ public class BusAutoConfigurationClassPathTests { public void refreshListenerCreatedWithoutActuator() { new ApplicationContextRunner() .withConfiguration(AutoConfigurations.of(RefreshAutoConfiguration.class, - ServiceMatcherAutoConfiguration.class, BusRefreshAutoConfiguration.class)) + PathServiceMatcherAutoConfiguration.class, BusRefreshAutoConfiguration.class)) .run(context -> assertThat(context).hasSingleBean(RefreshListener.class) .doesNotHaveBean(RefreshBusEndpoint.class)); } 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 93d7d5f..369aaed 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 @@ -28,6 +28,7 @@ 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.PathDestinationFactory; import org.springframework.cloud.bus.event.RefreshRemoteApplicationEvent; import org.springframework.cloud.bus.event.RemoteApplicationEvent; import org.springframework.cloud.bus.event.SentApplicationEvent; @@ -81,7 +82,7 @@ public class BusAutoConfigurationTests { this.context = SpringApplication.run(InboundMessageHandlerConfiguration.class, "--spring.cloud.bus.id=foo", "--server.port=0"); this.context.getBean(BusConstants.INPUT, MessageChannel.class) - .send(new GenericMessage<>(new RefreshRemoteApplicationEvent(this, "foo", null))); + .send(new GenericMessage<>(new RefreshRemoteApplicationEvent(this, "foo", (String) null))); assertThat(this.context.getBean(InboundMessageHandlerConfiguration.class).refresh).isNull(); } @@ -90,7 +91,7 @@ public class BusAutoConfigurationTests { this.context = SpringApplication.run(InboundMessageHandlerConfiguration.class, "--spring.cloud.bus.id=bar", "--server.port=0"); this.context.getBean(BusConstants.INPUT, MessageChannel.class) - .send(new GenericMessage<>(new RefreshRemoteApplicationEvent(this, "foo", null))); + .send(new GenericMessage<>(new RefreshRemoteApplicationEvent(this, "foo", (String) null))); assertThat(this.context.getBean(InboundMessageHandlerConfiguration.class).refresh).isNotNull(); } @@ -102,7 +103,7 @@ public class BusAutoConfigurationTests { 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))); + .send(new GenericMessage<>(new RefreshRemoteApplicationEvent(this, "foo", (String) null))); RefreshRemoteApplicationEvent refresh = this.context.getBean(InboundMessageHandlerConfiguration.class).refresh; assertThat(refresh).isNotNull(); TestStreamBusBridge busBridge = this.context.getBean(TestStreamBusBridge.class); @@ -119,7 +120,7 @@ public class BusAutoConfigurationTests { SentMessageConfiguration.class }, new String[] { "--spring.cloud.bus.trace.enabled=true", "--spring.cloud.bus.id=bar", "--server.port=0" }); - this.context.getBean(BusConsumer.class).accept(new RefreshRemoteApplicationEvent(this, "foo", null)); + this.context.getBean(BusConsumer.class).accept(new RefreshRemoteApplicationEvent(this, "foo", (String) null)); RefreshRemoteApplicationEvent refresh = this.context.getBean(InboundMessageHandlerConfiguration.class).refresh; assertThat(refresh).isNotNull(); SentMessageConfiguration sent = this.context.getBean(SentMessageConfiguration.class); @@ -134,8 +135,8 @@ public class BusAutoConfigurationTests { AckMessageConfiguration.class }, new String[] { "--spring.cloud.bus.trace.enabled=true", "--spring.cloud.bus.id=bar", "--server.port=0" }); - this.context.getBean(BusConsumer.class).accept( - new AckRemoteApplicationEvent(this, "foo", null, "ID", "bar", RefreshRemoteApplicationEvent.class)); + this.context.getBean(BusConsumer.class).accept(new AckRemoteApplicationEvent(this, "foo", + new PathDestinationFactory().getDestination(null), "ID", "bar", RefreshRemoteApplicationEvent.class)); AckMessageConfiguration ack = this.context.getBean(AckMessageConfiguration.class); assertThat(ack.latch.await(5, TimeUnit.SECONDS)).isTrue(); assertThat(ack.event).isNotNull(); @@ -146,7 +147,7 @@ public class BusAutoConfigurationTests { public void outboundFromSelf() throws Exception { this.context = SpringApplication.run(OutboundMessageHandlerConfiguration.class, "--debug=true", "--spring.cloud.bus.id=foo", "--server.port=0"); - this.context.publishEvent(new RefreshRemoteApplicationEvent(this, "foo", null)); + this.context.publishEvent(new RefreshRemoteApplicationEvent(this, "foo", (String) null)); TestStreamBusBridge busBridge = this.context.getBean(TestStreamBusBridge.class); busBridge.latch.await(2, TimeUnit.SECONDS); assertThat(busBridge.message).as("message was null").isNotNull(); @@ -156,7 +157,7 @@ public class BusAutoConfigurationTests { public void outboundNotFromSelf() { this.context = SpringApplication.run(OutboundMessageHandlerConfiguration.class, "--spring.cloud.bus.id=bar", "--server.port=0"); - this.context.publishEvent(new RefreshRemoteApplicationEvent(this, "foo", null)); + this.context.publishEvent(new RefreshRemoteApplicationEvent(this, "foo", (String) null)); assertThat(this.context.getBean(TestStreamBusBridge.class).message).isNull(); } @@ -230,9 +231,9 @@ public class BusAutoConfigurationTests { public void serviceMatcherIdIsConstantAfterRefresh() { this.context = SpringApplication.run(new Class[] { RefreshConfig.class, }, new String[] { "--server.port=0", "--spring.main.allow-bean-definition-overriding=true" }); - String originalServiceId = this.context.getBean(ServiceMatcher.class).getServiceId(); + String originalServiceId = this.context.getBean(ServiceMatcher.class).getBusId(); this.context.getBean(ContextRefresher.class).refresh(); - String newServiceId = this.context.getBean(ServiceMatcher.class).getServiceId(); + String newServiceId = this.context.getBean(ServiceMatcher.class).getBusId(); assertThat(newServiceId).isEqualTo(originalServiceId); } diff --git a/spring-cloud-bus/src/test/java/org/springframework/cloud/bus/ServiceMatcherTests.java b/spring-cloud-bus/src/test/java/org/springframework/cloud/bus/PathServiceMatcherTests.java similarity index 97% rename from spring-cloud-bus/src/test/java/org/springframework/cloud/bus/ServiceMatcherTests.java rename to spring-cloud-bus/src/test/java/org/springframework/cloud/bus/PathServiceMatcherTests.java index 93de933..3013910 100644 --- a/spring-cloud-bus/src/test/java/org/springframework/cloud/bus/ServiceMatcherTests.java +++ b/spring-cloud-bus/src/test/java/org/springframework/cloud/bus/PathServiceMatcherTests.java @@ -31,7 +31,7 @@ import static org.assertj.core.api.Assertions.assertThat; * @author Dave Syer * */ -public class ServiceMatcherTests { +public class PathServiceMatcherTests { private static final Map EMPTY_MAP = Collections.emptyMap(); @@ -46,7 +46,7 @@ public class ServiceMatcherTests { BusProperties properties = new BusProperties(); properties.setId(id); DefaultBusPathMatcher pathMatcher = new DefaultBusPathMatcher(new AntPathMatcher(":")); - this.matcher = new ServiceMatcher(pathMatcher, properties.getId()); + this.matcher = new PathServiceMatcher(pathMatcher, properties.getId()); } @Test diff --git a/spring-cloud-bus/src/test/java/org/springframework/cloud/bus/ServiceMatcherWithConfigNamesTests.java b/spring-cloud-bus/src/test/java/org/springframework/cloud/bus/PathServiceMatcherWithConfigNamesTests.java similarity index 97% rename from spring-cloud-bus/src/test/java/org/springframework/cloud/bus/ServiceMatcherWithConfigNamesTests.java rename to spring-cloud-bus/src/test/java/org/springframework/cloud/bus/PathServiceMatcherWithConfigNamesTests.java index 08c5e82..867658e 100644 --- a/spring-cloud-bus/src/test/java/org/springframework/cloud/bus/ServiceMatcherWithConfigNamesTests.java +++ b/spring-cloud-bus/src/test/java/org/springframework/cloud/bus/PathServiceMatcherWithConfigNamesTests.java @@ -31,7 +31,7 @@ import static org.assertj.core.api.Assertions.assertThat; * @author Stefan Pfeiffer * */ -public class ServiceMatcherWithConfigNamesTests { +public class PathServiceMatcherWithConfigNamesTests { private static final Map EMPTY_MAP = Collections.emptyMap(); @@ -46,7 +46,7 @@ public class ServiceMatcherWithConfigNamesTests { BusProperties properties = new BusProperties(); properties.setId(id); DefaultBusPathMatcher pathMatcher = new DefaultBusPathMatcher(new AntPathMatcher(":")); - this.matcher = new ServiceMatcher(pathMatcher, properties.getId(), configNames); + this.matcher = new PathServiceMatcher(pathMatcher, properties.getId(), configNames); } @Test diff --git a/spring-cloud-bus/src/test/java/org/springframework/cloud/bus/RefreshListenerIntegrationTests.java b/spring-cloud-bus/src/test/java/org/springframework/cloud/bus/RefreshListenerIntegrationTests.java index e0a0b2e..a5f1e32 100644 --- a/spring-cloud-bus/src/test/java/org/springframework/cloud/bus/RefreshListenerIntegrationTests.java +++ b/spring-cloud-bus/src/test/java/org/springframework/cloud/bus/RefreshListenerIntegrationTests.java @@ -26,12 +26,11 @@ import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.boot.test.mock.mockito.MockBean; import org.springframework.boot.test.web.client.TestRestTemplate; -import org.springframework.cloud.context.refresh.ContextRefresher; -import org.springframework.context.ApplicationEventPublisher; import org.springframework.http.HttpStatus; import org.springframework.test.context.junit4.SpringRunner; import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; @@ -47,11 +46,8 @@ public class RefreshListenerIntegrationTests { @Autowired private TestRestTemplate rest; - @Autowired - private ApplicationEventPublisher context; - @MockBean - ContextRefresher contextRefresher; + private BusBridge busBridge; @Test public void testEndpoint() { @@ -60,7 +56,7 @@ public class RefreshListenerIntegrationTests { .isEqualTo(HttpStatus.NO_CONTENT); assertThat(rest.postForEntity("/actuator/busrefresh/foobar", new HashMap<>(), String.class).getStatusCode()) .isEqualTo(HttpStatus.NO_CONTENT); - verify(contextRefresher, times(1)).refresh(); + verify(busBridge, times(2)).send(any()); } @SpringBootApplication diff --git a/spring-cloud-bus/src/test/java/org/springframework/cloud/bus/endpoint/RefreshBusEndpointTests.java b/spring-cloud-bus/src/test/java/org/springframework/cloud/bus/endpoint/RefreshBusEndpointTests.java index f787097..4284e8f 100644 --- a/spring-cloud-bus/src/test/java/org/springframework/cloud/bus/endpoint/RefreshBusEndpointTests.java +++ b/spring-cloud-bus/src/test/java/org/springframework/cloud/bus/endpoint/RefreshBusEndpointTests.java @@ -27,7 +27,8 @@ public class RefreshBusEndpointTests { @Test public void instanceId() throws Exception { - RefreshBusEndpoint endpoint = new RefreshBusEndpoint(null, "foo"); + RefreshBusEndpoint endpoint = new RefreshBusEndpoint(null, "foo", + originalDestination -> () -> originalDestination); assertThat(endpoint.getInstanceId()).isEqualTo("foo"); } diff --git a/spring-cloud-starter-bus-stream/pom.xml b/spring-cloud-starter-bus-stream/pom.xml new file mode 100644 index 0000000..b2cf55b --- /dev/null +++ b/spring-cloud-starter-bus-stream/pom.xml @@ -0,0 +1,33 @@ + + + 4.0.0 + + org.springframework.cloud + spring-cloud-bus-parent + 3.0.0-SNAPSHOT + .. + + spring-cloud-starter-bus-stream + spring-cloud-starter-bus-stream + Spring Cloud Starter + https://projects.spring.io/spring-cloud + + Pivotal Software, Inc. + https://www.spring.io + + + ${basedir}/../.. + + + + org.springframework.cloud + spring-cloud-stream + + + org.springframework.cloud + spring-cloud-bus + + +