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

Cafe_DSL: Rework the cafe-dsl to the latest DSL

Browse Source
This commit is contained in:
Artem Bilan
2014-11-01 15:20:24 +02:00
parent 5d08c6f67e
commit d3ccc2530c
2 changed files with 44 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
build.gradle
View File

@@ -168,6 +168,7 @@ subprojects { subproject ->
hibernateValidatorVersion = '4.2.0.Final'
ftpServerVersion = '1.0.6'
flexjsonVersion = '2.0'
guavaVersion = '16.0.1'
groovyVersion = '2.3.0'
hsqldbVersion = '2.3.2'
h2Version = '1.3.175'
@@ -193,7 +194,7 @@ subprojects { subproject ->
subethasmtpVersion = '1.2'
slf4jVersion = '1.7.6'
springIntegrationVersion = '4.1.0.RC1'
springIntegrationDslVersion = '1.0.0.RC1'
springIntegrationDslVersion = '1.0.0.BUILD-SNAPSHOT'
springVersion = '4.1.1.RELEASE'
springSecurityVersion = '3.2.4.RELEASE'
springWebFlowVersion = '2.3.3.RELEASE'
@@ -551,6 +552,7 @@ project('cafe-dsl') {
compile project(":cafe-si")
compile 'org.springframework.boot:spring-boot-starter-integration'
compile "org.springframework.integration:spring-integration-core"
compile "com.google.guava:guava:$guavaVersion"
compile "org.springframework.integration:spring-integration-java-dsl:$springIntegrationDslVersion"
testCompile 'org.springframework.boot:spring-boot-starter-test'

123
dsl/cafe-dsl/src/main/java/org/springframework/integration/samples/dsl/cafe/Application.java
View File

@@ -16,8 +16,8 @@
package org.springframework.integration.samples.dsl.cafe;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
@@ -26,13 +26,10 @@ import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.DependsOn;
import org.springframework.integration.annotation.Gateway;
import org.springframework.integration.annotation.IntegrationComponentScan;
import org.springframework.integration.annotation.MessagingGateway;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.dsl.channel.MessageChannels;
import org.springframework.integration.dsl.core.Pollers;
import org.springframework.integration.samples.cafe.Delivery;
import org.springframework.integration.samples.cafe.Drink;
@@ -42,6 +39,8 @@ import org.springframework.integration.samples.cafe.OrderItem;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.integration.stream.CharacterStreamWritingMessageHandler;
import com.google.common.util.concurrent.Uninterruptibles;
/**
* @author Artem Bilan
* @since 3.0
@@ -70,7 +69,7 @@ public class Application {
@MessagingGateway
public interface Cafe {
@Gateway(requestChannel = "orders")
@Gateway(requestChannel = "orders.input")
void placeOrder(Order order);
}
@@ -79,90 +78,50 @@ public class Application {
private AtomicInteger coldDrinkCounter = new AtomicInteger();
@Bean
public Executor taskExecutor() {
return Executors.newCachedThreadPool();
}
@Bean(name = PollerMetadata.DEFAULT_POLLER)
public PollerMetadata poller() {
return Pollers.fixedDelay(1000).get();
}
@Bean
public IntegrationFlow ordersFlow() {
return IntegrationFlows.from("orders")
.<Order>split(Order::getItems, null)
.channel(MessageChannels.executor(this.taskExecutor()))
.<OrderItem, String>route(orderItem -> orderItem.isIced() ? "coldDrinks" : "hotDrinks")
.get();
}
@Bean
@DependsOn("preparedDrinksFlow")
public IntegrationFlow coldDrinksFlow() {
return IntegrationFlows.from(MessageChannels.queue("coldDrinks", 10))
.<OrderItem>handle((orderItem, h) -> {
try {
Thread.sleep(1000);
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
return null;
}
System.out.println(Thread.currentThread().getName()
+ " prepared cold drink #" + this.coldDrinkCounter.incrementAndGet() + " for order #"
+ orderItem.getOrderNumber() + ": " + orderItem);
return orderItem;
})
.channel("preparedDrinks")
.get();
}
@Bean
@DependsOn("preparedDrinksFlow")
public IntegrationFlow hotDrinksFlow() {
return IntegrationFlows.from(MessageChannels.queue("hotDrinks", 10))
.<OrderItem>handle((orderItem, h) -> {
try {
Thread.sleep(5000);
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
return null;
}
System.out.println(Thread.currentThread().getName()
+ " prepared hot drink #" + this.hotDrinkCounter.incrementAndGet() + " for order #"
+ orderItem.getOrderNumber() + ": " + orderItem);
return orderItem;
})
.channel("preparedDrinks")
.get();
}
@Bean
public IntegrationFlow preparedDrinksFlow() {
return IntegrationFlows.from("preparedDrinks")
public IntegrationFlow orders() {
return f -> f
.split(Order.class, Order::getItems)
.channel(c -> c.executor(Executors.newCachedThreadPool()))
.<OrderItem, Boolean>route(OrderItem::isIced, mapping -> mapping
.subFlowMapping("true", sf -> sf
.channel(c -> c.queue(10))
.publishSubscribeChannel(c -> c
.subscribe(s -> s.handle(m -> Uninterruptibles.sleepUninterruptibly(1, TimeUnit.SECONDS)))
.subscribe(sub -> sub
.<OrderItem, String>transform(p ->
Thread.currentThread().getName()
+ " prepared cold drink #" + this.coldDrinkCounter.incrementAndGet()
+ " for order #" + p.getOrderNumber() + ": " + p)
.handle(m -> System.out.println(m.getPayload())))))
.subFlowMapping("false", sf -> sf
.channel(c -> c.queue(10))
.publishSubscribeChannel(c -> c
.subscribe(s -> s.handle(m -> Uninterruptibles.sleepUninterruptibly(5, TimeUnit.SECONDS)))
.subscribe(sub -> sub
.<OrderItem, String>transform(p ->
Thread.currentThread().getName()
+ " prepared hot drink #" + this.hotDrinkCounter.incrementAndGet()
+ " for order #" + p.getOrderNumber() + ": " + p)
.handle(m -> System.out.println(m.getPayload()))))))
.<OrderItem, Drink>transform(orderItem ->
new Drink(orderItem.getOrderNumber(),
orderItem.getDrinkType(),
orderItem.isIced(),
orderItem.getShots())
)
.aggregate(aggregator ->
aggregator.outputProcessor(g ->
new Delivery(g.getMessages()
.stream()
.map(message -> (Drink) message.getPayload())
.collect(Collectors.toList()))
)
.correlationStrategy(m -> ((Drink) m.getPayload()).getOrderNumber())
, null)
.handle(CharacterStreamWritingMessageHandler.stdout())
.get();
new Drink(orderItem.getOrderNumber(),
orderItem.getDrinkType(),
orderItem.isIced(),
orderItem.getShots()))
.aggregate(aggregator -> aggregator
.outputProcessor(g ->
new Delivery(g.getMessages()
.stream()
.map(message -> (Drink) message.getPayload())
.collect(Collectors.toList())))
.correlationStrategy(m -> ((Drink) m.getPayload()).getOrderNumber()), null)
.handle(CharacterStreamWritingMessageHandler.stdout());
}
}