IntegrationFlowAdapter
该接口可以直接实现并指定为扫描组件,如下例所示:IntegrationFlow
@Component
public class MyFlow implements IntegrationFlow {
@Override
public void configure(IntegrationFlowDefinition<?> f) {
f.<String, String>transform(String::toUpperCase);
}
}
它由 选取,并在应用程序上下文中正确解析和注册。IntegrationFlowBeanPostProcessor
为了方便并获得松散耦合架构的好处,我们提供了 Base Class 实现。
它需要一个方法实现来生成 by using one of methods,如下例所示:IntegrationFlowAdapter
buildFlow()
IntegrationFlowDefinition
from()
@Component
public class MyFlowAdapter extends IntegrationFlowAdapter {
private final AtomicBoolean invoked = new AtomicBoolean();
public Instant nextExecutionTime(TriggerContext triggerContext) {
return this.invoked.getAndSet(true) ? null : Instant.now();
}
@Override
protected IntegrationFlowDefinition<?> buildFlow() {
return fromSupplier(this::messageSource,
e -> e.poller(p -> p.trigger(this::nextExecutionTime)))
.split(this)
.transform(this)
.aggregate(this)
.enrichHeaders(Collections.singletonMap("thing1", "THING1"))
.filter(this)
.handle(this)
.channel(c -> c.queue("myFlowAdapterOutput"));
}
public String messageSource() {
return "T,H,I,N,G,2";
}
@Splitter
public String[] split(String payload) {
return StringUtils.commaDelimitedListToStringArray(payload);
}
@Transformer
public String transform(String payload) {
return payload.toLowerCase();
}
@Aggregator
public String aggregate(List<String> payloads) {
return payloads.stream().collect(Collectors.joining());
}
@Filter
public boolean filter(@Header Optional<String> thing1) {
return thing1.isPresent();
}
@ServiceActivator
public String handle(String payload, @Header String thing1) {
return payload + ":" + thing1;
}
}