Merge pull request #277 from ryanjbaxter/add-shutdow-event

Add a shutdown event, endpoint, and listener

.github/workflows/maven.yaml

@@ -4,21 +4,21 @@
 name: Build
 on:
     push:
-        branches: [ main, 3.1.x ]
+        branches: [ main, 4.1.x, 4.0.x, 3.1.x ]
     pull_request:
-        branches: [ main, 3.1.x ]
+        branches: [ main, 4.1.x, 4.0.x, 3.1.x ]
 jobs:
     build:
         runs-on: ubuntu-latest
         steps:
-            - uses: actions/checkout@v2
+            - uses: actions/checkout@v4
             - name: Set up JDK
-              uses: actions/setup-java@v2
+              uses: actions/setup-java@v4
               with:
                   distribution: 'temurin'
                   java-version: '17'
             - name: Cache local Maven repository
-              uses: actions/cache@v2
+              uses: actions/cache@v4
               with:
                   path: ~/.m2/repository
                   key: ${{ runner.os }}-maven-${{ hashFiles('**/pom.xml') }}
@@ -27,12 +27,12 @@ jobs:
             - name: Build with Maven
               run: ./mvnw -s .settings.xml clean org.jacoco:jacoco-maven-plugin:prepare-agent install -U -P sonar -nsu --batch-mode -Dmaven.test.redirectTestOutputToFile=true -Dorg.slf4j.simpleLogger.log.org.apache.maven.cli.transfer.Slf4jMavenTransferListener=warn
             - name: Publish Test Report
-              uses: mikepenz/action-junit-report@v2
+              uses: mikepenz/action-junit-report@v5
               if: always() # always run even if the previous step fails
               with:
                   report_paths: '**/surefire-reports/TEST-*.xml'
             - name: Archive code coverage results
-              uses: actions/upload-artifact@v2
+              uses: actions/upload-artifact@v4
               with:
                   name: surefire-reports
                   path: '**/surefire-reports/*'

docs/modules/ROOT/pages/quickstart.adoc

@@ -20,11 +20,11 @@ spring:
 ----
 
 The bus currently supports sending messages to all nodes listening or all nodes for a
-particular service (as defined by Eureka). The `/bus/*` actuator namespace has some HTTP
-endpoints. Currently, two are implemented. The first, `/bus/env`, sends key/value pairs to
-update each node's Spring Environment. The second, `/bus/refresh`, reloads each
+particular service (as defined by Eureka). The `/bus*` actuator namespace has some HTTP
+endpoints. Currently, three are implemented. The first, `/busenv`, sends key/value pairs to
+update each node's Spring Environment. The second, `/busrefresh`, reloads each
 application's configuration, as though they had all been pinged on their `/refresh`
-endpoint.
+endpoint.  The third `/busshutdown` sends a shutdown event to gracefully shutdown the application instance(s).
 
 NOTE: The Spring Cloud Bus starters cover Rabbit and Kafka, because those are the two most
 common implementations. However, Spring Cloud Stream is quite flexible, and the binder

docs/modules/ROOT/pages/spring-cloud-bus/bus-endpoints.adoc

@@ -2,9 +2,9 @@
 = Bus Endpoints
 :page-section-summary-toc: 1
 
-Spring Cloud Bus provides two endpoints, `/actuator/busrefresh` and `/actuator/busenv`
+Spring Cloud Bus provides three endpoints, `/actuator/busrefresh`, `/actutator/busshutdown` and `/actuator/busenv`
 that correspond to individual actuator endpoints in Spring Cloud Commons,
-`/actuator/refresh` and `/actuator/env` respectively.
+`/actuator/refresh`, `/actuator/shutdown`, and `/actuator/env` respectively.
 
 [[bus-refresh-endpoint]]
 == Bus Refresh Endpoint
@@ -41,4 +41,33 @@ The `/actuator/busenv` endpoint accepts `POST` requests with the following shape
 	"name": "key1",
 	"value": "value1"
 }
