diff --git a/README.md b/README.md index d74a0f4b..50dac281 100644 --- a/README.md +++ b/README.md @@ -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 diff --git a/advanced/dynamic-tcp-client/.gitignore b/advanced/dynamic-tcp-client/.gitignore new file mode 100644 index 00000000..57a6aeaa --- /dev/null +++ b/advanced/dynamic-tcp-client/.gitignore @@ -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/ diff --git a/advanced/dynamic-tcp-client/README.adoc b/advanced/dynamic-tcp-client/README.adoc new file mode 100644 index 00000000..8406b45a --- /dev/null +++ b/advanced/dynamic-tcp-client/README.adoc @@ -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}] +``` diff --git a/advanced/dynamic-tcp-client/pom.xml b/advanced/dynamic-tcp-client/pom.xml new file mode 100644 index 00000000..b78ea604 --- /dev/null +++ b/advanced/dynamic-tcp-client/pom.xml @@ -0,0 +1,122 @@ + + + 4.0.0 + + org.springframework.boot + spring-boot-starter-parent + 1.4.0.RELEASE + + org.springframework.integration.samples + dynamic-tcp-client + 4.3.0.BUILD-SNAPSHOT + Dynamic TCP Client + Dynamic TCP Client + http://projects.spring.io/spring-integration + + SpringIO + https://spring.io + + + + The Apache Software License, Version 2.0 + http://www.apache.org/licenses/LICENSE-2.0.txt + repo + + + + + garyrussell + Gary Russell + grussell@pivotal.io + + project lead + + + + markfisher + Mark Fisher + mfisher@pivotal.io + + project founder and lead emeritus + + + + ghillert + Gunnar Hillert + ghillert@pivotal.io + + + abilan + Artem Bilan + abilan@pivotal.io + + + + scm:git:scm:git:git://github.com/spring-projects/spring-integration-samples.git + scm:git:scm:git:ssh://git@github.com:spring-projects/spring-integration-samples.git + https://github.com/spring-projects/spring-integration-samples + + + + org.springframework.boot + spring-boot-starter-integration + compile + + + org.springframework.integration + spring-integration-ip + 4.3.2.RELEASE + compile + + + org.springframework.integration + spring-integration-java-dsl + 1.2.0.M2 + compile + + + junit + junit + 4.12 + test + + + org.hamcrest + hamcrest-all + 1.3 + test + + + org.mockito + mockito-core + 1.10.19 + test + + + org.springframework + spring-test + 4.3.2.RELEASE + test + + + org.springframework.boot + spring-boot-starter-test + test + + + + + repo.spring.io.milestone + Spring Framework Maven Milestone Repository + https://repo.spring.io/libs-milestone + + + + + + org.springframework.boot + spring-boot-maven-plugin + + + + diff --git a/advanced/dynamic-tcp-client/src/main/java/org/springframework/integration/samples/dynamictcp/DynamicTcpClientApplication.java b/advanced/dynamic-tcp-client/src/main/java/org/springframework/integration/samples/dynamictcp/DynamicTcpClientApplication.java new file mode 100644 index 00000000..236bb320 --- /dev/null +++ b/advanced/dynamic-tcp-client/src/main/java/org/springframework/integration/samples/dynamictcp/DynamicTcpClientApplication.java @@ -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 subFlows = + new LinkedHashMap(MAX_CACHED, .75f, true) { + + @Override + protected boolean removeEldestEntry(Entry eldest) { + if (size() > MAX_CACHED) { + removeSubFlow(eldest); + return true; + } + else { + return false; + } + } + + }; + + @Autowired + private ConfigurableApplicationContext context; + + @Autowired + private IntegrationFlowContext flowContext; + + @Override + protected synchronized Collection 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 eldest) { + String hostPort = eldest.getKey(); + this.flowContext.remove(hostPort + ".flow"); + ((DefaultSingletonBeanRegistry) this.context.getBeanFactory()).destroySingleton(hostPort + ".cf"); + } + + } + +} diff --git a/advanced/dynamic-tcp-client/src/main/resources/application.properties b/advanced/dynamic-tcp-client/src/main/resources/application.properties new file mode 100644 index 00000000..e69de29b diff --git a/advanced/dynamic-tcp-client/src/test/java/org/springframework/integration/samples/dynamictcp/DynamicTcpClientApplicationTests.java b/advanced/dynamic-tcp-client/src/test/java/org/springframework/integration/samples/dynamictcp/DynamicTcpClientApplicationTests.java new file mode 100644 index 00000000..b619f4d0 --- /dev/null +++ b/advanced/dynamic-tcp-client/src/test/java/org/springframework/integration/samples/dynamictcp/DynamicTcpClientApplicationTests.java @@ -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() { + } + +} diff --git a/build.gradle b/build.gradle index a7ec078d..7f72fbc8 100644 --- a/build.gradle +++ b/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 { diff --git a/gradle.properties b/gradle.properties index bc2cc5fa..22e9996e 100644 --- a/gradle.properties +++ b/gradle.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