move from sandbox to separate project
This commit is contained in:
Spencer Gibb
15
.gitignore
vendored
Normal file
15
.gitignore
vendored
Normal file
@@ -0,0 +1,15 @@
|
||||
*~
|
||||
#*
|
||||
*#
|
||||
.#*
|
||||
.classpath
|
||||
.project
|
||||
.settings
|
||||
.springBeans
|
||||
.gradle
|
||||
build
|
||||
bin
|
||||
/target/
|
||||
*.swp
|
||||
.idea
|
||||
*.iml
|
||||
78
pom.xml
Normal file
78
pom.xml
Normal file
@@ -0,0 +1,78 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
|
||||
<groupId>org.springframework.platform</groupId>
|
||||
<artifactId>spring-platform-bus</artifactId>
|
||||
<version>1.0.0.BUILD-SNAPSHOT</version>
|
||||
<packaging>jar</packaging>
|
||||
|
||||
<name>spring-platform-bus</name>
|
||||
<description>Spring Patform Bus</description>
|
||||
|
||||
<parent>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-parent</artifactId>
|
||||
<version>1.1.5.BUILD-SNAPSHOT</version>
|
||||
<relativePath /> <!-- lookup parent from repository -->
|
||||
</parent>
|
||||
|
||||
<dependencyManagement>
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot</artifactId>
|
||||
<version>1.1.5.BUILD-SNAPSHOT</version>
|
||||
<type>pom</type>
|
||||
<scope>import</scope>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
</dependencyManagement>
|
||||
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-actuator</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-amqp</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-integration</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.integration</groupId>
|
||||
<artifactId>spring-integration-java-dsl</artifactId>
|
||||
<version>1.0.0.M2</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.integration</groupId>
|
||||
<artifactId>spring-integration-amqp</artifactId>
|
||||
<version>4.0.3.RELEASE</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.integration</groupId>
|
||||
<artifactId>spring-integration-event</artifactId>
|
||||
<version>4.0.3.RELEASE</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.platform</groupId>
|
||||
<artifactId>spring-platform-config-client</artifactId>
|
||||
<version>1.0.0.BUILD-SNAPSHOT</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.projectlombok</groupId>
|
||||
<artifactId>lombok</artifactId>
|
||||
<version>1.12.6</version>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
|
||||
<properties>
|
||||
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
|
||||
<java.version>1.7</java.version>
|
||||
</properties>
|
||||
|
||||
</project>
|
||||
@@ -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<EmbeddedServletContainerInitializedEvent> servletInitListener() {
|
||||
return new ApplicationListener<EmbeddedServletContainerInitializedEvent>() {
|
||||
@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<HeaderEnricherSpec>() {
|
||||
@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<RemoteApplicationEvent>() {
|
||||
@Override
|
||||
public boolean accept(RemoteApplicationEvent source) {
|
||||
return AmqpBusAutoConfiguration.this.isFromSelf(source);
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
@Bean
|
||||
public GenericSelector rejectMessagesFromSelf() {
|
||||
/*return new GenericSelector<Message>() {
|
||||
@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<RemoteApplicationEvent>() {
|
||||
@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();
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
}
|
||||
}
|
||||
2
src/main/resources/META-INF/spring.factories
Normal file
2
src/main/resources/META-INF/spring.factories
Normal file
@@ -0,0 +1,2 @@
|
||||
org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
|
||||
org.springframework.platform.cloudfoundry.broker.ServiceBrokerAutoConfiguration
|
||||
Reference in New Issue
Block a user