-----
+----
+
+[[bus-shutdown-endpoint]]
+== Bus Shutdown Endpoint
+The `/actuator/busshutdown` shuts down the application https://docs.spring.io/spring-boot/reference/web/graceful-shutdown.html[gracefully].
+
+To expose the `/actuator/busshutdown` endpoint, you need to add following configuration to your
+application:
+
+[source,properties]
+----
+management.endpoints.web.exposure.include=busshutdown
+----
+
+You can make a request to the `busshutdown` endpoint by issuing a `POST` request.
+
+If you would like to target a specific application you can issue a `POST` request to `/busshutdown` and optionally
+specify the bus id:
+
+[source,bash]
+----
+$ curl -X POST http://localhost:8080/actuator/busshutdown
+----
+
+You can also target a specific application instance by specifying the bus id:
+
+[source,bash]
+----
+$ curl -X POST http://localhost:8080/actuator/busshutdown/busid:123
+----

docs/modules/ROOT/partials/_configprops.adoc

@@ -9,6 +9,7 @@
 |spring.cloud.bus.env.enabled | `+++true+++` | Flag to switch off environment change events (default on).
 |spring.cloud.bus.id | `+++application+++` | The identifier for this application instance.
 |spring.cloud.bus.refresh.enabled | `+++true+++` | Flag to switch off refresh events (default on).
+|spring.cloud.bus.shutdown.enabled | `+++true+++` | Flag to switch off shutdown events (default on).
 |spring.cloud.bus.trace.enabled | `+++false+++` | Flag to switch on tracing of acks (default off).
 
 |===

spring-cloud-bus-rsocket/src/main/java/org/springframework/cloud/bus/rsocket/BusRSocketAutoConfiguration.java

@@ -26,6 +26,7 @@ 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.BusShutdownAutoConfiguration;
 import org.springframework.cloud.bus.ConditionalOnBusEnabled;
 import org.springframework.cloud.bus.PathServiceMatcherAutoConfiguration;
 import org.springframework.context.annotation.Bean;
