Add dynamic-tcp-client Sample
Utilizing the new DSL dynamic flow registration. Polishing - forgot to save this before pushing. sync Polishing * Upgrade to Boot-1.4.1
This commit is contained in:
Gary Russell
committed by
Artem Bilan
Artem Bilan
parent
2c1a070930
commit
2e103f6f34
@@ -105,6 +105,7 @@ This category targets advanced developers who are quite familiar with Spring Int
|
||||
|
||||
* **advanced-testing-examples** - Example test cases that show advanced techniques to test Spring Integration applications
|
||||
* **dynamic-ftp** - Demonstrates one technique for sending files to dynamic destinations.
|
||||
* **dynamic-tcp-client** - Demonstrates a technique for dynamically creating TCP clients.
|
||||
|
||||
## Applications
|
||||
|
||||
|
||||
24
advanced/dynamic-tcp-client/.gitignore
vendored
Normal file
24
advanced/dynamic-tcp-client/.gitignore
vendored
Normal file
@@ -0,0 +1,24 @@
|
||||
target/
|
||||
.mvn
|
||||
mvn*
|
||||
|
||||
### STS ###
|
||||
.classpath
|
||||
.factorypath
|
||||
.project
|
||||
.settings
|
||||
.springBeans
|
||||
|
||||
### IntelliJ IDEA ###
|
||||
.idea
|
||||
*.iws
|
||||
*.iml
|
||||
*.ipr
|
||||
|
||||
### NetBeans ###
|
||||
nbproject/private/
|
||||
build/
|
||||
nbbuild/
|
||||
dist/
|
||||
nbdist/
|
||||
.nb-gradle/
|
||||
16
advanced/dynamic-tcp-client/README.adoc
Normal file
16
advanced/dynamic-tcp-client/README.adoc
Normal file
@@ -0,0 +1,16 @@
|
||||
= Dynamic TCP Client
|
||||
|
||||
Demonstrates a technique to dynamically add TCP clients on-demand, with caching and LRU removal.
|
||||
|
||||
Uses the http://https://github.com/spring-projects/spring-integration-java-dsl[Spring Integration Java DSL] Runtime flow registration feature.
|
||||
|
||||
The code starts two inbound channel adapters on ports 1234 and 5678 and sends a message to each.
|
||||
|
||||
Run from your favorite IDE, or from the command line `./gradlew :dynamic-tcp-client:run`.
|
||||
|
||||
The output messages show that each was received from a different socket...
|
||||
|
||||
```
|
||||
GenericMessage [payload=byte[3], headers={ip_tcp_remotePort=59000, ip_connectionId=localhost:59000:1234:fe482d5d-46d2-4708-bde8-afdcee6d3275, ip_localInetAddress=/127.0.0.1, ip_address=127.0.0.1, history=inOne,outputChannel, id=4c66210d-3855-28ad-833c-f6862d4263fb, ip_hostname=localhost, timestamp=1474483130778}]
|
||||
GenericMessage [payload=byte[3], headers={ip_tcp_remotePort=59001, ip_connectionId=localhost:59001:5678:e54f0ffe-83bc-40de-861f-9fa03df6e43d, ip_localInetAddress=/127.0.0.1, ip_address=127.0.0.1, history=inTwo,outputChannel, id=d6bd4319-00e1-550d-9511-3348d7fae907, ip_hostname=localhost, timestamp=1474483130784}]
|
||||
```
|
||||
122
advanced/dynamic-tcp-client/pom.xml
Normal file
122
advanced/dynamic-tcp-client/pom.xml
Normal file
@@ -0,0 +1,122 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
<parent>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-parent</artifactId>
|
||||
<version>1.4.0.RELEASE</version>
|
||||
</parent>
|
||||
<groupId>org.springframework.integration.samples</groupId>
|
||||
<artifactId>dynamic-tcp-client</artifactId>
|
||||
<version>4.3.0.BUILD-SNAPSHOT</version>
|
||||
<name>Dynamic TCP Client</name>
|
||||
<description>Dynamic TCP Client</description>
|
||||
<url>http://projects.spring.io/spring-integration</url>
|
||||
<organization>
|
||||
<name>SpringIO</name>
|
||||
<url>https://spring.io</url>
|
||||
</organization>
|
||||
<licenses>
|
||||
<license>
|
||||
<name>The Apache Software License, Version 2.0</name>
|
||||
<url>http://www.apache.org/licenses/LICENSE-2.0.txt</url>
|
||||
<distribution>repo</distribution>
|
||||
</license>
|
||||
</licenses>
|
||||
<developers>
|
||||
<developer>
|
||||
<id>garyrussell</id>
|
||||
<name>Gary Russell</name>
|
||||
<email>grussell@pivotal.io</email>
|
||||
<roles>
|
||||
<role>project lead</role>
|
||||
</roles>
|
||||
</developer>
|
||||
<developer>
|
||||
<id>markfisher</id>
|
||||
<name>Mark Fisher</name>
|
||||
<email>mfisher@pivotal.io</email>
|
||||
<roles>
|
||||
<role>project founder and lead emeritus</role>
|
||||
</roles>
|
||||
</developer>
|
||||
<developer>
|
||||
<id>ghillert</id>
|
||||
<name>Gunnar Hillert</name>
|
||||
<email>ghillert@pivotal.io</email>
|
||||
</developer>
|
||||
<developer>
|
||||
<id>abilan</id>
|
||||
<name>Artem Bilan</name>
|
||||
<email>abilan@pivotal.io</email>
|
||||
</developer>
|
||||
</developers>
|
||||
<scm>
|
||||
<connection>scm:git:scm:git:git://github.com/spring-projects/spring-integration-samples.git</connection>
|
||||
<developerConnection>scm:git:scm:git:ssh://git@github.com:spring-projects/spring-integration-samples.git</developerConnection>
|
||||
<url>https://github.com/spring-projects/spring-integration-samples</url>
|
||||
</scm>
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-integration</artifactId>
|
||||
<scope>compile</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.integration</groupId>
|
||||
<artifactId>spring-integration-ip</artifactId>
|
||||
<version>4.3.2.RELEASE</version>
|
||||
<scope>compile</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.integration</groupId>
|
||||
<artifactId>spring-integration-java-dsl</artifactId>
|
||||
<version>1.2.0.M2</version>
|
||||
<scope>compile</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>junit</groupId>
|
||||
<artifactId>junit</artifactId>
|
||||
<version>4.12</version>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.hamcrest</groupId>
|
||||
<artifactId>hamcrest-all</artifactId>
|
||||
<version>1.3</version>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.mockito</groupId>
|
||||
<artifactId>mockito-core</artifactId>
|
||||
<version>1.10.19</version>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework</groupId>
|
||||
<artifactId>spring-test</artifactId>
|
||||
<version>4.3.2.RELEASE</version>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-test</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
<repositories>
|
||||
<repository>
|
||||
<id>repo.spring.io.milestone</id>
|
||||
<name>Spring Framework Maven Milestone Repository</name>
|
||||
<url>https://repo.spring.io/libs-milestone</url>
|
||||
</repository>
|
||||
</repositories>
|
||||
<build>
|
||||
<plugins>
|
||||
<plugin>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-maven-plugin</artifactId>
|
||||
</plugin>
|
||||
</plugins>
|
||||
</build>
|
||||
</project>
|
||||
@@ -0,0 +1,162 @@
|
||||
package org.springframework.integration.samples.dynamictcp;
|
||||
|
||||
import java.util.Collection;
|
||||
import java.util.Collections;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.Map.Entry;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.beans.factory.support.DefaultSingletonBeanRegistry;
|
||||
import org.springframework.boot.SpringApplication;
|
||||
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||
import org.springframework.context.ConfigurableApplicationContext;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.integration.annotation.IntegrationComponentScan;
|
||||
import org.springframework.integration.annotation.MessagingGateway;
|
||||
import org.springframework.integration.channel.QueueChannel;
|
||||
import org.springframework.integration.config.EnableMessageHistory;
|
||||
import org.springframework.integration.dsl.IntegrationFlow;
|
||||
import org.springframework.integration.dsl.context.IntegrationFlowContext;
|
||||
import org.springframework.integration.ip.tcp.TcpReceivingChannelAdapter;
|
||||
import org.springframework.integration.ip.tcp.TcpSendingMessageHandler;
|
||||
import org.springframework.integration.ip.tcp.connection.TcpNetClientConnectionFactory;
|
||||
import org.springframework.integration.ip.tcp.connection.TcpNetServerConnectionFactory;
|
||||
import org.springframework.integration.router.AbstractMessageRouter;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.MessageChannel;
|
||||
import org.springframework.messaging.handler.annotation.Header;
|
||||
import org.springframework.util.Assert;
|
||||
|
||||
@SpringBootApplication
|
||||
@IntegrationComponentScan
|
||||
@EnableMessageHistory
|
||||
public class DynamicTcpClientApplication {
|
||||
|
||||
public static void main(String[] args) {
|
||||
ConfigurableApplicationContext context = SpringApplication.run(DynamicTcpClientApplication.class, args);
|
||||
ToTCP toTcp = context.getBean(ToTCP.class);
|
||||
toTcp.send("foo", "localhost", 1234);
|
||||
toTcp.send("foo", "localhost", 5678);
|
||||
QueueChannel outputChannel = context.getBean("outputChannel", QueueChannel.class);
|
||||
System.out.println(outputChannel.receive(10000));
|
||||
System.out.println(outputChannel.receive(10000));
|
||||
context.close();
|
||||
}
|
||||
|
||||
// Client side
|
||||
|
||||
@MessagingGateway(defaultRequestChannel = "toTcp.input")
|
||||
public interface ToTCP {
|
||||
|
||||
public void send(String data, @Header("host") String host, @Header("port") int port);
|
||||
|
||||
}
|
||||
|
||||
@Bean
|
||||
public IntegrationFlow toTcp() {
|
||||
return f -> f.route(router());
|
||||
}
|
||||
|
||||
@Bean
|
||||
public TcpRouter router() {
|
||||
return new TcpRouter();
|
||||
}
|
||||
|
||||
// Two servers
|
||||
|
||||
@Bean
|
||||
public TcpReceivingChannelAdapter inOne(TcpNetServerConnectionFactory cfOne) {
|
||||
TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter();
|
||||
adapter.setConnectionFactory(cfOne);
|
||||
adapter.setOutputChannel(outputChannel());
|
||||
return adapter;
|
||||
}
|
||||
|
||||
@Bean
|
||||
public TcpNetServerConnectionFactory cfOne() {
|
||||
return new TcpNetServerConnectionFactory(1234);
|
||||
}
|
||||
|
||||
@Bean
|
||||
public TcpReceivingChannelAdapter inTwo(TcpNetServerConnectionFactory cfTwo) {
|
||||
TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter();
|
||||
adapter.setConnectionFactory(cfTwo);
|
||||
adapter.setOutputChannel(outputChannel());
|
||||
return adapter;
|
||||
}
|
||||
|
||||
@Bean
|
||||
public TcpNetServerConnectionFactory cfTwo() {
|
||||
return new TcpNetServerConnectionFactory(5678);
|
||||
}
|
||||
|
||||
@Bean
|
||||
public QueueChannel outputChannel() {
|
||||
return new QueueChannel();
|
||||
}
|
||||
|
||||
public static class TcpRouter extends AbstractMessageRouter {
|
||||
|
||||
private final static int MAX_CACHED = 10; // When this is exceeded, we remove the LRU.
|
||||
|
||||
@SuppressWarnings("serial")
|
||||
private final LinkedHashMap<String, MessageChannel> subFlows =
|
||||
new LinkedHashMap<String, MessageChannel>(MAX_CACHED, .75f, true) {
|
||||
|
||||
@Override
|
||||
protected boolean removeEldestEntry(Entry<String, MessageChannel> eldest) {
|
||||
if (size() > MAX_CACHED) {
|
||||
removeSubFlow(eldest);
|
||||
return true;
|
||||
}
|
||||
else {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
};
|
||||
|
||||
@Autowired
|
||||
private ConfigurableApplicationContext context;
|
||||
|
||||
@Autowired
|
||||
private IntegrationFlowContext flowContext;
|
||||
|
||||
@Override
|
||||
protected synchronized Collection<MessageChannel> determineTargetChannels(Message<?> message) {
|
||||
MessageChannel channel = this.subFlows
|
||||
.get("" + message.getHeaders().get("host") + message.getHeaders().get("port"));
|
||||
if (channel == null) {
|
||||
channel = createNewSubflow(message);
|
||||
}
|
||||
return Collections.singletonList(channel);
|
||||
}
|
||||
|
||||
private MessageChannel createNewSubflow(Message<?> message) {
|
||||
String host = (String) message.getHeaders().get("host");
|
||||
Integer port = (Integer) message.getHeaders().get("port");
|
||||
Assert.state(host != null && port != null, "host and/or port header missing");
|
||||
String hostPort = host + port;
|
||||
|
||||
TcpNetClientConnectionFactory cf = new TcpNetClientConnectionFactory(host, port);
|
||||
this.context.getBeanFactory().registerSingleton(hostPort + ".cf", cf);
|
||||
this.context.getBeanFactory().initializeBean(cf, hostPort+".cf");
|
||||
cf.afterPropertiesSet();
|
||||
TcpSendingMessageHandler handler = new TcpSendingMessageHandler();
|
||||
handler.setConnectionFactory(cf);
|
||||
IntegrationFlow flow = f -> f.handle(handler);
|
||||
this.flowContext.register(hostPort + ".flow", flow);
|
||||
MessageChannel channel = this.flowContext.messagingTemplateFor(hostPort + ".flow").getDefaultDestination();
|
||||
this.subFlows.put(hostPort, channel);
|
||||
return channel;
|
||||
}
|
||||
|
||||
private void removeSubFlow(Entry<String, MessageChannel> eldest) {
|
||||
String hostPort = eldest.getKey();
|
||||
this.flowContext.remove(hostPort + ".flow");
|
||||
((DefaultSingletonBeanRegistry) this.context.getBeanFactory()).destroySingleton(hostPort + ".cf");
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,16 @@
|
||||
package org.springframework.integration.samples.dynamictcp;
|
||||
|
||||
import org.junit.Test;
|
||||
import org.junit.runner.RunWith;
|
||||
import org.springframework.boot.test.context.SpringBootTest;
|
||||
import org.springframework.test.context.junit4.SpringRunner;
|
||||
|
||||
@RunWith(SpringRunner.class)
|
||||
@SpringBootTest
|
||||
public class DynamicTcpClientApplicationTests {
|
||||
|
||||
@Test
|
||||
public void contextLoads() {
|
||||
}
|
||||
|
||||
}
|
||||
23
build.gradle
23
build.gradle
@@ -1401,6 +1401,29 @@ project('file-split-ftp') {
|
||||
}
|
||||
}
|
||||
|
||||
project('dynamic-tcp-client') {
|
||||
description = 'Dynamic TCP Client'
|
||||
|
||||
apply plugin: 'spring-boot'
|
||||
|
||||
dependencies {
|
||||
compile 'org.springframework.boot:spring-boot-starter-integration'
|
||||
compile "org.springframework.integration:spring-integration-ip:$springIntegrationVersion"
|
||||
compile "org.springframework.integration:spring-integration-java-dsl:$springIntegrationDslVersion"
|
||||
|
||||
testCompile 'org.springframework.boot:spring-boot-starter-test'
|
||||
}
|
||||
|
||||
springBoot {
|
||||
mainClass = 'org.springframework.integration.samples.dynamictcp.DynamicTcpClientApplication'
|
||||
}
|
||||
|
||||
task run(type: JavaExec) {
|
||||
main 'org.springframework.integration.samples.dynamictcp.DynamicTcpClientApplication'
|
||||
classpath = sourceSets.main.runtimeClasspath
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
sonarqube {
|
||||
properties {
|
||||
|
||||
@@ -1,3 +1,3 @@
|
||||
version=4.3.0.BUILD-SNAPSHOT
|
||||
springBootVersion=1.4.0.RELEASE
|
||||
springBootVersion=1.4.1.RELEASE
|
||||
org.gradle.daemon=true
|
||||
|
||||
Reference in New Issue
Block a user