Spring integration flow to define integration flow for reading contents from a file and persisting it in a database.

IntegrationFlowConfiguration.java

@Configuration
public class IntegrationFlowConfiguration {

  @Autowired
  PersonRepository personRepository;

  @Bean
  public IntegrationFlow fileInputFlow() {
    return IntegrationFlows.from(
        //Setting up the inbound adapter for the flow
        Files
            .inboundAdapter(new File("/tmp/in"))
            .autoCreateDirectory(true)
            .patternFilter("*.txt"), p -> p.poller(Pollers.fixedDelay(10, TimeUnit.SECONDS)
            .errorChannel(MessageChannels.direct().get())))
        // Transform the file content to string
        .transform(Files.toStringTransformer())
        //Transform the file content to list of lines in the file
        .<String, List<String>>transform(wholeText -> Arrays.asList(wholeText.split(Pattern.quote("\n"))))
        //Split the list to a single person record line
        .split()
        //Transform each line in the file and map to a Person record
        .<String, Person>transform(eachPersonText -> {
          List<String> tokenizedString = Arrays.asList(eachPersonText.split(Pattern.quote("|")));
          try {
            return Person.builder()
                .personId(Long.parseLong(tokenizedString.get(0).trim()))
                .personName(tokenizedString.get(1).trim())
                .personPhoneNumber(tokenizedString.get(2).trim())
                .build();
          } catch (Exception e) {
            return null;
          }
        })
        .filter(Objects::nonNull)
        // Save the record to the database.
        .handle((GenericHandler<Person>) (personRecordToSave, headers) -> personRepository
            .save(personRecordToSave))
        .log(Level.INFO)
        .get();
  }

}