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

Adds support for RSocket (#239)

Browse Source 
* Creates Destination.Factory to create Destinations

This allows the default PathDestinationFactory to send to all via ant patterns and RSocket to use tags.

* Adds spring-cloud-bus-rsocket to dependencies pom

* Moves stream to separate autoconfig

* Adds AntPathMatcher to handle isFromSelf

* Adds support for application/cbor

* Adds support for default tags to send on queries.

* Adds BusRSocketEnvironmentPostProcessor

This adds a bus=true tag to the RSocket routing client for registration with the RSocket routing broker

All tags in the remote event must be present from local configuration.

* Updates repo.spring.io urls

* Adds spring-cloud-starter-bus-stream

which includes spring-cloud-bus and spring-cloud-stream. This is because spring-cloud-stream is now optional in spring-cloud-bus. This is useful for bus users that want to use a different stream binder than amqp and kafka.
This commit is contained in:
Spencer Gibb
2020-10-31 13:55:27 -04:00
committed by GitHub
parent 5bb214977d
commit de0168ed6d
43 changed files with 881 additions and 173 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.mvn/maven.config
View File

@@ -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

8
.settings.xml
View File

@@ -23,7 +23,7 @@
<repository>
<id>spring-snapshots</id>
<name>Spring Snapshots</name>
<url>https://repo.spring.io/libs-snapshot-local</url>
<url>https://repo.spring.io/snapshot</url>
<snapshots>
<enabled>true</enabled>
</snapshots>
@@ -31,7 +31,7 @@
<repository>
<id>spring-milestones</id>
<name>Spring Milestones</name>
<url>https://repo.spring.io/libs-milestone-local</url>
<url>https://repo.spring.io/milestone</url>
<snapshots>
<enabled>false</enabled>
</snapshots>
@@ -49,7 +49,7 @@
<pluginRepository>
<id>spring-snapshots</id>
<name>Spring Snapshots</name>
<url>https://repo.spring.io/libs-snapshot-local</url>
<url>https://repo.spring.io/snapshot</url>
<snapshots>
<enabled>true</enabled>
</snapshots>
@@ -57,7 +57,7 @@
<pluginRepository>
<id>spring-milestones</id>
<name>Spring Milestones</name>
<url>https://repo.spring.io/libs-milestone-local</url>
<url>https://repo.spring.io/milestone</url>
<snapshots>
<enabled>false</enabled>
</snapshots>

25
pom.xml
View File

@@ -21,15 +21,18 @@
<modules>
<module>spring-cloud-bus-dependencies</module>
<module>spring-cloud-bus</module>
<module>spring-cloud-bus-rsocket</module>
<module>spring-cloud-bus-tests</module>
<module>spring-cloud-starter-bus-amqp</module>
<module>spring-cloud-starter-bus-kafka</module>
<module>spring-cloud-starter-bus-stream</module>
<module>docs</module>
</modules>
<properties>
<spring-cloud-commons.version>3.0.0-SNAPSHOT</spring-cloud-commons.version>
<spring-cloud-stream.version>3.1.0-SNAPSHOT</spring-cloud-stream.version>
<spring-cloud-function.version>3.1.0-SNAPSHOT</spring-cloud-function.version>
<bintray.package>bus</bintray.package>
</properties>
@@ -75,7 +78,7 @@
<repository>
<id>spring-snapshots</id>
<name>Spring Snapshots</name>
<url>https://repo.spring.io/libs-snapshot-local</url>
<url>https://repo.spring.io/snapshot</url>
<snapshots>
<enabled>true</enabled>
</snapshots>
@@ -86,7 +89,7 @@
<repository>
<id>spring-milestones</id>
<name>Spring Milestones</name>
<url>https://repo.spring.io/libs-milestone-local</url>
<url>https://repo.spring.io/milestone</url>
<snapshots>
<enabled>false</enabled>
</snapshots>
@@ -104,7 +107,7 @@
<pluginRepository>
<id>spring-snapshots</id>
<name>Spring Snapshots</name>
<url>https://repo.spring.io/libs-snapshot-local</url>
<url>https://repo.spring.io/snapshot</url>
<snapshots>
<enabled>true</enabled>
</snapshots>
@@ -115,7 +118,7 @@
<pluginRepository>
<id>spring-milestones</id>
<name>Spring Milestones</name>
<url>https://repo.spring.io/libs-milestone-local</url>
<url>https://repo.spring.io/milestone</url>
<snapshots>
<enabled>false</enabled>
</snapshots>
@@ -123,7 +126,7 @@
<pluginRepository>
<id>spring-releases</id>
<name>Spring Releases</name>
<url>https://repo.spring.io/libs-release-local</url>
<url>https://repo.spring.io/release</url>
<snapshots>
<enabled>false</enabled>
</snapshots>
@@ -159,6 +162,18 @@
<type>pom</type>
<scope>import</scope>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-function-dependencies</artifactId>
<version>${spring-cloud-function.version}</version>
<type>pom</type>
<scope>import</scope>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-function-rsocket</artifactId>
<version>${spring-cloud-function.version}</version>
</dependency>
</dependencies>
</dependencyManagement>

18
spring-cloud-bus-dependencies/pom.xml
View File

@@ -26,11 +26,21 @@
<artifactId>spring-cloud-starter-bus-kafka</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-bus-stream</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-bus</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-bus-rsocket</artifactId>
<version>${project.version}</version>
</dependency>
</dependencies>
</dependencyManagement>
<profiles>
@@ -40,7 +50,7 @@
<repository>
<id>spring-snapshots</id>
<name>Spring Snapshots</name>
<url>https://repo.spring.io/libs-snapshot-local</url>
<url>https://repo.spring.io/snapshot</url>
<snapshots>
<enabled>true</enabled>
</snapshots>
@@ -51,7 +61,7 @@
<repository>
<id>spring-milestones</id>
<name>Spring Milestones</name>
<url>https://repo.spring.io/libs-milestone-local</url>
<url>https://repo.spring.io/milestone</url>
<snapshots>
<enabled>false</enabled>
</snapshots>
@@ -69,7 +79,7 @@
<pluginRepository>
<id>spring-snapshots</id>
<name>Spring Snapshots</name>
<url>https://repo.spring.io/libs-snapshot-local</url>
<url>https://repo.spring.io/snapshot</url>
<snapshots>
<enabled>true</enabled>
</snapshots>
@@ -80,7 +90,7 @@
<pluginRepository>
<id>spring-milestones</id>
<name>Spring Milestones</name>
<url>https://repo.spring.io/libs-milestone-local</url>
<url>https://repo.spring.io/milestone</url>
<snapshots>
<enabled>false</enabled>
</snapshots>

75
spring-cloud-bus-rsocket/pom.xml Normal file
View File

@@ -0,0 +1,75 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns="http://maven.apache.org/POM/4.0.0"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<artifactId>spring-cloud-bus-rsocket</artifactId>
<packaging>jar</packaging>
<name>spring-cloud-bus-rsocket</name>
<description>Spring Cloud Bus</description>
<parent>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-bus-parent</artifactId>
<version>3.0.0-SNAPSHOT</version>
<relativePath>..</relativePath> <!-- lookup parent from repository -->
</parent>
<properties>
<rsocket-routing.version>0.2.0-SNAPSHOT</rsocket-routing.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-configuration-processor</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-actuator</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-bus</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-function-rsocket</artifactId>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.dataformat</groupId>
<artifactId>jackson-dataformat-cbor</artifactId>
</dependency>
<dependency>
<groupId>io.rsocket.routing</groupId>
<artifactId>rsocket-routing-client-spring</artifactId>
<version>${rsocket-routing.version}</version>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-autoconfigure-processor</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.junit.vintage</groupId>
<artifactId>junit-vintage-engine</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-test-support</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
</project>

61
spring-cloud-bus-rsocket/src/main/java/org/springframework/cloud/bus/rsocket/BusRSocketAutoConfiguration.java Normal file
View File

@@ -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);
}
}

