Add JGroups integration for Spring Cloud Bus

bus/pom.xml

@@ -23,6 +23,7 @@
 		<module>spring-cloud-starter-bus-amqp</module>
 		<module>spring-cloud-starter-bus-kafka</module>
 		<module>spring-cloud-starter-bus-stream</module>
+		<module>spring-cloud-starter-bus-jgroups</module>
 	</modules>
 
 	<licenses>
@@ -77,6 +78,11 @@
 				<artifactId>spring-cloud-stream</artifactId>
 				<version>${project.version}</version>
 			</dependency>
+			<dependency>
+				<groupId>org.jgroups</groupId>
+				<artifactId>jgroups</artifactId>
+				<version>5.5.7.Final</version>
+			</dependency>
 		</dependencies>
 	</dependencyManagement>
 

bus/spring-cloud-starter-bus-jgroups/pom.xml

@@ -0,0 +1,48 @@
+<?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>5.1.0-SNAPSHOT</version>
+		<relativePath>..</relativePath>
+	</parent>
+
+	<artifactId>spring-cloud-starter-bus-jgroups</artifactId>
+	<name>spring-cloud-starter-bus-jgroups</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.jgroups</groupId>
+			<artifactId>jgroups</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>tools.jackson.core</groupId>
+			<artifactId>jackson-databind</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-bus</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-test</artifactId>
+			<scope>test</scope>
+		</dependency>
+	</dependencies>
+
+</project>

bus/spring-cloud-starter-bus-jgroups/src/main/java/org/springframework/cloud/bus/jgroups/JGroupsBusAutoConfiguration.java