@@ -39,7 +40,7 @@ import org.springframework.context.annotation.Configuration;
 @EnableConfigurationProperties(BusRSocketProperties.class)
 @ConditionalOnClass({ RSocket.class, RoutingRSocketRequester.class })
 @AutoConfigureBefore({ BusAutoConfiguration.class, BusRefreshAutoConfiguration.class,
-		PathServiceMatcherAutoConfiguration.class })
+		PathServiceMatcherAutoConfiguration.class, BusShutdownAutoConfiguration.class })
 public class BusRSocketAutoConfiguration {
 
 	@Bean

spring-cloud-bus-tests/pom.xml

@@ -17,10 +17,6 @@
 		<relativePath>..</relativePath> <!-- lookup parent from repository -->
 	</parent>
 
-	<properties>
-		<testcontainers.version>1.17.6</testcontainers.version>
-	</properties>
-
 	<dependencies>
 		<dependency>
 			<groupId>org.springframework.boot</groupId>
@@ -47,16 +43,19 @@
 			<artifactId>spring-boot-starter-test</artifactId>
 			<scope>test</scope>
 		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-testcontainers</artifactId>
+			<scope>test</scope>
+		</dependency>
 		<dependency>
 			<groupId>org.testcontainers</groupId>
 			<artifactId>junit-jupiter</artifactId>
-			<version>${testcontainers.version}</version>
 			<scope>test</scope>
 		</dependency>
 		<dependency>
 			<groupId>org.testcontainers</groupId>
 			<artifactId>rabbitmq</artifactId>
-			<version>${testcontainers.version}</version>
 			<scope>test</scope>
 		</dependency>
 	</dependencies>

spring-cloud-bus-tests/src/test/java/org/springframework/cloud/bus/ShutdownListenerIntegrationTests.java

@@ -0,0 +1,94 @@
+/*
+ * Copyright 2012-2024 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.bus;
+
+import java.util.concurrent.CountDownLatch;
+
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.testcontainers.containers.RabbitMQContainer;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.SpringBootConfiguration;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.builder.SpringApplicationBuilder;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.testcontainers.service.connection.ServiceConnection;
+import org.springframework.cloud.bus.event.EnvironmentChangeRemoteApplicationEvent;
+import org.springframework.context.ApplicationListener;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.test.web.reactive.server.WebTestClient;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.springframework.boot.test.context.SpringBootTest.WebEnvironment.RANDOM_PORT;
+
+/**
+ * @author Ryan Baxter
+ */
+@SpringBootTest(webEnvironment = RANDOM_PORT,
+		properties = { "management.endpoints.web.exposure.include=*",
+				"spring.cloud.stream.bindings.springCloudBusOutput.producer.errorChannelEnabled=true",
+				"logging.level.org.springframework.cloud.bus=TRACE", "spring.cloud.bus.id=app:1" })
+@Testcontainers
+public class ShutdownListenerIntegrationTests {
+
+	private static ConfigurableApplicationContext context;
+
+	@Container
+	@ServiceConnection
+	private static final RabbitMQContainer rabbitMQContainer = new RabbitMQContainer("rabbitmq:4.0-management");
+
+	@BeforeAll
+	static void before() {
+		context = new SpringApplicationBuilder(TestConfig.class)
+			.properties("server.port=0", "spring.rabbitmq.host=" + rabbitMQContainer.getHost(),
+					"spring.rabbitmq.port=" + rabbitMQContainer.getAmqpPort(),
+					"management.endpoints.web.exposure.include=*", "spring.cloud.bus.id=app:2", "debug=true")
+			.run();
+	}
+
+	@AfterAll
+	static void after() {
+		if (context != null) {
+			context.close();
+		}
+	}
+
+	@Test
+	void testShutdown(@Autowired WebTestClient client) {
+		assertThat(rabbitMQContainer.isRunning());
+		client.post().uri("/actuator/busshutdown/app:2").exchange().expectStatus().is2xxSuccessful();
+		assertThat(context.isClosed());
+	}
+
+	@SpringBootConfiguration
+	@EnableAutoConfiguration
+	static class TestConfig implements ApplicationListener<EnvironmentChangeRemoteApplicationEvent> {
+
+		CountDownLatch latch = new CountDownLatch(1);
+
+		@Override
+		public void onApplicationEvent(EnvironmentChangeRemoteApplicationEvent event) {
+			latch.countDown();
+		}
+
+	}
+
+}

spring-cloud-bus/src/main/java/org/springframework/cloud/bus/BusShutdownAutoConfiguration.java

@@ -0,0 +1,55 @@
+/*
+ * Copyright 2012-2024 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.condition.ConditionalOnClass;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.cloud.bus.endpoint.ShutdownBusEndpoint;
+import org.springframework.cloud.bus.event.Destination;
+import org.springframework.cloud.bus.event.ShutdownListener;
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * @author Ryan Baxter
+ */
+@Configuration(proxyBeanMethods = false)
+@ConditionalOnBusEnabled
+public class BusShutdownAutoConfiguration {
+
+	@Bean
+	@ConditionalOnProperty(value = "spring.cloud.bus.shutdown.enabled", matchIfMissing = true)
+	@ConditionalOnMissingBean
+	public ShutdownListener shutdownListener(ServiceMatcher serviceMatcher) {
+		return new ShutdownListener(serviceMatcher);
+	}
+
+	@Configuration(proxyBeanMethods = false)
+	@ConditionalOnClass(name = { "org.springframework.boot.actuate.endpoint.annotation.Endpoint" })
+	protected static class BusShutdownEndpointConfiguration {
+
+		@Bean
+		public ShutdownBusEndpoint shutdownBusEndpoint(ApplicationEventPublisher publisher, BusProperties bus,
+				Destination.Factory destinationFactory) {
+			return new ShutdownBusEndpoint(publisher, bus.getId(), destinationFactory);
+		}
+
+	}
+
+}

spring-cloud-bus/src/main/java/org/springframework/cloud/bus/endpoint/ShutdownBusEndpoint.java

@@ -0,0 +1,48 @@
+/*
+ * Copyright 2012-2024 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.endpoint;
+
+import org.springframework.boot.actuate.endpoint.annotation.Endpoint;
+import org.springframework.boot.actuate.endpoint.annotation.Selector;
+import org.springframework.boot.actuate.endpoint.annotation.WriteOperation;
+import org.springframework.cloud.bus.event.Destination;
+import org.springframework.cloud.bus.event.ShutdownRemoteApplicationEvent;
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.util.StringUtils;
+
+/**
+ * @author Ryan Baxter
+ */
+@Endpoint(id = "busshutdown")
+public class ShutdownBusEndpoint extends AbstractBusEndpoint {
+
+	public ShutdownBusEndpoint(ApplicationEventPublisher publisher, String id, Destination.Factory destinationFactory) {
+		super(publisher, id, destinationFactory);
+	}
+
+	@WriteOperation
+	public void busShutdownWithDestination(@Selector(match = Selector.Match.ALL_REMAINING) String[] destinations) {
+		String destination = StringUtils.arrayToDelimitedString(destinations, ":");
+		publish(new ShutdownRemoteApplicationEvent(this, getInstanceId(), getDestination(destination)));
+	}
+
+	@WriteOperation
+	public void busShutdown() {
+		publish(new ShutdownRemoteApplicationEvent(this, getInstanceId(), getDestination(null)));
+	}
+
+}

spring-cloud-bus/src/main/java/org/springframework/cloud/bus/event/ShutdownListener.java

@@ -0,0 +1,65 @@
+/*
+ * Copyright 2012-2024 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.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.beans.BeansException;
+import org.springframework.boot.SpringApplication;
+import org.springframework.cloud.bus.ServiceMatcher;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.ApplicationContextAware;
+import org.springframework.context.ApplicationListener;
+
+/**
+ * @author Ryan Baxter
+ */
+public class ShutdownListener implements ApplicationListener<ShutdownRemoteApplicationEvent>, ApplicationContextAware {
+
+	private static final Log LOG = LogFactory.getLog(ShutdownListener.class);
+
+	private ApplicationContext context;
+
+	private ServiceMatcher serviceMatcher;
+
+	public ShutdownListener(ServiceMatcher serviceMatcher) {
+		this.serviceMatcher = serviceMatcher;
+	}
+
+	@Override
+	public void onApplicationEvent(ShutdownRemoteApplicationEvent event) {
+		if (serviceMatcher.isForSelf(event)) {
+			LOG.warn("Received remote shutdown request from " + event.getOriginService() + ".  Shutting down.");
+			shutdown();
+		}
+		else {
+			LOG.info("Shutdown not performed, the event was targeting " + event.getDestinationService());
+		}
+
+	}
+
+	protected int shutdown() {
+		return SpringApplication.exit(context, () -> 0);
+	}
+
+	@Override
+	public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
+		this.context = applicationContext;
+	}
+
+}

