Spring Integration - 过滤器 - 将消息发送到不同的端点

Question

我的输入是一个 csv 文件，如下所示：

USER_ID, USER_NAME, FREQUENCY, FREQUENCY_DETAIL
A123, AAA, ANNUALLY, 1-JUN
B123, BBB, INVALID_FREQUENCY, 21-JUN
C123, CCC, ANNUALLY, APR
D123, DDD, WEEKLY, 1-DEC

验证：

USER_ID -> alphanumeric
USERNAME -> alphabets only
FREQUENCY -> must be one of DAILY, WEEKLY, MONTHLY , ANNUALLY
FREQUENCY DETAIL -> Pattern \\d{1,2}-\\w{3}

我的 Bean 如下：

class UserBean {
    String userID;
    String userName;
    String frequency;
    String frequencyDetail;

    String status = "SUCCESS"; //Code generated. If any of the above fields is not valid, set as "ERROR, <field that failed>". E.g.: ERROR, FREQUENCY_DETAIL
}

我的流程如下：

1. 从文件夹中读取 csv 文件
2. 转换为 UserBean（使用 openCsv）
3. 如果任何 bean 的状态包含“ERROR”，则写入名为“errorSummaryReportChannel”的单独通道。

此通道将用户 ID + 状态写入文件。

4.对于所有 bean（status = "SUCCESS" 或 status = "ERROR"），转换为 JSON 并记录输出。

我需要帮助的地方：

第 3 步

如果 userBean 的状态为“ERROR”，则写入“errorSummaryReportChannel”。 对于所有状态，继续正常流程。

我认为必须添加输出通道和丢弃通道，但找不到任何示例。

我的代码：

@Configuration
public class CreateUserConfiguration {
    @Bean
    public IntegrationFlow createUser() {
        return IntegrationFlows.from(Files.inboundAdapter(new File(INPUT_DIR)))
                .enrichHeaders(h -> h.header("errorChannel", "exceptionChannel", true))
                .transform(csvToUserBeanTransformer, "convertCsvToUserBean")
                .split(userBeanSplitter, "splitUserBeans")
                .wireTap(flow -> flow.<PrimeAOBean>filter(primeAOBean -> primeAOBean.getStatus().equalsIgnoreCase("SUCCESS"), errorFlow -> errorFlow.discardChannel("errorSummaryReportGenerationChannel")))
                .transform(userBeanToJSONTransformer, "convertUserBeanToJSON")
                .handle(Files.outboundAdapter(new File(OUTPUT_SUCCESS_DIRECTORY)))
                .get();
    }

    @Bean
    public IntegrationFlow logErrorSummary() {
        return IntegrationFlows.from("errorSummaryReportGenerationChannel")
                .handle((p,h) -> {
                    return ((UserBean)(p)).getUserID() + "\t" + ((UserBean)(p)).getStatus();
                })
                .transform(Transformers.objectToString())
                .handle(Files.outboundAdapter(new File(OUTPUT_FAILED_REPORT_FILE_NAME)))
                .get();
    }

@Bean
public IntegrationFlow logError() {
    return IntegrationFlows.from("exceptionChannel")
            .enrichHeaders(h -> h.headerExpression("errorFileName", "payload.failedMessage.headers.fileName"))
            .transform(Transformers.toJson())
            .handle(Files.outboundAdapter(new File(generateOutputDirectory(OUTPUT_FAILED_DIRECTORY))))
            .get();
}

}

有人可以帮忙吗？

Answer 1

流中filter（或任何其他端点）的输出通道是在流中的端点之前自动创建的通道。所以，在你的情况下是这样的：

 .filter()
 .transform()

只要消息通过过滤器，就会被发送到该转换器。我们可以将其视为SUCCESS 进程。

现在让我们首先编写一个过滤函数来使这些消息正常流动！

有一个这样的方法适合你：

/**
 * Populate a {@link MessageFilter} with {@link MethodInvokingSelector}
 * for the provided {@link GenericSelector}.
 * Typically used with a Java 8 Lambda expression:
 * <pre class="code">
 * {@code
 *  .filter("World"::equals)
 * }
 * </pre>
 * Use {@link #filter(Class, GenericSelector)} if you need to access the entire
 * message.
 * @param genericSelector the {@link GenericSelector} to use.
 * @param <P> the source payload type.
 * @return the current {@link IntegrationFlowDefinition}.
 */
public <P> B filter(GenericSelector<P> genericSelector) {

因此，我们的任务是为您的 UserBean 实现 GenericSelector：

 .<UserBean>filter(userBean -> userBean.getStatus().equalsIgnoreCase("SUCCESS"))

要将所有其他 (ERROR) 消息发送到丢弃通道 (errorSummaryReportGenerationChannel)，我们需要以某种方式将 discardChannel 选项添加到过滤器中。为此，Java DSL 中有一个重载的 filter() 方法：

.<UserBean>filter(userBean -> userBean.getStatus().equalsIgnoreCase("SUCCESS"),
               endpoint -> endpoint.discardChannel("errorSummaryReportGenerationChannel"))

查看他们的 JavaDocs 了解更多信息。