Spring/spring-integration-samples
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INTSAMPLES-144: Barrier Sample

Browse Source 
JIRA: https://jira.spring.io/browse/INTSAMPLES-144

INTSAMPLES-144: Use o-c-a for Release

INTSAMPLES-144: Switch to Spring Boot

INTSAMPLES-144: Polishing

* Upgrade to SI-4.2, SF-4.2, Boot-1.3
* Fix: https://jira.spring.io/browse/INTSAMPLES-145
This commit is contained in:
Gary Russell
2015-08-04 13:16:15 -04:00
committed by Artem Bilan
parent dd7871a7ba
commit db3d78109f
24 changed files with 541 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files

44
basic/barrier/README.md Normal file
View File

@@ -0,0 +1,44 @@
Barrier Sample
==============
This example demonstrates the use of a process barrier component to suspend a thread until some asynchronous operation
completes. It uses an **HTTP Inbound Gateway**, splits the request, sends the splits to rabbitmq and then waits for
the publisher confirms. Finally, the results are returned to the caller.
The sample is a Spring Boot application that loads 2 contexts:
* Client - Sends **A,B,C** to the server
* Server - Web application
The server context has 3 integration flows:
```
http -> splitter -> amqp
|-> barrier
amqp(Acks) -> aggregator -> barrier (release)
qmqp(inbound) -> nullChannel
```
The last flow drains the messages and allows the auto-delete queue to be removed when the application is closed.
## Running the sample
$ gradlew :barrier:run
This will package the application and run it using the [Gradle Application Plugin](http://www.gradle.org/docs/current/userguide/application_plugin.html)
#### Using an IDE such as SpringSource Tool Suite™ (STS)
In STS (Eclipse), go to package **org.springframework.integration.samples.barrier**, right-click **Application** and select **Run as** --> **Java Application** (or Spring Boot Application).
### Output
The gateway (**client**) initiates a simple request posting "A,B,C" to the **server** and the **server** responds with the results.
You should see the following output from the server:
++++++++++++ Replied with: Result: A: ack=true, B: ack=true, C: ack=true ++++++++++++

140
basic/barrier/pom.xml Normal file
View File

@@ -0,0 +1,140 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd" xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
<modelVersion>4.0.0</modelVersion>
<parent>
<artifactId>spring-boot-starter-parent</artifactId>
<groupId>org.springframework.boot</groupId>
<version>1.2.5.RELEASE</version>
</parent>
<groupId>org.springframework.integration.samples</groupId>
<artifactId>barrier</artifactId>
<version>4.1.0.BUILD-SNAPSHOT</version>
<name>Barrier Sample</name>
<description>Barrier Sample</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>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build>
<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>
<dependencies>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-http</artifactId>
<version>4.2.0.BUILD-SNAPSHOT</version>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.11</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-test</artifactId>
<version>4.1.7.RELEASE</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.springframework.integration</groupId>
<artifactId>spring-integration-core</artifactId>
<version>4.2.0.BUILD-SNAPSHOT</version>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-integration</artifactId>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>log4j</groupId>
<artifactId>log4j</artifactId>
<version>1.2.17</version>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-amqp</artifactId>
<version>4.2.0.BUILD-SNAPSHOT</version>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>org.mockito</groupId>
<artifactId>mockito-core</artifactId>
<version>1.9.5</version>
<scope>test</scope>
</dependency>
</dependencies>
</project>

42
basic/barrier/src/main/java/org/springframework/integration/samples/barrier/AckAggregator.java Normal file
View File

@@ -0,0 +1,42 @@
/*
* Copyright 2015 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
*
* http://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.integration.samples.barrier;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.integration.aggregator.MessageGroupProcessor;
import org.springframework.integration.store.MessageGroup;
import org.springframework.messaging.Message;
/**
* @author Gary Russell
* @since 4.2
*
*/
public class AckAggregator implements MessageGroupProcessor {
@Override
public Object processMessageGroup(MessageGroup group) {
StringBuilder builder = new StringBuilder("Result: ");
for (Message<?> message : group.getMessages()) {
if (builder.length() > 8) {
builder.append(", ");
}
builder.append(message.getPayload() + ": ack=" + message.getHeaders().get(AmqpHeaders.PUBLISH_CONFIRM));
}
return builder.toString();
}
}

52
basic/barrier/src/main/java/org/springframework/integration/samples/barrier/Application.java Normal file
View File

@@ -0,0 +1,52 @@
/*
* Copyright 2015 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
*
* http://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.integration.samples.barrier;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.ImportResource;
/**
* @author Gary Russell
* @since 4.2
*/
@Configuration
@EnableAutoConfiguration
@ImportResource("/META-INF/spring/integration/server-context.xml")
public class Application {
public static void main(String[] args) throws Exception {
ConfigurableApplicationContext server = SpringApplication.run(Application.class, args);
ConfigurableApplicationContext client
= new SpringApplicationBuilder("/META-INF/spring/integration/client-context.xml")
.web(false)
.run(args);
RequestGateway requestGateway = client.getBean("requestGateway", RequestGateway.class);
String request = "A,B,C";
System.out.println("\n\n++++++++++++ Sending: " + request + " ++++++++++++\n");
String reply = requestGateway.echo(request);
System.out.println("\n\n++++++++++++ Replied with: " + reply + " ++++++++++++\n");
client.close();
server.close();
System.exit(0); // AMQP-519
}
}

28
basic/barrier/src/main/java/org/springframework/integration/samples/barrier/RequestGateway.java Normal file
View File

@@ -0,0 +1,28 @@
/*
* Copyright 2002-2012 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
*
* http://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.integration.samples.barrier;
/**
* @author Oleg Zhurakousky
* @author Gunnar Hillert
*
*/
public interface RequestGateway {
String echo(String request);
}

21
basic/barrier/src/main/resources/META-INF/spring/integration/client-context.xml Normal file
View File

@@ -0,0 +1,21 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-http="http://www.springframework.org/schema/integration/http"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/integration/http http://www.springframework.org/schema/integration/http/spring-integration-http.xsd">
<int:gateway id="requestGateway"
service-interface="org.springframework.integration.samples.barrier.RequestGateway"
default-request-channel="requestChannel"/>
<int:channel id="requestChannel"/>
<int-http:outbound-gateway request-channel="requestChannel"
url="http://localhost:8080/receiveGateway"
http-method="POST"
expected-response-type="java.lang.String"/>
</beans>

84
basic/barrier/src/main/resources/META-INF/spring/integration/server-context.xml Normal file
View File

@@ -0,0 +1,84 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-http="http://www.springframework.org/schema/integration/http"
xmlns:int-amqp="http://www.springframework.org/schema/integration/amqp"
xmlns:rabbit="http://www.springframework.org/schema/rabbit"
xsi:schemaLocation="http://www.springframework.org/schema/integration/amqp http://www.springframework.org/schema/integration/amqp/spring-integration-amqp.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration/file http://www.springframework.org/schema/integration/file/spring-integration-file.xsd
http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit.xsd
http://www.springframework.org/schema/integration/http http://www.springframework.org/schema/integration/http/spring-integration-http.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd">
<int-http:inbound-gateway request-channel="receiveChannel"
path="/receiveGateway"
error-channel="errorChannel"
supported-methods="POST"/>
<int:channel id="receiveChannel" />
<int:header-enricher input-channel="receiveChannel" output-channel="processChannel">
<int:header name="ackCorrelation" expression="headers['id']" />
</int:header-enricher>
<int:publish-subscribe-channel id="processChannel" />
<int:chain input-channel="processChannel" order="1">
<int:header-filter header-names="content-type, content-length" />
<int:splitter delimiters="," />
<int-amqp:outbound-channel-adapter amqp-template="amqpTemplate"
exchange-name="barrier.sample.exchange" routing-key="barrier.sample.key"
confirm-ack-channel="confirmations"
confirm-nack-channel="confirmations"
return-channel="errorChannel"
confirm-correlation-expression="#this"/>
</int:chain>
<!-- Suspend the HTTP thread until the publisher confirms are asynchronously received -->
<int:barrier id="barrier" input-channel="processChannel" order="2"
correlation-strategy-expression="headers['ackCorrelation']"
output-channel="transform" timeout="10000" />
<int:transformer input-channel="transform" expression="payload[1]" />
<!-- Aggregate the publisher confirms and send the result to the barrier release channel -->
<int:chain input-channel="confirmations" output-channel="release">
<int:header-filter header-names="replyChannel, errorChannel" />
<int:service-activator expression="payload" /> <!-- INT-3791; use s-a to retain ack header -->
<int:aggregator>
<bean class="org.springframework.integration.samples.barrier.AckAggregator" />
</int:aggregator>
</int:chain>
<int:channel id="release" />
<int:outbound-channel-adapter channel="release" ref="barrier.handler" method="trigger" />
<!-- Consumer -> nullChannel -->
<int-amqp:inbound-channel-adapter channel="nullChannel"
queue-names="barrier.sample.queue"
connection-factory="connectionFactory" />
<!-- Infrastructure -->
<rabbit:connection-factory id="connectionFactory" host="localhost" publisher-confirms="true"
publisher-returns="true"/>
<rabbit:template id="amqpTemplate" connection-factory="connectionFactory" mandatory="true" />
<rabbit:admin connection-factory="connectionFactory" />
<rabbit:queue name="barrier.sample.queue" auto-delete="true" />
<rabbit:direct-exchange name="barrier.sample.exchange" auto-delete="true">
<rabbit:bindings>
<rabbit:binding queue="barrier.sample.queue" key="barrier.sample.key" />
</rabbit:bindings>
</rabbit:direct-exchange>
</beans>

14
basic/barrier/src/main/resources/logback.xml Normal file
View File

@@ -0,0 +1,14 @@
<configuration>
<appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
<!-- encoders are assigned the type
ch.qos.logback.classic.encoder.PatternLayoutEncoder by default -->
<encoder>
<pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>
</encoder>
</appender>
<root level="warn">
<appender-ref ref="STDOUT" />
</root>
</configuration>

38
basic/barrier/src/test/java/org/springframework/integration/samples/barrier/ApplicationTests.java Normal file
View File

@@ -0,0 +1,38 @@
package org.springframework.integration.samples.barrier;
/*
* Copyright 2015 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
*
* http://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.
*/
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.boot.test.SpringApplicationConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Gary Russell
* @since 4.2
*/
@RunWith(SpringJUnit4ClassRunner.class)
@SpringApplicationConfiguration(classes = Application.class)
public class ApplicationTests {
@Test
public void contextLoads() {
}
}

2
basic/http/README.md
View File

@@ -48,5 +48,5 @@ In STS (Eclipse), go to package **org.springframework.integration.samples.http**
The gateway (**client**) initiates a simple request posting "Hello" to the **server** and the **server** responds by appending **from the other side** to the message payload and returns. You should see the following output from the server:
INFO : org.springframework.integration.samples.http.HttpClientDemo - Replied with: Hello from the other side
++++++++++++ Replied with: Hello from the other side ++++++++++++

17
basic/http/src/main/java/org/springframework/integration/samples/http/HttpClientDemo.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2011 the original author or authors.
* Copyright 2002-2015 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.
@@ -17,24 +17,23 @@ package org.springframework.integration.samples.http;
import org.apache.log4j.Logger;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
/**
* @author Oleg Zhurakousky
* @author Gary Russell
*
*/
public class HttpClientDemo {
private static Logger logger = Logger.getLogger(HttpClientDemo.class);
/**
* @param args
*/
public static void main(String[] args) {
ApplicationContext context = new ClassPathXmlApplicationContext("/META-INF/spring/integration/http-outbound-config.xml");
ConfigurableApplicationContext context = new ClassPathXmlApplicationContext(
"/META-INF/spring/integration/http-outbound-config.xml");
RequestGateway requestGateway = context.getBean("requestGateway", RequestGateway.class);
String reply = requestGateway.echo("Hello");
logger.info("Replied with: " + reply);
String reply = requestGateway.echo("Hello");
logger.info("\n\n++++++++++++ Replied with: " + reply + " ++++++++++++\n");
}
}

5
basic/http/src/main/webapp/WEB-INF/servlet-config.xml
View File

@@ -16,5 +16,8 @@
<int:channel id="receiveChannel"/>
<int:service-activator input-channel="receiveChannel" expression="payload + ' from the other side'"/>
<int:chain input-channel="receiveChannel">
<int:header-filter header-names="content-type" />
<int:service-activator expression="payload + ' from the other side'"/>
</int:chain>
</beans>

28
basic/http/src/test/java/log4j.xml
View File

@@ -1,28 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE log4j:configuration SYSTEM "log4j.dtd">
<log4j:configuration xmlns:log4j="http://jakarta.apache.org/log4j/">
<!-- Appenders -->
<appender name="console" class="org.apache.log4j.ConsoleAppender">
<param name="Target" value="System.out" />
<layout class="org.apache.log4j.PatternLayout">
<param name="ConversionPattern" value="%d{HH:mm:ss.SSS} %-5p [%t][%c] %m%n" />
</layout>
</appender>
<!-- Loggers -->
<logger name="org.springframework">
<level value="warn" />
</logger>
<logger name="org.springframework.integration.samples">
<level value="debug" />
</logger>
<!-- Root Logger -->
<root>
<priority value="warn" />
<appender-ref ref="console" />
</root>
</log4j:configuration>

21
basic/sftp/src/test/java/org/springframework/integration/samples/sftp/EmbeddedSftpServer.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014 the original author or authors.
* Copyright 2014-2015 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.
@@ -27,9 +27,7 @@ import java.util.Collections;
import org.apache.sshd.SshServer;
import org.apache.sshd.common.NamedFactory;
import org.apache.sshd.common.file.FileSystemView;
import org.apache.sshd.common.file.nativefs.NativeFileSystemFactory;
import org.apache.sshd.common.file.nativefs.NativeFileSystemView;
import org.apache.sshd.common.file.virtualfs.VirtualFileSystemFactory;
import org.apache.sshd.common.util.Base64;
import org.apache.sshd.server.Command;
import org.apache.sshd.server.PublickeyAuthenticator;
@@ -76,20 +74,7 @@ public class EmbeddedSftpServer implements InitializingBean, SmartLifecycle {
this.server.setKeyPairProvider(new SimpleGeneratorHostKeyProvider("hostkey.ser"));
this.server.setSubsystemFactories(Collections.<NamedFactory<Command>>singletonList(new SftpSubsystem.Factory()));
final String virtualDir = new FileSystemResource("").getFile().getAbsolutePath();
this.server.setFileSystemFactory(new NativeFileSystemFactory() {
@Override
public FileSystemView createFileSystemView(org.apache.sshd.common.Session session) {
return new NativeFileSystemView(session.getUsername(), false) {
@Override
public String getVirtualUserDir() {
return virtualDir;
}
};
}
});
server.setFileSystemFactory(new VirtualFileSystemFactory(virtualDir));
}
private PublicKey decodePublicKey() throws Exception {

1
basic/sftp/src/test/resources/META-INF/spring/integration/SftpInboundReceiveSample-context.xml
View File

@@ -20,6 +20,7 @@
<property name="privateKeyPassphrase" value="${passphrase}"/>
<property name="port" value="#{serverPort}"/>
<property name="user" value="${username}"/>
<property name="allowUnknownKeys" value="true"/>
</bean>
<int-sftp:inbound-channel-adapter id="sftpInbondAdapter"

1
basic/sftp/src/test/resources/META-INF/spring/integration/SftpOutboundGatewaySample-context.xml
View File

@@ -23,6 +23,7 @@
<property name="privateKeyPassphrase" value="${passphrase}"/>
<property name="port" value="#{serverPort}"/>
<property name="user" value="${username}"/>
<property name="allowUnknownKeys" value="true"/>
</bean>
<int-sftp:outbound-gateway id="gatewayLS"

1
basic/sftp/src/test/resources/META-INF/spring/integration/SftpOutboundTransferSample-context.xml
View File

@@ -20,6 +20,7 @@
<property name="privateKeyPassphrase" value="${passphrase}"/>
<property name="port" value="#{serverPort}"/>
<property name="user" value="${username}"/>
<property name="allowUnknownKeys" value="true"/>
</bean>
<int:channel id="inputChannel"/>