spring-cloud-bus/src/main/java/org/springframework/cloud/bus/event/ShutdownRemoteApplicationEvent.java

@@ -0,0 +1,35 @@
+/*
+ * Copyright 2012-2024 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;
+
+/**
+ * Event which indicates the application should be shutdown.
+ *
+ * @author Ryan Baxter
+ */
+public class ShutdownRemoteApplicationEvent extends RemoteApplicationEvent {
+
+	@SuppressWarnings("unused")
+	private ShutdownRemoteApplicationEvent() {
+		// for serializers
+	}
+
+	public ShutdownRemoteApplicationEvent(Object source, String originService, Destination destination) {
+		super(source, originService, destination);
+	}
+
+}

spring-cloud-bus/src/main/resources/META-INF/additional-spring-configuration-metadata.json

@@ -12,6 +12,12 @@
 			"description": "Flag to switch off refresh events (default on).",
 			"defaultValue": true
 		},
+		{
+			"name": "spring.cloud.bus.shutdown.enabled",
+			"type": "java.lang.Boolean",
+			"description": "Flag to switch off shutdown events (default on).",
+			"defaultValue": true
+		},
 		{
 			"name": "spring.cloud.bus.trace.enabled",
 			"type": "java.lang.Boolean",

spring-cloud-bus/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports

@@ -1,5 +1,6 @@
 org.springframework.cloud.bus.PathServiceMatcherAutoConfiguration
 org.springframework.cloud.bus.BusAutoConfiguration
 org.springframework.cloud.bus.BusRefreshAutoConfiguration
+org.springframework.cloud.bus.BusShutdownAutoConfiguration
 org.springframework.cloud.bus.BusStreamAutoConfiguration
-org.springframework.cloud.bus.jackson.BusJacksonAutoConfiguration
+org.springframework.cloud.bus.jackson.BusJacksonAutoConfiguration

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

@@ -32,11 +32,12 @@ 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;
+import org.springframework.cloud.bus.event.ShutdownListener;
+import org.springframework.cloud.bus.event.ShutdownRemoteApplicationEvent;
 import org.springframework.cloud.bus.event.UnknownRemoteApplicationEvent;
 import org.springframework.cloud.context.refresh.ContextRefresher;
 import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
 import org.springframework.cloud.stream.config.BindingProperties;
-import org.springframework.cloud.stream.config.BindingServiceProperties;
 import org.springframework.cloud.stream.function.StreamBridge;
 import org.springframework.context.ApplicationListener;
 import org.springframework.context.ConfigurableApplicationContext;
@@ -48,7 +49,6 @@ import org.springframework.messaging.support.GenericMessage;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
 
 public class BusAutoConfigurationTests {
 
@@ -74,28 +74,61 @@ 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, "bar", "bar")));
+			.send(new GenericMessage<>(new RefreshRemoteApplicationEvent(this, "bar",
+					new PathDestinationFactory().getDestination("bar"))));
 		assertThat(this.context.getBean(InboundMessageHandlerConfiguration.class).refresh).isNull();
 	}
 
