Add a shutdown event, endpoint, and listener

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,32 @@ 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 with the bus id:
+
+[source,bash]
+----
+$ curl -X POST http://localhost:8080/actuator/busshutdown/busid
+----
+
+You can also target a specific application instance:
+
+[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.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 @@
 @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;
+	}
+
+}