IntegrationFlowAdapter
IntegrationFlow
接口可直接实现,并指定为要扫描的组件,如下例所示:
@Component
public class MyFlow implements IntegrationFlow {
@Override
public void configure(IntegrationFlowDefinition<?> f) {
f.<String, String>transform(String::toUpperCase);
}
}
它由 IntegrationFlowBeanPostProcessor
接收,并在应用程序上下文中正确解析并注册。
为了方便并且获得松散耦合架构的优势,我们提供了 IntegrationFlowAdapter
基类实现。它需要一个 buildFlow()
方法实现来使用 from()
方法之一生成 IntegrationFlowDefinition
,如下例所示:
@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;
}
}