Custom integrations and validation for messages in @SqsListener (3.x)

[jacek-kaczmarek]

Hi,

first of all thank you for providing Spring Cloud AWS 3.0.

Context

I am a provider of a company's internal library for multiple users (hundreds) that uses @SqsListener (2.4, now moving to 3.x). We provide custom integrations (attribute passing from messages and payload checks). I would like to ask if the way I intend to integrate with SqsListener is correct / thinkable.

• I need access to the payload (e.g. to validate the object), I would like to customise argument manipulation behaviours (the method argument resolvers seem to be a good place),
• message interceptors seem to be too early for me (unserialized messages and also batching works against some of my use cases).

Questions

1. is my assumption correct that replacing the SqsListenerAnnotationBeanPostProcessor is one of the (intended) ways to customise behaviour?
2. if so, do you think this possibility can be treated as an "API" and will be retained (see code below)?
3. as you see there is some duplication of the SCAWSv3 code in my code below to achieve customisation, where I only need to customise resolvers. Perhaps there is a better way you could suggest.
4. the validator seems to be null in the basic SCAWSv3 setup (I've checked dynamically and also in code - statically (SCAWSv3 source code). I'm initialising it on my own (code below) and it seems to work but I may be missing the point.

Please advise.

Example custom integration

public class OurCustomListenerAnnotationBeanPostProcessor
        extends SqsListenerAnnotationBeanPostProcessor {

   /** duplication: code is 1:1 with SCAWSv3, except one line */
    @Override
    protected Collection<HandlerMethodArgumentResolver> createAdditionalArgumentResolvers() {
        return Arrays.asList(
                new VisibilityHandlerMethodArgumentResolver(SqsHeaders.SQS_VISIBILITY_TIMEOUT_HEADER),
                new SqsMessageMethodArgumentResolver(),
                new QueueAttributesMethodArgumentResolver(),
                validationEnabledAndDecoratedPayloadMethodArgumentResolver()); // <--- this line is the only difference from SC AWS 3
    }

    /** here we can turn on validation and customise behaviours, it works OK for us */
    protected HandlerMethodArgumentResolver validationEnabledAndDecoratedPayloadMethodArgumentResolver() {
        return new CustomFunctionalityXDecoratingResolver(
                new PayloadMethodArgumentResolver(createCompositeMessageConverter(), getBeanFactory().getBean(Validator.class)),
                getBeanFactory().getBean(CustomXBean.class));
    }
}

[tomazfernandes]

Hey @jacek-kaczmarek, nice work there.

A few comments:

1 and 2: extending SqsListenerAnnotationBeanPostProcessor is certainly a fair extension point and those methods are indeed considered part of the API and won't change on a minor version.

3: Take a look at the SqsListenerConfigurer interface, which also allows managing argument resolvers.

message interceptors seem to be too early for me (unserialized messages and also batching works against some of my use cases).

You can have deserialized messages in the Interceptor if you configure a SqsMessagingMessageConverter that can serialize it based on some information. For instance, SqsTemplate adds type information in the header that's used to deserialize payloads at that point.

You'll still receive the messages in batches though. Perhaps you can tell me a little bit more about your use case and we might be able to come up with a feature to support that in the framework.

4. I hadn't looked too deeply in the validation part due to time constraints, but I'd expect validation to work OOTB. If that's something that's not working as expected feel free to create an issue and open a PR so we can take a closer look.

Let me know if that makes sense to you.

Thanks.

[jacek-kaczmarek]

Hi @tomazfernandes,

thank you for the prompt response and sorry for delay from my side.

Ad 1 and 2: Custom post processor thank you, we will consider both 1/2 and 3 (below) then.

Ad 3. Configurer Thanks for the alternative.

I see the benefit of the configurer (3) over a custom bean post processor (1/2) that it is:

• less coupling, less code (more natural to use configurers)
• I can re-use preconfigured converters and argument resolvers from you rather than copy-paste

• with the consumer API I end up chaining calls (I want to reuse converters) and "editing" the whole collections to only enrich one element (but I understand the tradeoff: more user power but also more boilerplate)

At the end of the post I'm attaching example implementation (as it may be an interesting extension use case for you and other users). Feel free to comment.

Ad 4. Validator I have debugged the code and I do think the validator is hardcoded tonull in the createArgumentResolvers because the 1-constructor arg (null validator) is called there (also checked in debugger and checked that validation @Valid did not take place). I did not see any other occurence of the (2-arg) constructor in the Spring Cloud AWS code.

Please don't worry too much about it as I may be missing something and I do have a working solution to add a validator (below) an it works. But letting you know just-in-case.

Reference example:

    @Bean
    public SqsListenerConfigurer configurer(Validator validator, EffectStorage effectStorage) {

        return new SqsListenerConfigurer() {
            @Override
            public void configure(EndpointRegistrar registrar) {
                registrar.manageMessageConverters(
                        configuredConverters ->
                                registrar.manageMethodArgumentResolvers(
                                        argumentResolvers -> {
                                            List<HandlerMethodArgumentResolver> newResolvers = argumentResolvers.stream()
                                                    .map(resolver -> enrichIfPayloadMethodResolver(resolver, configuredConverters, validator, effectStorage))
                                                    .collect(Collectors.toList());
                                            argumentResolvers.clear(); // I understand I have to "hold-on" to the reference (edit in place)?
                                            argumentResolvers.addAll(newResolvers);
                                        }
                                )
                );
            }

            // find the right resolver, reuse converters, add validator, add decorating effect
            private static HandlerMethodArgumentResolver enrichIfPayloadMethodResolver(HandlerMethodArgumentResolver resolver, Collection<MessageConverter> configuredConverters, Validator validator, EffectStorage effectStorage) {
                return resolver instanceof PayloadMethodArgumentResolver ?
                        new EffectDecoratingResolver( 
                                new PayloadMethodArgumentResolver(new CompositeMessageConverter(configuredConverters), validator),
                                effectStorage) : resolver;
            }
        };
    }

[tomazfernandes]

Hey @jacek-kaczmarek, I'll answer this one here since it's simpler.

You're right in that currently validation does not work OOTB - I don't think Spring Framework provides a default one in its core dependencies for us to use here.

I think we could add a setValidator to the Registrar class, and use them in the bean post processor, that's how it's done in Spring Kafka.

Also, I guess we could work on the Registrar API so that it would allow a better way to manage the resolvers.

If you'd be willing to write a PR I'd be happy to review it.

Thanks.

[jacek-kaczmarek]

@tomazfernandes,

thanks for the chat and hints. We have tried this out in our repo and it seems to work.

In the next days someone from our team will create a PR, most probably @maksym-telepchuk-ocado.

Cheers.

[jacek-kaczmarek-ocado]

@tomazfernandes

we have submitted a PR a few days ago. The change works very well in all our cases.

#893

Could you please kindly take a look?