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

Instead of ApplicationContext ID use BusProperties to address events

Browse Source 
The ApplicationContext is not a very good place to store something
that has to be unique per instance across a scaled up app. This change
extracts the old value out into a new spring.cloud.bus.id.

It also adds a unique identifier to the end of the id. Hopefully
this interops with older versions. If not we can change the default
and/or add a property to switch off this change.
This commit is contained in:
Dave Syer
2017-11-20 10:16:12 +00:00
parent 1ddc238db7
commit cb79556992
8 changed files with 73 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
docs/src/main/asciidoc/spring-cloud-bus.adoc
View File

@@ -15,15 +15,21 @@ include::quickstart.adoc[]
== Addressing an Instance
The HTTP endpoints accept a "destination" parameter, e.g. "/bus/refresh?destination=customers:9000", where the destination is an `ApplicationContext` ID. If the ID is owned by an instance on the Bus then it will process the message and all other instances will ignore it. Spring Boot sets the ID for you in the `ContextIdApplicationContextInitializer` to a combination of the `spring.application.name`, active profiles and `server.port` by default.
Each instance of the application has a service ID, whose value can be set using `spring.cloud.bus.id`, and whose value is expected to be a colon-separated list of identifiers, in order of least specific to most specific. The default value is constructed from the environment as a combination of the `spring.application.name` and `server.port` (or `spring.application.index` if set). The default value of the ID is constructed in the form `app:index:id` where:
* `app` is the `vcap.application.name` if it exists, or `spring.application.name`
* `index` is the `vcap.application.instance_index` if it exists, or else `spring.application.index`, or else `local.server.port` (or `server.port` or `0`).
* `id` is the `vcap.application.instance_id` if it exists, or else a random value.
The HTTP endpoints accept a "destination" parameter, e.g. "/bus/refresh?destination=customers:9000", where the destination is a service ID. If the ID is owned by an instance on the Bus then it will process the message and all other instances will ignore it.
== Addressing all instances of a service
The "destination" parameter is used in a Spring `PathMatcher` (with the path separator as a colon `:`) to determine if an instance will process the message. Using the example from above, "/bus/refresh?destination=customers:**" will target all instances of the "customers" service regardless of the profiles and ports set as the `ApplicationContext` ID.
The "destination" parameter is used in a Spring `PathMatcher` (with the path separator as a colon `:`) to determine if an instance will process the message. Using the example from above, "/bus/refresh?destination=customers:**" will target all instances of the "customers" service regardless of the rest of the service ID.
== Application Context ID must be unique
== Service ID must be unique
The bus tries to eliminate processing an event twice, once from the original `ApplicationEvent` and once from the queue. To do this, it checks the sending application context id againts the current application context id. If multiple instances of a service have the same application context id, events will not be processed. Running on a local machine, each service will be on a different port and that will be part of the application context id. Cloud Foundry supplies an index to differentiate. To ensure that the application context id is the unique, set `spring.application.index` to something unique for each instance of a service. For example, in lattice, set `spring.application.index=${INSTANCE_INDEX}` in application.properties (or bootstrap.properties if using configserver).
The bus tries to eliminate processing an event twice, once from the original `ApplicationEvent` and once from the queue. To do this, it checks the sending service ID againts the current service ID. If multiple instances of a service have the same ID, events will not be processed. Running on a local machine, each service will be on a different port and that will be part of the ID. Cloud Foundry supplies an index to differentiate. To ensure that the ID is unique outside Cloud Foundry, set `spring.application.index` to something unique for each instance of a service.
== Customizing the Message Broker

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

