diff --git a/docs/src/main/asciidoc/spring-cloud-bus.adoc b/docs/src/main/asciidoc/spring-cloud-bus.adoc index 72a76ab..106a430 100644 --- a/docs/src/main/asciidoc/spring-cloud-bus.adoc +++ b/docs/src/main/asciidoc/spring-cloud-bus.adoc @@ -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 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 945381a..2122736 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 @@ -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()); } } } 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 8ef6a61..896fc43 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 @@ -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 map) { MapPropertySource target = null; diff --git a/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/BusProperties.java b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/BusProperties.java index cad4afb..52aa5f3 100644 --- a/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/BusProperties.java +++ b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/BusProperties.java @@ -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). 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 8026b7c..a3f281c 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,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() { 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 c51d079..d75697a 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 @@ -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 } 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/ServiceMatcherTests.java index 1ee475a..b74221f 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/ServiceMatcherTests.java @@ -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 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 diff --git a/spring-cloud-bus/src/test/resources/application.properties b/spring-cloud-bus/src/test/resources/application.properties index da40352..d18f0f6 100644 --- a/spring-cloud-bus/src/test/resources/application.properties +++ b/spring-cloud-bus/src/test/resources/application.properties @@ -1 +1 @@ -spring.main.web-environment: false +spring.main.web-environment: false \ No newline at end of file