@@ -0,0 +1,55 @@
+/*
+ * Copyright 2015-present 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.jgroups;
+
+import java.util.function.Consumer;
+
+import org.jgroups.JChannel;
+
+import tools.jackson.databind.ObjectMapper;
+
+import org.springframework.boot.autoconfigure.AutoConfiguration;
+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.bus.BusAutoConfiguration;
+import org.springframework.cloud.bus.BusBridge;
+import org.springframework.cloud.bus.BusStreamAutoConfiguration;
+import org.springframework.cloud.bus.ConditionalOnBusEnabled;
+import org.springframework.cloud.bus.event.RemoteApplicationEvent;
+import org.springframework.context.annotation.Bean;
+
+@AutoConfiguration(before = { BusStreamAutoConfiguration.class, BusAutoConfiguration.class })
+@ConditionalOnBusEnabled
+@ConditionalOnClass({ JChannel.class, ObjectMapper.class })
+@EnableConfigurationProperties(JGroupsBusProperties.class)
+public class JGroupsBusAutoConfiguration {
+
+	@Bean
+	@ConditionalOnMissingBean
+	JGroupsChannelFactory jGroupsChannelFactory() {
+		return new JGroupsChannelFactory();
+	}
+
+	@Bean
+	@ConditionalOnMissingBean(BusBridge.class)
+	JGroupsBusBridge jGroupsBusBridge(JGroupsBusProperties properties, ObjectMapper objectMapper,
+			Consumer<RemoteApplicationEvent> eventConsumer, JGroupsChannelFactory channelFactory) {
+		return new JGroupsBusBridge(properties, objectMapper, eventConsumer, channelFactory);
+	}
+
+}

bus/spring-cloud-starter-bus-jgroups/src/main/java/org/springframework/cloud/bus/jgroups/JGroupsBusBridge.java

@@ -0,0 +1,80 @@
+/*
+ * Copyright 2015-present 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.jgroups;
+
+import java.nio.charset.StandardCharsets;
+import java.util.function.Consumer;
+
+import org.jgroups.BytesMessage;
+import org.jgroups.JChannel;
+import org.jgroups.Message;
+import org.jgroups.Receiver;
+
+import tools.jackson.databind.ObjectMapper;
+
+import org.springframework.cloud.bus.BusBridge;
+import org.springframework.cloud.bus.event.RemoteApplicationEvent;
+
+public class JGroupsBusBridge implements BusBridge {
+
+	private final JChannel channel;
+
+	private final ObjectMapper objectMapper;
+
+	public JGroupsBusBridge(JGroupsBusProperties properties, ObjectMapper objectMapper,
+			Consumer<RemoteApplicationEvent> eventConsumer, JGroupsChannelFactory channelFactory) {
+		try {
+			this.objectMapper = objectMapper;
+			this.channel = channelFactory.create();
+			this.channel.setReceiver(new Receiver() {
+				@Override
+				public void receive(Message message) {
+					try {
+						String payload = new String(message.getArray(), message.getOffset(), message.getLength(),
+								StandardCharsets.UTF_8);
+						RemoteApplicationEvent event = JGroupsBusBridge.this.objectMapper.readValue(payload,
+								RemoteApplicationEvent.class);
+						eventConsumer.accept(event);
+					}
+					catch (Exception ex) {
+						throw new IllegalStateException("Unable to receive event from JGroups", ex);
+					}
+				}
+			});
+			this.channel.connect(properties.getClusterName());
+		}
+		catch (Exception ex) {
+			throw new IllegalStateException("Unable to initialize JGroups channel", ex);
+		}
+	}
+
+	@Override
+	public void send(RemoteApplicationEvent event) {
+		try {
+			byte[] payload = this.objectMapper.writeValueAsBytes(event);
+			this.channel.send(new BytesMessage(null, payload));
+		}
+		catch (Exception ex) {
+			throw new IllegalStateException("Unable to send event over JGroups", ex);
+		}
+	}
+
+	public void close() {
+		this.channel.close();
+	}
+
+}

bus/spring-cloud-starter-bus-jgroups/src/main/java/org/springframework/cloud/bus/jgroups/JGroupsBusProperties.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright 2015-present 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.jgroups;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+@ConfigurationProperties(JGroupsBusProperties.PREFIX)
+public class JGroupsBusProperties {
+
+	public static final String PREFIX = "spring.cloud.bus.jgroups";
+
+	private String clusterName = "spring-cloud-bus";
+
+	public String getClusterName() {
+		return this.clusterName;
+	}
+
+	public void setClusterName(String clusterName) {
+		this.clusterName = clusterName;
+	}
+
+}

bus/spring-cloud-starter-bus-jgroups/src/main/java/org/springframework/cloud/bus/jgroups/JGroupsChannelFactory.java

@@ -0,0 +1,27 @@
+/*
+ * Copyright 2015-present 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.jgroups;
+
+import org.jgroups.JChannel;
+
+class JGroupsChannelFactory {
+
+	JChannel create() throws Exception {
+		return new JChannel();
+	}
+
+}

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

@@ -0,0 +1 @@
+org.springframework.cloud.bus.jgroups.JGroupsBusAutoConfiguration

bus/spring-cloud-starter-bus-jgroups/src/test/java/org/springframework/cloud/bus/jgroups/JGroupsBusAutoConfigurationTests.java

@@ -0,0 +1,64 @@
+/*
+ * Copyright 2015-present 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.jgroups;
+
+import java.util.function.Consumer;
+
+import org.jgroups.JChannel;
+import org.junit.jupiter.api.Test;
+
+import tools.jackson.databind.ObjectMapper;
+import tools.jackson.databind.json.JsonMapper;
+
+import org.springframework.boot.test.context.runner.ApplicationContextRunner;
+import org.springframework.cloud.bus.BusBridge;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+
+class JGroupsBusAutoConfigurationTests {
+
+	private final ApplicationContextRunner contextRunner = new ApplicationContextRunner()
+		.withPropertyValues("spring.cloud.bus.enabled=true")
+		.withUserConfiguration(JGroupsBusAutoConfiguration.class)
+		.withBean(ObjectMapper.class, () -> JsonMapper.builder().build())
+		.withBean(Consumer.class, () -> mock(Consumer.class))
+		.withBean(JGroupsChannelFactory.class, () -> {
+			JGroupsChannelFactory factory = mock(JGroupsChannelFactory.class);
+			JChannel channel = mock(JChannel.class);
+			try {
+				doReturn(channel).when(factory).create();
+			}
+			catch (Exception ex) {
+				throw new IllegalStateException(ex);
+			}
+			return factory;
+		});
+
+	@Test
+	void createsJGroupsBusBridge() {
+		this.contextRunner.run(context -> assertThat(context).hasSingleBean(JGroupsBusBridge.class));
+	}
+
+	@Test
+	void backsOffWhenBusBridgeAlreadyExists() {
+		this.contextRunner.withBean(BusBridge.class, () -> mock(BusBridge.class))
+			.run(context -> assertThat(context).doesNotHaveBean(JGroupsBusBridge.class));
+	}
+
+}