Convert InputStream into POJO

Question

I am currently using spring integration to grab a .csv file via sftp and process it and send it to kafka. I need to transform this .csv file (which comes as an input stream that I split line by line using Files.splitter() for processing.) into a POJO. I've tried using ObjectInputStream, but since the transformation comes line by line it doesn't seem to work. I was wondering if someone had a better suggestion or way to turn this .csv InputStream into a POJO or could point me in the right direction.

CODE:

    @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();
    }

DOMAIN CLASS:

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

The StreamToEmailInteractionConverter() is the class I need to implement. It currently Implements the GenericTransformer Interface so I can plug it into the transformer method in the readCsvFileFlow method. Any suggestions or help would be much appreciated. I also have sample .csv data if you would like me to attach that as well. Thanks

Answer 1

The Files.splitter() produces messages for each line in the input file with a string as a payload. So, ObjectToJsonTransformer is wrong here since the payload for each line is already a string. You may look into a JsonToObjectTransformer if that string is in a JSON format. Otherwise you need to come up with a logic converting that string into your EmailInteractionTest if that is what expected to be as a result representation of each line in your file.