未从句柄调用Spring集成流

Question

我有以下2个组件，它们应该首先从Mongo中删除文档，然后从Elastic中删除。

主要流程：

@Component
public class DeleteDocumentFlow {

    @Autowired
    private StoreInMongoFlow storeInMongoFlow;

    @Bean
    public IntegrationFlow deleteDocument() {
        return IntegrationFlows.from(Channels.METADATA_DELETE_STATUS.name())
                .handle(storeInMongoFlow.deleteDocumentInMongo())
                .channel("deleteDocumentInES.input")
                .get();
    }
}

服务：

@Component
public class StoreInMongoFlow {
    @Bean
    public IntegrationFlow deleteDocumentInMongo() {
        return flow -> flow.
                <Metadata>handle((p, h) -> {
                    DBObject obj = BasicDBObjectBuilder.start("i", p.getId()).get();
                    DeleteResult documentEntry = this.mongoTemplate.remove(obj, "docs");
                    return documentEntry.getDeletedCount();
                })
                .log(LoggingHandler.Level.INFO, m -> "Number of documents deleted: " + m.getPayload());
    }
}

不幸的是，deleteDocumentInMongo从未被调用。我可以在日志中看到bean已正确注册。

我是做错了什么，还是你需要更多的调试信息？如果我窃听手柄，则deleteDocumentInES。执行输入，但忽略mongo流。

Answer 1

你肯定做了一些根本错误的事情。您试图将一个集成流（IntegrationFlow）视为一个服务，从句柄（）调用。这不是设计集成流的目的。有关更多信息，请参阅文档：https://docs.spring.io/spring-integration/docs/current/reference/html/dsl.html#java-数字用户线

DSL提供了一个IntegrationFlow组件来定义通道和它们之间endpoint的组合，但现在IntegrationFlow只起配置作用，在应用程序上下文中填充真正的bean，并且在运行时不使用。

如果您将逻辑声明为单独的IntegrationFlow，则无需担心句柄（）-只需使用通道（"deleteDocumentInMongo.input"）将消息从主流点发送到该MongoDB子流的第一个通道。

如果你想用Elastic做同样的操作，你应该考虑用一个PublishSubscribeChannel来发送一条消息和从这个通道开始的两个流。

由于您用log（）结束了deleteDocumentInMongo流，因此您无法得到任何回复和您的。无法访问频道（“deleteDocumentInES.input”）。

请阅读更多文档，了解什么是发布订阅、请求回复、服务激活器和流本身。