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;
}
}