将InputStream转换为POJO

Question

我目前正在使用 spring 集成通过 sftp 获取 .csv 文件并对其进行处理并将其发送到 kafka。我需要将这个 .csv 文件（它作为输入流，我使用

Files.splitter()

逐行分割进行处理。）转换为 POJO。我尝试过使用 ObjectInputStream，但由于转换是逐行进行的，因此它似乎不起作用。我想知道是否有人有更好的建议或方法将此 .csv InputStream 转换为 POJO 或者可以为我指出正确的方向。

代码：

    @Bean
    public IntegrationFlow sftpFileTransferFlow(SessionFactory<SftpClient.DirEntry> sftpSessionFactory,
                                                IntegrationFlowProperties properties,
                                                MessageChannel inboundFilesMessageChannel) {

        return IntegrationFlow
                .from(Sftp.inboundStreamingAdapter(new RemoteFileTemplate<>(sftpSessionFactory))
                          .filter(new SftpRegexPatternFileListFilter(properties.getRemoteFilePattern()))
                          .remoteDirectory(properties.getRemoteDirectory()),
                      e -> e.id("sftpInboundAdapter")
                            .autoStartup(true)
                            .poller(Pollers.fixedRate(5000)))
                .log(LoggingHandler.Level.DEBUG, "DataSftpToKafkaIntegrationFlow",
                     "headers['file_remoteDirectory'] + + T(java.io.File).separator  + headers['file_remoteFile']")
                .channel(inboundFilesMessageChannel)
                .get();
    }

    @Bean
    public IntegrationFlow readCsvFileFlow(MessageChannel inboundFilesMessageChannel,
                                           QueueChannel kafkaPojoMessageChannel) {
        
        return IntegrationFlow.from(inboundFilesMessageChannel)
                              .split(Files.splitter()
                                          .markers(true)
                                          .charset(StandardCharsets.UTF_8)
                                          .firstLineAsHeader("myHeaders")
                                          .applySequence(true))
                              // .transform(new StreamToEmailInteractionConverter()) // TODO: Figure this out so it doesn't get put on topic as .csv
                              .transform(new ObjectToJsonTransformer())
                              .log(LoggingHandler.Level.DEBUG,
                                   "DataSftpToKafkaIntegrationFlow",
                                   m -> "Payload: " + m.getPayload())
                              .channel(kafkaPojoMessageChannel)
                              .get();
    }

域类：

@Getter
@Setter
public class EmailInteractionTest {
    @CsvBindByName(column = "email")
    private String email;
    @CsvBindByName(column = "RECIPIENT_ID")
    private String recipientId;
    @CsvBindByName(column = "ENCODED_RECIPIENT_ID")
    private String encodedRecipientId;
    @CsvBindByName(column = "contactId")
    private String contactId;
    @CsvBindByName(column = "code")
    private String code;
    @CsvBindByName(column = "messageId")
    private String messageId;
    @CsvBindByName(column = "userAgent")
    private String userAgent;
    @CsvBindByName(column = "messageName")
    private String messageName;
    @CsvBindByName(column = "mailingTemplateId")
    private String mailingTemplateId;
    @CsvBindByName(column = "subjectLine")
    private String subjectLine;
    @CsvBindByName(column = "docType")
    private String docType;
    @CsvBindByName(column = "reportId")
    private String reportId;
    @CsvBindByName(column = "sendType")
    private String sendType;
    @CsvBindByName(column = "bounceType")
    private String bounceType;
    @CsvBindByName(column = "urlDescription")
    private String urlDescription;
    @CsvBindByName(column = "clickUrl")
    private String clickUrl;
    @CsvBindByName(column = "optOutDetails")
    private String optOutDetails;
    @CsvBindByName(column = "messageGroupId")
    private String messageGroupId;
    @CsvBindByName(column = "programId")
    private String programId;
    @CsvBindByName(column = "timestamp")
    private String timestamp;
    @CsvBindByName(column = "originatedFrom")
    private String originatedFrom;
    @CsvBindByName(column = "eventId")
    private String eventId;
    @CsvBindByName(column = "externalSystemName")
    private String externalSystemName;
    @CsvBindByName(column = "externalSystemReferenceId")
    private String externalSystemReferenceId;
    @CsvBindByName(column = "trackingCode")
    private String trackingCode;
}

StreamToEmailInteractionConverter()

是我需要实现的类。它目前实现了

GenericTransformer

接口，因此我可以将其插入到

transformer

方法中的

readCsvFileFlow

方法中。任何建议或帮助将不胜感激。如果您希望我也附上该数据，我还有示例 .csv 数据。谢谢

Answer 1

Files.splitter()

为输入文件中的每一行生成消息，并以字符串作为负载。因此，这里

ObjectToJsonTransformer

是错误的，因为每行的有效负载已经是一个字符串。如果该字符串是 JSON 格式，您可以查看

JsonToObjectTransformer

。否则，您需要想出一个逻辑将该字符串转换为您的

EmailInteractionTest

（如果这是文件中每一行的预期结果表示形式）。