+	@Test
+	public void shutdownInboundNotForSelf() {
+		this.context = SpringApplication.run(ShutdownInboundMessageHandlerConfiguration.class,
+				"--spring.cloud.bus.id=foo", "--server.port=0");
+		this.context.getBean(BusConstants.INPUT, MessageChannel.class)
+			.send(new GenericMessage<>(new ShutdownRemoteApplicationEvent(this, "bar",
+					new PathDestinationFactory().getDestination("bar"))));
+		assertThat(this.context.getBean(ShutdownInboundMessageHandlerConfiguration.class).shutdown).isNull();
+	}
+
 	@Test
 	public void inboundFromSelf() {
 		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", (String) null)));
+			.send(new GenericMessage<>(
+					new RefreshRemoteApplicationEvent(this, "foo", new PathDestinationFactory().getDestination(null))));
 		assertThat(this.context.getBean(InboundMessageHandlerConfiguration.class).refresh).isNull();
 	}
 
+	@Test
+	public void shutdownInboundFromSelf() {
+		this.context = SpringApplication.run(ShutdownInboundMessageHandlerConfiguration.class,
+				"--spring.cloud.bus.id=foo", "--server.port=0");
+		this.context.getBean(BusConstants.INPUT, MessageChannel.class)
+			.send(new GenericMessage<>(new ShutdownRemoteApplicationEvent(this, "foo",
+					new PathDestinationFactory().getDestination(null))));
+		assertThat(this.context.getBean(ShutdownInboundMessageHandlerConfiguration.class).shutdown).isNull();
+	}
+
 	@Test
 	public void inboundNotFromSelf() {
 		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", (String) null)));
+			.send(new GenericMessage<>(
+					new RefreshRemoteApplicationEvent(this, "foo", new PathDestinationFactory().getDestination(null))));
 		assertThat(this.context.getBean(InboundMessageHandlerConfiguration.class).refresh).isNotNull();
 	}
 