45
spring-cloud-bus-rsocket/src/main/java/org/springframework/cloud/bus/rsocket/BusRSocketEnvironmentPostProcessor.java Normal file
View File

@@ -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<String, Object> defaults = new HashMap<>();
defaults.put(RoutingClientProperties.CONFIG_PREFIX + ".tags.bus", true);
addOrReplace(environment.getPropertySources(), defaults, DEFAULTS_PROPERTY_SOURCE_NAME, false);
}
}

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

@@ -0,0 +1,50 @@
/*
* Copyright 2015-2020 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.bus.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<MutableKey, String> defaultTags = new LinkedHashMap<>();
@Override
public void afterPropertiesSet() {
if (CollectionUtils.isEmpty(defaultTags)) {
defaultTags.put(MutableKey.of("bus"), "true");
}
}
public Map<MutableKey, String> getDefaultTags() {
return this.defaultTags;
}
@Override
public String toString() {
return new ToStringCreator(this).append("defaultTags", defaultTags).toString();
}
}

87
spring-cloud-bus-rsocket/src/main/java/org/springframework/cloud/bus/rsocket/RSocketRequesterBusBridge.java Normal file
View File

@@ -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<Key, String> getTagsFromDestination(String delimitedProperties) {
String[] properties = StringUtils.tokenizeToStringArray(delimitedProperties, ":");
Map<Key, String> 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;
}
}

88
spring-cloud-bus-rsocket/src/main/java/org/springframework/cloud/bus/rsocket/RSocketServiceMatcher.java Normal file
View File

@@ -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<Key, String> 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<Key, String> tags = getTagsFromDestination(event.getDestinationService());
for (Map.Entry<Key, String> 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;
}
}

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

@@ -0,0 +1,50 @@
/*
* Copyright 2015-2020 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.bus.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<String> 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;
};
}
}

6
spring-cloud-bus-rsocket/src/main/resources/META-INF/spring.factories Normal file
View File

@@ -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

2
spring-cloud-bus-tests/src/test/java/org/springframework/cloud/bus/jackson/BusJacksonIntegrationTests.java
View File

@@ -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")

6
spring-cloud-bus/pom.xml
View File

@@ -40,11 +40,17 @@
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-core</artifactId>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.dataformat</groupId>
<artifactId>jackson-dataformat-cbor</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-autoconfigure-processor</artifactId>

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

@@ -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> 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);
}
}

2
spring-cloud-bus/src/main/java/org/springframework/cloud/bus/BusConstants.java
View File

@@ -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.

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

@@ -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<RemoteApplicationEvent> {
private final ServiceMatcher serviceMatcher;
private final BusBridge busBridge;
private final ObjectProvider<BusBridge> 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> 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<RemoteApplicationEvent> {
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);
}
}

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

@@ -76,7 +76,7 @@ public class BusEnvironmentPostProcessor implements EnvironmentPostProcessor {
addOrReplace(environment.getPropertySources(), defaults, DEFAULTS_PROPERTY_SOURCE_NAME, false);
}
private void addOrReplace(MutablePropertySources propertySources, Map<String, Object> map,
public static void addOrReplace(MutablePropertySources propertySources, Map<String, Object> map,
String propertySourceName, boolean first) {
MapPropertySource target = null;
if (propertySources.contains(propertySourceName)) {

8
spring-cloud-bus/src/main/java/org/springframework/cloud/bus/BusRefreshAutoConfiguration.java
View File

@@ -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);
}
}

45
spring-cloud-bus/src/main/java/org/springframework/cloud/bus/BusStreamAutoConfiguration.java Normal file
View File

@@ -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);
}
}

79
spring-cloud-bus/src/main/java/org/springframework/cloud/bus/PathServiceMatcher.java Normal file
View File

@@ -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;
}
}

7
spring-cloud-bus/src/main/java/org/springframework/cloud/bus/ServiceMatcherAutoConfiguration.java → spring-cloud-bus/src/main/java/org/springframework/cloud/bus/PathServiceMatcherAutoConfiguration.java
View File

@@ -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);
}
}

55
spring-cloud-bus/src/main/java/org/springframework/cloud/bus/ServiceMatcher.java
View File

@@ -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();
}

21
spring-cloud-bus/src/main/java/org/springframework/cloud/bus/endpoint/AbstractBusEndpoint.java
View File

@@ -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);
}
}

17
spring-cloud-bus/src/main/java/org/springframework/cloud/bus/endpoint/EnvironmentBusEndpoint.java
View File

@@ -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<String, String> 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<String, String> params = Collections.singletonMap(name, value);
publish(new EnvironmentChangeRemoteApplicationEvent(this, getInstanceId(), null, params));
publish(new EnvironmentChangeRemoteApplicationEvent(this, getInstanceId(), getDestination(null), params));
}
}

14
spring-cloud-bus/src/main/java/org/springframework/cloud/bus/endpoint/RefreshBusEndpoint.java
View File

@@ -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)));
}
}

5
spring-cloud-bus/src/main/java/org/springframework/cloud/bus/event/AckRemoteApplicationEvent.java
View File

@@ -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<? extends RemoteApplicationEvent> 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<? extends RemoteApplicationEvent>) Class.forName(eventName);

22
spring-cloud-bus/src/main/java/org/springframework/cloud/bus/BusPropertiesAutoConfiguration.java → spring-cloud-bus/src/main/java/org/springframework/cloud/bus/event/Destination.java
View File

@@ -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);
}
}

10
spring-cloud-bus/src/main/java/org/springframework/cloud/bus/event/EnvironmentChangeRemoteApplicationEvent.java
View File

@@ -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<String, String> values) {
super(source, originService, destinationService);
this(source, originService, new PathDestinationFactory().getDestination(destinationService), values);
}
public EnvironmentChangeRemoteApplicationEvent(Object source, String originService, Destination destination,
Map<String, String> values) {
super(source, originService, destination);
Assert.notNull(values, "values may not be null");
this.values = values;
}

42
spring-cloud-bus/src/main/java/org/springframework/cloud/bus/event/PathDestinationFactory.java Normal file
View File

@@ -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;
}
}

2
spring-cloud-bus/src/main/java/org/springframework/cloud/bus/event/RefreshListener.java
View File

@@ -50,7 +50,7 @@ public class RefreshListener implements ApplicationListener<RefreshRemoteApplica
log.info("Keys refreshed " + keys);
}
else {
log.info("Refresh not performed, the event was targetting " + event.getDestinationService());
log.info("Refresh not performed, the event was targeting " + event.getDestinationService());
}
}

9
spring-cloud-bus/src/main/java/org/springframework/cloud/bus/event/RefreshRemoteApplicationEvent.java
View File

@@ -27,8 +27,13 @@ public class RefreshRemoteApplicationEvent extends RemoteApplicationEvent {
// for serializers
}
public RefreshRemoteApplicationEvent(Object source, String originService, String destinationService) {
super(source, originService, destinationService);
@Deprecated
public RefreshRemoteApplicationEvent(Object source, String originService, String destination) {
this(source, originService, new PathDestinationFactory().getDestination(destination));
}
public RefreshRemoteApplicationEvent(Object source, String originService, Destination destination) {
super(source, originService, destination);
}
}

40
spring-cloud-bus/src/main/java/org/springframework/cloud/bus/event/RemoteApplicationEvent.java
View File

@@ -23,7 +23,7 @@ import com.fasterxml.jackson.annotation.JsonTypeInfo;
import org.springframework.context.ApplicationEvent;
import org.springframework.core.style.ToStringCreator;
import org.springframework.util.StringUtils;
import org.springframework.util.Assert;
/**
* @author Spencer Gibb
@@ -35,6 +35,12 @@ public abstract class RemoteApplicationEvent extends ApplicationEvent {
private static final Object TRANSIENT_SOURCE = new Object();
private static final String TRANSIENT_ORIGIN = "____transient_origin_service___";
private static final String TRANSIENT_DESTINATION = "____transient_destination___";
protected static final PathDestinationFactory DEFAULT_DESTINATION_FACTORY = new PathDestinationFactory();
private final String originService;
private final String destinationService;
@@ -43,32 +49,32 @@ public abstract class RemoteApplicationEvent extends ApplicationEvent {
protected RemoteApplicationEvent() {
// for serialization libs like jackson
this(TRANSIENT_SOURCE, null, null);
this(TRANSIENT_SOURCE, TRANSIENT_ORIGIN, DEFAULT_DESTINATION_FACTORY.getDestination(TRANSIENT_DESTINATION));
}
@Deprecated
protected RemoteApplicationEvent(Object source, String originService, String destinationService) {
this(source, originService, DEFAULT_DESTINATION_FACTORY.getDestination(destinationService));
}
protected RemoteApplicationEvent(Object source, String originService, Destination destination) {
super(source);
this.originService = originService;
if (destinationService == null) {
destinationService = "**";
if (!originService.equals(TRANSIENT_ORIGIN)) {
Assert.notNull(originService, "originService may not be null");
this.originService = originService;
}
// If the destinationService 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(destinationService)) {
if (StringUtils.countOccurrencesOf(destinationService, ":") <= 1
&& !StringUtils.endsWithIgnoreCase(destinationService, ":**")) {
// All instances of the destination unless specifically requested
destinationService = destinationService + ":**";
}
else {
this.originService = null;
}
this.destinationService = destinationService;
Assert.notNull(destination, "destination may not be null");
this.destinationService = destination.getDestinationAsString();
Assert.hasText(destinationService, "destinationService may not be empty");
this.id = UUID.randomUUID().toString();
}
@Deprecated
protected RemoteApplicationEvent(Object source, String originService) {
this(source, originService, null);
this(source, originService, DEFAULT_DESTINATION_FACTORY.getDestination(null));
}
public String getOriginService() {

2
spring-cloud-bus/src/main/java/org/springframework/cloud/bus/event/UnknownRemoteApplicationEvent.java
View File

@@ -37,7 +37,7 @@ public class UnknownRemoteApplicationEvent extends RemoteApplicationEvent {
public UnknownRemoteApplicationEvent(Object source, String typeInfo, byte[] payload) {
// Initialize originService with an empty String, to avoid NullPointer in
// AntPathMatcher.
super(source, "", null);
super(source, "", () -> "unknown");
this.typeInfo = typeInfo;
this.payload = payload;
}

21
spring-cloud-bus/src/main/java/org/springframework/cloud/bus/jackson/BusJacksonAutoConfiguration.java
View File

@@ -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;

4
spring-cloud-bus/src/main/resources/META-INF/spring.factories
View File

@@ -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=\

2
spring-cloud-bus/src/test/java/org/springframework/cloud/bus/BusAutoConfigurationClassPathTests.java
View File

@@ -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));
}

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

@@ -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);
}

4
spring-cloud-bus/src/test/java/org/springframework/cloud/bus/ServiceMatcherTests.java → spring-cloud-bus/src/test/java/org/springframework/cloud/bus/PathServiceMatcherTests.java
View File

@@ -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<String, String> 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

4
spring-cloud-bus/src/test/java/org/springframework/cloud/bus/ServiceMatcherWithConfigNamesTests.java → spring-cloud-bus/src/test/java/org/springframework/cloud/bus/PathServiceMatcherWithConfigNamesTests.java
View File

@@ -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<String, String> 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

10
spring-cloud-bus/src/test/java/org/springframework/cloud/bus/RefreshListenerIntegrationTests.java
View File

@@ -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

3
spring-cloud-bus/src/test/java/org/springframework/cloud/bus/endpoint/RefreshBusEndpointTests.java
View File

@@ -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");
}

33
spring-cloud-starter-bus-stream/pom.xml Normal file
View File

@@ -0,0 +1,33 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns="http://maven.apache.org/POM/4.0.0"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-bus-parent</artifactId>
<version>3.0.0-SNAPSHOT</version>
<relativePath>..</relativePath>
</parent>
<artifactId>spring-cloud-starter-bus-stream</artifactId>
<name>spring-cloud-starter-bus-stream</name>
<description>Spring Cloud Starter</description>
<url>https://projects.spring.io/spring-cloud</url>
<organization>
<name>Pivotal Software, Inc.</name>
<url>https://www.spring.io</url>
</organization>
<properties>
<main.basedir>${basedir}/../..</main.basedir>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-bus</artifactId>
</dependency>
</dependencies>
</project>