@@ -167,9 +167,11 @@ public class BusAutoConfiguration implements ApplicationEventPublisherAware {
}
@Bean
public ServiceMatcher serviceMatcher(@BusPathMatcher PathMatcher pathMatcher) {
public ServiceMatcher serviceMatcher(@BusPathMatcher PathMatcher pathMatcher,
BusProperties bus) {
ServiceMatcher serviceMatcher = new ServiceMatcher();
serviceMatcher.setMatcher(pathMatcher);
serviceMatcher.setBusProperties(bus);
return serviceMatcher;
}
@@ -190,8 +192,9 @@ public class BusAutoConfiguration implements ApplicationEventPublisherAware {
@ConditionalOnProperty(value = "endpoints.spring.cloud.bus.refresh.enabled", matchIfMissing = true)
protected static class BusRefreshEndpointConfiguration {
@Bean
public RefreshBusEndpoint refreshBusEndpoint(ApplicationContext context) {
return new RefreshBusEndpoint(context, context.getId());
public RefreshBusEndpoint refreshBusEndpoint(ApplicationContext context,
BusProperties bus) {
return new RefreshBusEndpoint(context, bus.getId());
}
}
@@ -227,8 +230,8 @@ public class BusAutoConfiguration implements ApplicationEventPublisherAware {
protected static class EnvironmentBusEndpointConfiguration {
@Bean
public EnvironmentBusEndpoint environmentBusEndpoint(
ApplicationContext context) {
return new EnvironmentBusEndpoint(context, context.getId());
ApplicationContext context, BusProperties bus) {
return new EnvironmentBusEndpoint(context, bus.getId());
}
}
}

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

@@ -46,9 +46,15 @@ public class BusEnvironmentPostProcessor implements EnvironmentPostProcessor {
+ ".content-type",
environment.getProperty("spring.cloud.bus.content-type",
"application/json"));
map.put("spring.cloud.bus.id", getDefaultServiceId(environment));
addOrReplace(environment.getPropertySources(), map);
}
// TODO: move this to commons
private String getDefaultServiceId(ConfigurableEnvironment environment) {
return "${vcap.application.name:${spring.application.name:${spring.application.name:application}}}:${vcap.application.instance_index:${spring.application.index:${local.server.port:${server.port:0}}}}:${vcap.application.instance_id:${random.value}}";
}
private void addOrReplace(MutablePropertySources propertySources,
Map<String, Object> map) {
MapPropertySource target = null;

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

@@ -46,7 +46,10 @@ public class BusProperties {
* Name of Spring Cloud Stream destination for messages.
*/
private String destination = "springCloudBus";
/**
* The identifier for this application instance.
*/
private String id = "application";
/**
* Flag to indicate that the bus is enabled.
*/
@@ -84,6 +87,14 @@ public class BusProperties {
this.enabled = enabled;
}
public String getId() {
return id;
}
public void setId(String id) {
this.id = id;
}
public static class Env {
/**
* Flag to switch off environment change events (default on).

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

@@ -17,21 +17,17 @@
package org.springframework.cloud.bus;
import org.springframework.beans.BeansException;
import org.springframework.cloud.bus.event.RemoteApplicationEvent;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.util.PathMatcher;
/**
* @author Spencer Gibb
*/
public class ServiceMatcher implements ApplicationContextAware {
private ApplicationContext context;
public class ServiceMatcher {
private BusProperties context;
private PathMatcher matcher;
@Override
public void setApplicationContext(ApplicationContext context) throws BeansException {
public void setBusProperties(BusProperties context) {
this.context = context;
}
@@ -47,8 +43,8 @@ public class ServiceMatcher implements ApplicationContextAware {
public boolean isForSelf(RemoteApplicationEvent event) {
String destinationService = event.getDestinationService();
return (destinationService == null || destinationService.trim().isEmpty() || this.matcher
.match(destinationService, getServiceId()));
return (destinationService == null || destinationService.trim().isEmpty()
|| this.matcher.match(destinationService, getServiceId()));
}
public String getServiceId() {

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

@@ -17,11 +17,6 @@
package org.springframework.cloud.bus;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@@ -29,6 +24,7 @@ import javax.annotation.PostConstruct;
import org.junit.After;
import org.junit.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.context.PropertyPlaceholderAutoConfiguration;
@@ -51,6 +47,11 @@ import org.springframework.messaging.support.ChannelInterceptor;
import org.springframework.messaging.support.ChannelInterceptorAdapter;
import org.springframework.messaging.support.GenericMessage;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
public class BusAutoConfigurationTests {
private ConfigurableApplicationContext context;
@@ -62,10 +63,18 @@ public class BusAutoConfigurationTests {
}
}
@Test
public void defaultId() {
this.context = SpringApplication.run(InboundMessageHandlerConfiguration.class);
assertTrue("Wrong ID: " + context.getBean(BusProperties.class).getId(),
this.context.getBean(BusProperties.class).getId()
.startsWith("application:0:"));
}
@Test
public void inboundNotForSelf() {
this.context = SpringApplication.run(InboundMessageHandlerConfiguration.class);
this.context.setId("foo");
this.context.getBean(BusProperties.class).setId("foo");
this.context.getBean(SpringCloudBusClient.INPUT, MessageChannel.class)
.send(new GenericMessage<>(
new RefreshRemoteApplicationEvent(this, "bar", "bar")));
@@ -76,7 +85,7 @@ public class BusAutoConfigurationTests {
@Test
public void inboundFromSelf() {
this.context = SpringApplication.run(InboundMessageHandlerConfiguration.class);
this.context.setId("foo");
this.context.getBean(BusProperties.class).setId("foo");
this.context.getBean(SpringCloudBusClient.INPUT, MessageChannel.class)
.send(new GenericMessage<>(
new RefreshRemoteApplicationEvent(this, "foo", null)));
@@ -87,7 +96,7 @@ public class BusAutoConfigurationTests {
@Test
public void inboundNotFromSelf() {
this.context = SpringApplication.run(InboundMessageHandlerConfiguration.class);
this.context.setId("bar");
this.context.getBean(BusProperties.class).setId("bar");
this.context.getBean(SpringCloudBusClient.INPUT, MessageChannel.class)
.send(new GenericMessage<>(
new RefreshRemoteApplicationEvent(this, "foo", null)));
@@ -101,7 +110,7 @@ public class BusAutoConfigurationTests {
.run(new Class[] { InboundMessageHandlerConfiguration.class,
OutboundMessageHandlerConfiguration.class,
SentMessageConfiguration.class }, new String[] {});
this.context.setId("bar");
this.context.getBean(BusProperties.class).setId("bar");
this.context.getBean(SpringCloudBusClient.INPUT, MessageChannel.class)
.send(new GenericMessage<>(
new RefreshRemoteApplicationEvent(this, "foo", null)));
@@ -123,7 +132,7 @@ public class BusAutoConfigurationTests {
OutboundMessageHandlerConfiguration.class,
SentMessageConfiguration.class },
new String[] { "--spring.cloud.bus.trace.enabled=true" });
this.context.setId("bar");
this.context.getBean(BusProperties.class).setId("bar");
this.context.getBean(SpringCloudBusClient.INPUT, MessageChannel.class)
.send(new GenericMessage<>(
new RefreshRemoteApplicationEvent(this, "foo", null)));
@@ -143,7 +152,7 @@ public class BusAutoConfigurationTests {
OutboundMessageHandlerConfiguration.class,
AckMessageConfiguration.class },
new String[] { "--spring.cloud.bus.trace.enabled=true" });
this.context.setId("bar");
this.context.getBean(BusProperties.class).setId("bar");
this.context.getBean(SpringCloudBusClient.INPUT, MessageChannel.class)
.send(new GenericMessage<>(new AckRemoteApplicationEvent(this, "foo",
null, "ID", "bar", RefreshRemoteApplicationEvent.class)));
@@ -157,7 +166,7 @@ public class BusAutoConfigurationTests {
public void outboundFromSelf() throws Exception {
this.context = SpringApplication.run(OutboundMessageHandlerConfiguration.class,
"--debug=true");
this.context.setId("foo");
this.context.getBean(BusProperties.class).setId("foo");
this.context.publishEvent(new RefreshRemoteApplicationEvent(this, "foo", null));
OutboundMessageHandlerConfiguration outbound = this.context
.getBean(OutboundMessageHandlerConfiguration.class);
@@ -168,7 +177,7 @@ public class BusAutoConfigurationTests {
@Test
public void outboundNotFromSelf() {
this.context = SpringApplication.run(OutboundMessageHandlerConfiguration.class);
this.context.setId("bar");
this.context.getBean(BusProperties.class).setId("bar");
this.context.publishEvent(new RefreshRemoteApplicationEvent(this, "foo", null));
assertNull(
this.context.getBean(OutboundMessageHandlerConfiguration.class).message);
@@ -177,7 +186,7 @@ public class BusAutoConfigurationTests {
@Test
public void inboundNotFromSelfPathPattern() {
this.context = SpringApplication.run(InboundMessageHandlerConfiguration.class);
this.context.setId("bar:1000");
this.context.getBean(BusProperties.class).setId("bar:1000");
this.context.getBean(SpringCloudBusClient.INPUT, MessageChannel.class)
.send(new GenericMessage<>(
new RefreshRemoteApplicationEvent(this, "foo", "bar:*")));
@@ -188,7 +197,7 @@ public class BusAutoConfigurationTests {
@Test
public void inboundNotFromSelfDeepPathPattern() {
this.context = SpringApplication.run(InboundMessageHandlerConfiguration.class);
this.context.setId("bar:test:1000");
this.context.getBean(BusProperties.class).setId("bar:test:1000");
this.context.getBean(SpringCloudBusClient.INPUT, MessageChannel.class)
.send(new GenericMessage<>(
new RefreshRemoteApplicationEvent(this, "foo", "bar:**")));
@@ -199,7 +208,7 @@ public class BusAutoConfigurationTests {
@Test
public void inboundNotFromSelfFlatPattern() {
this.context = SpringApplication.run(InboundMessageHandlerConfiguration.class);
this.context.setId("bar");
this.context.getBean(BusProperties.class).setId("bar");
this.context.getBean(SpringCloudBusClient.INPUT, MessageChannel.class)
.send(new GenericMessage<>(
new RefreshRemoteApplicationEvent(this, "foo", "bar*")));
@@ -213,10 +222,10 @@ public class BusAutoConfigurationTests {
@Test
public void inboundNotFromSelfUnknown() {
this.context = SpringApplication.run(InboundMessageHandlerConfiguration.class);
this.context.setId("bar");
this.context.getBean(BusProperties.class).setId("bar");
this.context.getBean(SpringCloudBusClient.INPUT, MessageChannel.class)
.send(new GenericMessage<>(
new UnknownRemoteApplicationEvent(this, "UnknownEvent", "yada".getBytes())));
.send(new GenericMessage<>(new UnknownRemoteApplicationEvent(this,
"UnknownEvent", "yada".getBytes())));
// No Exception expected
}

7
spring-cloud-bus/src/test/java/org/springframework/cloud/bus/ServiceMatcherTests.java
View File

@@ -22,8 +22,8 @@ import java.util.Map;
import org.junit.Before;
import org.junit.Test;
import org.springframework.cloud.bus.event.EnvironmentChangeRemoteApplicationEvent;
import org.springframework.context.support.StaticApplicationContext;
import org.springframework.util.AntPathMatcher;
import static org.hamcrest.Matchers.is;
@@ -38,14 +38,13 @@ public class ServiceMatcherTests {
private static final Map<String, String> EMPTY_MAP = Collections.emptyMap();
private ServiceMatcher matcher = new ServiceMatcher();
private StaticApplicationContext context = new StaticApplicationContext();
private BusProperties context = new BusProperties();
@Before
public void init() {
context.setId("one:two:8888");
context.refresh();
matcher.setMatcher(new DefaultBusPathMatcher(new AntPathMatcher(":")));
matcher.setApplicationContext(context);
matcher.setBusProperties(context);
}
@Test

2
spring-cloud-bus/src/test/resources/application.properties
View File

@@ -1 +1 @@
spring.main.web-environment: false
spring.main.web-environment: false