Spring data stream, FTP source custom headers

Question

Scenario :

• I'm using spring data starter apps to create an FTP source.

Code :

@SpringBootApplication
@Import({ FtpSourceConfiguration.class, AWSLocalStackConfig.class })
public class FtpSourceKinesisApplication {

  public static void main(String[] args) {
    SpringApplication.run(FtpSourceKinesisApplication.class, args);
  }
}

• FtpSourceConfiguration is the standard Config from FTP Starter, AWSLocalStackConfig is only to use Kinesis binder on localstack

Todo :

• I want to add a custom header on each Message like "source:ProviderA", "source:ProviderB"
• I will have multiple FTP sources pushing messages to the same topic.
• I want to identify the message source without having to create an intermediary topic to enrich headers.

Is there a way to do it without creating the IntegrationFlow myself ?

Answer 1

Based on Oleg's comment, this is what I did :

@Component
public class SourceHeaderEnricher implements ChannelInterceptor {

  SourceHeaderEnricher (Source source) {
    ((ChannelInterceptorAware) source.output()).addInterceptor(this);
  }

  @Override
  public Message<?> preSend(Message<?> message, MessageChannel channel) {
    MessageBuilder<?> builder = MessageBuilder.fromMessage(message);
    builder.setHeader("x-custom", "hello world");
    return builder.build();
  }
}

This is working if you are implementing your own app. To do it using the bare starter apps, apparently we need to wait for the next version.

Answer 2

@SylvainM Ironically we are currently working on providing such support to all our app starters. I mean you can do it now by simply adding channel interceptor to the output channel of your Sopurce, but what's coming down the pipeline is going to be much simpler and based on Spring Cloud Functions. Watch for the blog on that.