Bdtjtk

I am working on a usecase where I am supposed to poll S3 -> read the stream for the content -> do some processing and upload it to another bucket rather than writing the file in my server.

I know I can achieve it using S3StreamingMessageSource in Spring aws integration but the problem I am facing is that I do not know on how to process the message stream received by polling

public class S3PollerConfigurationUsingStreaming {

    @Value("${amazonProperties.bucketName}")

    private String bucketName;



    @Value("${amazonProperties.newBucket}")

    private String newBucket;



    @Autowired

    private AmazonClientService amazonClient;



    @Bean

    @InboundChannelAdapter(value = "s3Channel", poller = @Poller(fixedDelay = "100"))

    public MessageSource<InputStream> s3InboundStreamingMessageSource() {    

        S3StreamingMessageSource messageSource = new S3StreamingMessageSource(template());

        messageSource.setRemoteDirectory(bucketName);

        messageSource.setFilter(new S3PersistentAcceptOnceFileListFilter(new SimpleMetadataStore(),

                "streaming"));      

        return messageSource;

    }



    @Bean

    @Transformer(inputChannel = "s3Channel", outputChannel = "data")

    public org.springframework.integration.transformer.Transformer transformer() {

        return new StreamTransformer();

    }



    @Bean

    public S3RemoteFileTemplate template() {

        return new S3RemoteFileTemplate(new S3SessionFactory(amazonClient.getS3Client()));

    }



    @Bean

    public PollableChannel s3Channel() {

        return new QueueChannel();

    }



    @Bean

    IntegrationFlow fileStreamingFlow() {

        return IntegrationFlows

                .from(s3InboundStreamingMessageSource(),

                        e -> e.poller(p -> p.fixedDelay(30, TimeUnit.SECONDS)))

                .handle(streamFile())

                .get();

    }



}

Can someone please help me with the code to process the stream ?

Not sure what is your problem, but I see that you have a mix of concerns. If you use messaging annotations (see @InboundChannelAdapter in your config), what is the point to use the same s3InboundStreamingMessageSource in the IntegrationFlow definition?

Anyway it looks like you have already explored for yourself a StreamTransformer. This one has a charset property to convert your InputStreamfrom the remote S3 resource to the String. Otherwise it returns a byte. Everything else is up to you what and how to do with this converted content.

Also I don't see reason to have an s3Channel as a QueueChannel, since the start of your flow is pollable anyway by the @InboundChannelAdapter.

From big height I would say we have more questions to you, than vise versa...

UPDATE

Not clear what is your idea for InputStream processing, but that is really a fact that after S3StreamingMessageSource you are going to have exactly InputStream as a payload in the next handler.

Also not sure what is your streamFile(), but it must really expect InputStream as an input from the payload of the request message.
You also can use the mentioned StreamTransformer over there:

@Bean

IntegrationFlow fileStreamingFlow() {

    return IntegrationFlows

            .from(s3InboundStreamingMessageSource(),

                    e -> e.poller(p -> p.fixedDelay(30, TimeUnit.SECONDS)))

            .transform(Transformers.fromStream("UTF-8"))

            .get();

}

And the next .handle() will be ready for String as a payload.