+	@Test
+	public void shutdownInboundNotFromSelf() {
+		this.context = SpringApplication.run(ShutdownInboundMessageHandlerConfiguration.class,
+				"--spring.cloud.bus.id=bar", "--server.port=0");
+		this.context.getBean(BusConstants.INPUT, MessageChannel.class)
+			.send(new GenericMessage<>(new ShutdownRemoteApplicationEvent(this, "foo",
+					new PathDestinationFactory().getDestination(null))));
+		assertThat(this.context.getBean(ShutdownInboundMessageHandlerConfiguration.class).shutdown).isNotNull();
+	}
+
 	@Test
 	public void inboundNotFromSelfWithAck() throws Exception {
 		this.context = SpringApplication.run(
@@ -121,7 +154,8 @@ 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", (String) null));
+		this.context.getBean(BusConsumer.class)
+			.accept(new RefreshRemoteApplicationEvent(this, "foo", new PathDestinationFactory().getDestination(null)));
 		RefreshRemoteApplicationEvent refresh = this.context.getBean(InboundMessageHandlerConfiguration.class).refresh;
 		assertThat(refresh).isNotNull();
 		SentMessageConfiguration sent = this.context.getBean(SentMessageConfiguration.class);
@@ -149,7 +183,19 @@ 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", (String) null));
+		this.context.publishEvent(
+				new RefreshRemoteApplicationEvent(this, "foo", new PathDestinationFactory().getDestination(null)));
+		TestStreamBusBridge busBridge = this.context.getBean(TestStreamBusBridge.class);
+		busBridge.latch.await(2, TimeUnit.SECONDS);
+		assertThat(busBridge.message).as("message was null").isNotNull();
+	}
+
+	@Test
+	public void shutdownOutboundFromSelf() throws Exception {
+		this.context = SpringApplication.run(OutboundMessageHandlerConfiguration.class, "--debug=true",
+				"--spring.cloud.bus.id=foo", "--server.port=0");
+		this.context.publishEvent(
+				new ShutdownRemoteApplicationEvent(this, "foo", new PathDestinationFactory().getDestination(null)));
 		TestStreamBusBridge busBridge = this.context.getBean(TestStreamBusBridge.class);
 		busBridge.latch.await(2, TimeUnit.SECONDS);
 		assertThat(busBridge.message).as("message was null").isNotNull();
@@ -159,7 +205,17 @@ 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", (String) null));
+		this.context.publishEvent(
+				new RefreshRemoteApplicationEvent(this, "foo", new PathDestinationFactory().getDestination(null)));
+		assertThat(this.context.getBean(TestStreamBusBridge.class).message).isNull();
+	}
+
+	@Test
+	public void shutdownOutboundNotFromSelf() {
+		this.context = SpringApplication.run(OutboundMessageHandlerConfiguration.class, "--spring.cloud.bus.id=bar",
+				"--server.port=0");
+		this.context.publishEvent(
+				new ShutdownRemoteApplicationEvent(this, "foo", new PathDestinationFactory().getDestination(null)));
 		assertThat(this.context.getBean(TestStreamBusBridge.class).message).isNull();
 	}
 
@@ -168,28 +224,61 @@ public class BusAutoConfigurationTests {
 		this.context = SpringApplication.run(InboundMessageHandlerConfiguration.class, "--spring.cloud.bus.id=bar:1000",
 				"--server.port=0");
 		this.context.getBean(BusConstants.INPUT, MessageChannel.class)
-			.send(new GenericMessage<>(new RefreshRemoteApplicationEvent(this, "foo", "bar:*")));
+			.send(new GenericMessage<>(new RefreshRemoteApplicationEvent(this, "foo",
+					new PathDestinationFactory().getDestination("bar:*"))));
 		assertThat(this.context.getBean(InboundMessageHandlerConfiguration.class).refresh).isNotNull();
 	}
 
+	@Test
+	public void shutdownInboundNotFromSelfPathPattern() {
+		this.context = SpringApplication.run(ShutdownInboundMessageHandlerConfiguration.class,
+				"--spring.cloud.bus.id=bar:1000", "--server.port=0");
+		this.context.getBean(BusConstants.INPUT, MessageChannel.class)
+			.send(new GenericMessage<>(new ShutdownRemoteApplicationEvent(this, "foo",
+					new PathDestinationFactory().getDestination("bar:*"))));
+		assertThat(this.context.getBean(ShutdownInboundMessageHandlerConfiguration.class).shutdown).isNotNull();
+	}
+
 	@Test
 	public void inboundNotFromSelfDeepPathPattern() {
 		this.context = SpringApplication.run(InboundMessageHandlerConfiguration.class,
 				"--spring.cloud.bus.id=bar:test:1000", "--server.port=0");
 		this.context.getBean(BusConstants.INPUT, MessageChannel.class)
-			.send(new GenericMessage<>(new RefreshRemoteApplicationEvent(this, "foo", "bar:**")));
+			.send(new GenericMessage<>(new RefreshRemoteApplicationEvent(this, "foo",
+					new PathDestinationFactory().getDestination("bar:**"))));
 		assertThat(this.context.getBean(InboundMessageHandlerConfiguration.class).refresh).isNotNull();
 	}
 
+	@Test
+	public void shutdownInboundNotFromSelfDeepPathPattern() {
+		this.context = SpringApplication.run(ShutdownInboundMessageHandlerConfiguration.class,
+				"--spring.cloud.bus.id=bar:test:1000", "--server.port=0");
+		this.context.getBean(BusConstants.INPUT, MessageChannel.class)
+			.send(new GenericMessage<>(new ShutdownRemoteApplicationEvent(this, "foo",
+					new PathDestinationFactory().getDestination("bar:**"))));
+		assertThat(this.context.getBean(ShutdownInboundMessageHandlerConfiguration.class).shutdown).isNotNull();
+	}
+
 	@Test
 	public void inboundNotFromSelfFlatPattern() {
 		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", "bar*")));
+			.send(new GenericMessage<>(new RefreshRemoteApplicationEvent(this, "foo",
+					new PathDestinationFactory().getDestination("bar*"))));
 		assertThat(this.context.getBean(InboundMessageHandlerConfiguration.class).refresh).isNotNull();
 	}
 
+	@Test
+	public void shutdownInboundNotFromSelfFlatPattern() {
+		this.context = SpringApplication.run(ShutdownInboundMessageHandlerConfiguration.class,
+				"--spring.cloud.bus.id=bar", "--server.port=0");
+		this.context.getBean(BusConstants.INPUT, MessageChannel.class)
+			.send(new GenericMessage<>(new ShutdownRemoteApplicationEvent(this, "foo",
+					new PathDestinationFactory().getDestination("bar*"))));
+		assertThat(this.context.getBean(ShutdownInboundMessageHandlerConfiguration.class).shutdown).isNotNull();
+	}
+
 	// see https://github.com/spring-cloud/spring-cloud-bus/issues/74
 	@Test
 	public void inboundNotFromSelfUnknown() {
@@ -210,8 +299,6 @@ public class BusAutoConfigurationTests {
 		output.setDestination("mydestination");
 		properties.put(BusConstants.OUTPUT, output);
 
-		setupBusAutoConfig(properties);
-
 		BindingProperties inputProps = properties.get(BusConstants.INPUT);
 		assertThat(inputProps.getDestination()).isEqualTo("mydestination");
 
@@ -219,15 +306,6 @@ public class BusAutoConfigurationTests {
 		assertThat(outputProps.getDestination()).isEqualTo("mydestination");
 	}
 
-	private BusProperties setupBusAutoConfig(HashMap<String, BindingProperties> properties) {
-		BindingServiceProperties serviceProperties = mock(BindingServiceProperties.class);
-		when(serviceProperties.getBindings()).thenReturn(properties);
-
-		BusProperties bus = new BusProperties();
-		BusAutoConfiguration configuration = new BusAutoConfiguration();
-		return bus;
-	}
-
 	// see https://github.com/spring-cloud/spring-cloud-bus/issues/101
 	@Test
 	public void serviceMatcherIdIsConstantAfterRefresh() {
@@ -251,6 +329,11 @@ public class BusAutoConfigurationTests {
 			PropertyPlaceholderAutoConfiguration.class })
 	protected static class OutboundMessageHandlerConfiguration {
 
+		@Bean
+		ShutdownListener shutdownListener() {
+			return mock(ShutdownListener.class);
+		}
+
 		@Bean
 		@Primary
 		StreamBusBridge testStreamBusBridge(StreamBridge streamBridge, BusProperties properties) {
@@ -294,6 +377,28 @@ public class BusAutoConfigurationTests {
 
 	}
 
+	@Configuration(proxyBeanMethods = false)
+	@EnableAutoConfiguration
+	@ImportAutoConfiguration({ BusAutoConfiguration.class, TestChannelBinderConfiguration.class,
+			PropertyPlaceholderAutoConfiguration.class })
+	protected static class ShutdownInboundMessageHandlerConfiguration
+			implements ApplicationListener<ShutdownRemoteApplicationEvent> {
+
+		private ShutdownRemoteApplicationEvent shutdown;
+
+		@Bean
+		// Mock the shutdown listener so we don't try and shutdown the application
+		ShutdownListener customShutdownListener() {
+			return mock(ShutdownListener.class);
+		}
+
+		@Override
+		public void onApplicationEvent(ShutdownRemoteApplicationEvent event) {
+			this.shutdown = event;
+		}
+
+	}
+
 	@Configuration(proxyBeanMethods = false)
 	protected static class SentMessageConfiguration implements ApplicationListener<SentApplicationEvent> {
 

spring-cloud-bus/src/test/java/org/springframework/cloud/bus/ConditionalOnBusEnabledTests.java

@@ -33,6 +33,9 @@ import static org.assertj.core.api.Assertions.assertThat;
  */
 public class ConditionalOnBusEnabledTests {
 
+	/**
+	 * {@link ExpectedException} rule for verifying thrown exceptions.
+	 */
 	@Rule
 	public ExpectedException thrown = ExpectedException.none();
 

spring-cloud-bus/src/test/java/org/springframework/cloud/bus/PathServiceMatcherTests.java

@@ -136,7 +136,7 @@ public class PathServiceMatcherTests {
 	}
 
 	/**
-	 * see gh-678
+	 * see gh-678.
 	 */
 	@Test
 	public void forSelfWithMultipleProfiles() {
@@ -147,7 +147,7 @@ public class PathServiceMatcherTests {
 	}
 
 	/**
-	 * see gh-678
+	 * see gh-678.
 	 */
 	@Test
 	public void notForSelfWithMultipleProfiles() {
@@ -158,7 +158,7 @@ public class PathServiceMatcherTests {
 	}
 
 	/**
-	 * see gh-678
+	 * see gh-678.
 	 */
 	@Test
 	public void notForSelfWithMultipleProfilesDifferentPort() {

spring-cloud-bus/src/test/java/org/springframework/cloud/bus/jackson/RemoteApplicationEventScanTests.java

@@ -32,6 +32,7 @@ import org.springframework.boot.builder.SpringApplicationBuilder;
 import org.springframework.cloud.bus.event.AckRemoteApplicationEvent;
 import org.springframework.cloud.bus.event.EnvironmentChangeRemoteApplicationEvent;
 import org.springframework.cloud.bus.event.RefreshRemoteApplicationEvent;
+import org.springframework.cloud.bus.event.ShutdownRemoteApplicationEvent;
 import org.springframework.cloud.bus.event.UnknownRemoteApplicationEvent;
 import org.springframework.cloud.bus.event.test.TestRemoteApplicationEvent;
 import org.springframework.cloud.bus.event.test.TypedRemoteApplicationEvent;
@@ -123,6 +124,7 @@ public class RemoteApplicationEventScanTests {
 		expectedRegisterdClassesAsList.add(EnvironmentChangeRemoteApplicationEvent.class);
 		expectedRegisterdClassesAsList.add(RefreshRemoteApplicationEvent.class);
 		expectedRegisterdClassesAsList.add(UnknownRemoteApplicationEvent.class);
+		expectedRegisterdClassesAsList.add(ShutdownRemoteApplicationEvent.class);
 	}
 
 	@Configuration(proxyBeanMethods = false)

spring-cloud-bus/src/test/java/org/springframework/cloud/bus/jackson/SubtypeModuleTests.java

@@ -107,7 +107,7 @@ public class SubtypeModuleTests {
 	}
 
 	/**
-	 * see https://github.com/spring-cloud/spring-cloud-bus/issues/74
+	 * see https://github.com/spring-cloud/spring-cloud-bus/issues/74.
 	 */
 	@Test
 	public void testDeserializeAckRemoteApplicationEventWithKnownType() throws Exception {
@@ -124,7 +124,7 @@ public class SubtypeModuleTests {
 	}
 
 	/**
-	 * see https://github.com/spring-cloud/spring-cloud-bus/issues/74
+	 * see https://github.com/spring-cloud/spring-cloud-bus/issues/74.
 	 */
 	@Test
 	public void testDeserializeAckRemoteApplicationEventWithUnknownType() throws Exception {