From 274e5a86e6299fdb8207b61d450678dfa9a108ec Mon Sep 17 00:00:00 2001 From: Spencer Gibb Date: Tue, 19 Aug 2014 16:07:15 -0600 Subject: [PATCH] move from sandbox to separate project --- .gitignore | 15 ++ README.md | 1 + pom.xml | 78 ++++++++ .../bus/AmqpBusAutoConfiguration.java | 174 ++++++++++++++++++ .../platform/bus/RemoteApplicationEvent.java | 19 ++ src/main/resources/META-INF/spring.factories | 2 + 6 files changed, 289 insertions(+) create mode 100644 .gitignore create mode 100644 README.md create mode 100644 pom.xml create mode 100644 src/main/java/org/springframework/platform/bus/AmqpBusAutoConfiguration.java create mode 100644 src/main/java/org/springframework/platform/bus/RemoteApplicationEvent.java create mode 100644 src/main/resources/META-INF/spring.factories diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..5750647 --- /dev/null +++ b/.gitignore @@ -0,0 +1,15 @@ +*~ +#* +*# +.#* +.classpath +.project +.settings +.springBeans +.gradle +build +bin +/target/ +*.swp +.idea +*.iml diff --git a/README.md b/README.md new file mode 100644 index 0000000..55e6f49 --- /dev/null +++ b/README.md @@ -0,0 +1 @@ +### Spring Platform Bus diff --git a/pom.xml b/pom.xml new file mode 100644 index 0000000..e548b7e --- /dev/null +++ b/pom.xml @@ -0,0 +1,78 @@ + + + 4.0.0 + + org.springframework.platform + spring-platform-bus + 1.0.0.BUILD-SNAPSHOT + jar + + spring-platform-bus + Spring Patform Bus + + + org.springframework.boot + spring-boot-starter-parent + 1.1.5.BUILD-SNAPSHOT + + + + + + + org.springframework.boot + spring-boot + 1.1.5.BUILD-SNAPSHOT + pom + import + + + + + + + org.springframework.boot + spring-boot-starter-actuator + + + org.springframework.boot + spring-boot-starter-amqp + + + org.springframework.boot + spring-boot-starter-integration + + + org.springframework.integration + spring-integration-java-dsl + 1.0.0.M2 + + + org.springframework.integration + spring-integration-amqp + 4.0.3.RELEASE + + + org.springframework.integration + spring-integration-event + 4.0.3.RELEASE + + + org.springframework.platform + spring-platform-config-client + 1.0.0.BUILD-SNAPSHOT + + + org.projectlombok + lombok + 1.12.6 + + + + + UTF-8 + 1.7 + + + diff --git a/src/main/java/org/springframework/platform/bus/AmqpBusAutoConfiguration.java b/src/main/java/org/springframework/platform/bus/AmqpBusAutoConfiguration.java new file mode 100644 index 0000000..1bb2710 --- /dev/null +++ b/src/main/java/org/springframework/platform/bus/AmqpBusAutoConfiguration.java @@ -0,0 +1,174 @@ +package org.springframework.platform.bus; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.amqp.core.*; +import org.springframework.amqp.rabbit.connection.ConnectionFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; +import org.springframework.boot.context.embedded.EmbeddedServletContainerInitializedEvent; +import org.springframework.context.ApplicationListener; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.core.env.ConfigurableEnvironment; +import org.springframework.core.env.Environment; +import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.channel.interceptor.WireTap; +import org.springframework.integration.config.GlobalChannelInterceptor; +import org.springframework.integration.core.GenericSelector; +import org.springframework.integration.dsl.IntegrationFlow; +import org.springframework.integration.dsl.IntegrationFlows; +import org.springframework.integration.dsl.amqp.Amqp; +import org.springframework.integration.dsl.channel.MessageChannels; +import org.springframework.integration.event.inbound.ApplicationEventListeningMessageProducer; +import org.springframework.integration.event.outbound.ApplicationEventPublishingMessageHandler; +import org.springframework.integration.handler.LoggingHandler; +import org.springframework.platform.config.client.RefreshEndpoint; +import org.springframework.platform.context.restart.RestartEndpoint; + +/** + * @author Spencer Gibb + */ +@Configuration +@ConditionalOnClass(AmqpTemplate.class) +public class AmqpBusAutoConfiguration { + private static final Logger logger = LoggerFactory.getLogger(AmqpBusAutoConfiguration.class); + public static final String X_SPRING_PLATFORM_ORIGIN = "X-Spring-Platform-Origin"; + + @Autowired + private ConnectionFactory connectionFactory; + + @Autowired + private AmqpAdmin amqpAdmin; + + @Autowired + AmqpTemplate amqpTemplate; + + @Autowired + private ConfigurableEnvironment env; + + @Autowired(required = false) + private RefreshEndpoint refreshEndpoint; + + @Autowired(required = false) + private RestartEndpoint restartEndpoint; + private int port; + + //TODO: how to fail gracefully if no rabbit? + @Bean + ApplicationListener servletInitListener() { + return new ApplicationListener() { + @Override + public void onApplicationEvent(EmbeddedServletContainerInitializedEvent event) { + port = event.getEmbeddedServletContainer().getPort(); + } + }; + } + + @Bean + protected FanoutExchange platformBusExchange() { + //TODO: change to TopicExchange? + FanoutExchange exchange = new FanoutExchange("spring.platform.bus"); + amqpAdmin.declareExchange(exchange); + return exchange; + } + + @Bean + protected Queue localPlatformBusQueue() { + Queue queue = amqpAdmin.declareQueue(); + amqpAdmin.declareBinding(BindingBuilder.bind(queue).to(platformBusExchange())); + return queue; + } + + @Bean + public ApplicationEventListeningMessageProducer platformBusProducer() { + ApplicationEventListeningMessageProducer producer = new ApplicationEventListeningMessageProducer(); + producer.setEventTypes(RemoteApplicationEvent.class); + producer.setOutputChannel(new DirectChannel()); + return producer; + } + + @Bean + public IntegrationFlow platformBusOutboundFlow() { + return IntegrationFlows.from(platformBusProducer()) + /*.enrichHeaders(new ComponentConfigurer() { + @Override + public void configure(HeaderEnricherSpec headers) { + headers.header(X_SPRING_PLATFORM_ORIGIN, env.getProperty("spring.application.name")); + } + })*/ + .filter(acceptFromSelf()) + .handle(Amqp.outboundAdapter(this.amqpTemplate) + //.mappedRequestHeaders(X_SPRING_PLATFORM_ORIGIN) + .exchangeName("spring.platform.bus")) + .get(); + } + + @Bean + public GenericSelector acceptFromSelf() { + return new GenericSelector() { + @Override + public boolean accept(RemoteApplicationEvent source) { + return AmqpBusAutoConfiguration.this.isFromSelf(source); + } + }; + } + + @Bean + public GenericSelector rejectMessagesFromSelf() { + /*return new GenericSelector() { + @Override + public boolean accept(Message source) { + String appName = env.getProperty("spring.application.name"); + Object origin = source.getHeaders().get(X_SPRING_PLATFORM_ORIGIN); + // don't handle remote messages you sent! + return !origin.equals(appName); + } + };*/ + return new GenericSelector() { + @Override + public boolean accept(RemoteApplicationEvent source) { + return !AmqpBusAutoConfiguration.this.isFromSelf(source); + } + }; + } + + private boolean isFromSelf(RemoteApplicationEvent event) { + String originService = event.getOriginService(); + String appName = env.getProperty("spring.application.name"); + return originService.equals(appName); + } + + @Bean + public IntegrationFlow platformBusInboundFlow(Environment env) { + ApplicationEventPublishingMessageHandler messageHandler = new ApplicationEventPublishingMessageHandler(); + return IntegrationFlows.from(Amqp.inboundAdapter(connectionFactory, localPlatformBusQueue()) + /*.mappedRequestHeaders(X_SPRING_PLATFORM_ORIGIN)*/) + //TODO: only accept messages to all services or the particular service? should that be at the amqp level? + .filter(rejectMessagesFromSelf()) + //.channel(MessageChannels.direct().interceptor(new WireTap(wiretapChannel()))) + .handle(messageHandler) + .get(); + } + + @Bean + public DirectChannel wiretapChannel() { + return MessageChannels.direct().get(); + } + + @Bean + @GlobalChannelInterceptor(patterns = "platformBusInboundFlow*") + public WireTap wireTap() { + return new WireTap(wiretapChannel()); + } + + @Bean + public IntegrationFlow loggingFlow() { + LoggingHandler handler = new LoggingHandler("INFO"); + handler.setShouldLogFullMessage(true); + return IntegrationFlows.from(wiretapChannel()) + //.filter(rejectMessagesFromSelf()) + .handle(handler) + .get(); + } +} diff --git a/src/main/java/org/springframework/platform/bus/RemoteApplicationEvent.java b/src/main/java/org/springframework/platform/bus/RemoteApplicationEvent.java new file mode 100644 index 0000000..77dc34c --- /dev/null +++ b/src/main/java/org/springframework/platform/bus/RemoteApplicationEvent.java @@ -0,0 +1,19 @@ +package org.springframework.platform.bus; + +import lombok.Data; +import org.springframework.context.ApplicationEvent; + +/** + * @author Spencer Gibb + */ +@Data +public class RemoteApplicationEvent extends ApplicationEvent { + private final String originService; + private final String message; + + public RemoteApplicationEvent(Object source, String originService, String message) { + super(source); + this.originService = originService; + this.message = message; + } +} diff --git a/src/main/resources/META-INF/spring.factories b/src/main/resources/META-INF/spring.factories new file mode 100644 index 0000000..a8cd9d1 --- /dev/null +++ b/src/main/resources/META-INF/spring.factories @@ -0,0 +1,2 @@ +org.springframework.boot.autoconfigure.EnableAutoConfiguration=\ +org.springframework.platform.cloudfoundry.broker.ServiceBrokerAutoConfiguration \ No newline at end of file