Intercept JMS messages

[jnizet]

I'm working on a Spring Boot app which communicates with other apps, or with itself, using JMS. In addition to the functional message payload being exchanged, some contextual information is added to the messages, as properties. For example: the identity of the user (or process) who triggered the sending of the message.

Instead of having to add this contextual information each time I send a message, and to extract this contextual information each time I receive one, I would like to do that once, in a single place (it's then stored it in a thread-local variable, and/or stored in the slf4J MDC).

The way I currently do it is by overriding methods of JmsTemplate and of DefaultMessageListenerContainer. This, however is not as elegant as I would like it: I need to provide my own configuration to provide a custom JmsTemplate, and a custom DefaultMessageListenerContainerFactory, which itself creates a custom DefaultMessageListenerContainer, instead of simply using the ones auto-configured by Spring Boot, leading to code that is more verbose than necessary, and which duplicates what Spring Boot does already (properties-based customization, etc.)

I also thought about using AOP to intercept the calls to the JmsListener-annotated methods. But then that prevents me from simply using the type of the payload for the method argument: every method must take a Message as argument, just to allow extracting the message properties inside the aspect.

Unless there is already a better way to achieve what I want, I would find it nice if I could simply add interceptors to the JmsTemplate and to the DefaultMessageListenerContainerFactory.

• the JmsTemplate interceptors would allow to preprocess the message before sending it
• the DefaultMessageListenerContainerFactory interceptors would be set on each of the created DefaultMessageListenerContainers (as the other properties are), and would allow preprocessing a message before invoking the listener, and postprocessing it after the listener is invoked, in a way similar to an MVC interceptor.

[destebanm]

Hi! @jnizet do you have an example about this?

The way I currently do it is by overriding methods of JmsTemplate and of DefaultMessageListenerContainer. This, however is not as elegant as I would like it: I need to provide my own configuration to provide a custom JmsTemplate, and a custom DefaultMessageListenerContainerFactory, which itself creates a custom DefaultMessageListenerContainer, instead of simply using the ones auto-configured by Spring Boot, leading to code that is more verbose than necessary, and which duplicates what Spring Boot does already (properties-based customization, etc.)

I have exactly the same use case, and I am trying different solutions.

Thanks!

[jnizet]

@destebanm

Here's basically what I use for the listening part. For the JmsTemplate part, we plan to use inheritance, but we currently wrap the JmsTemplate into our own class for now, so I don't have any example.

    /**
     * Provides a custom <code>DefaultJmsListenerContainerFactory</code> which does the exact same thing as the one that would be
     * auto-configured by Spring Boot, except that it creates a <code>DefaultMessageListenerContainer</code> that
     * extracts the identity stored as properties in the message (if any) and stores it in the identity holder.
     * @see {@link org.springframework.boot.autoconfigure.jms.JmsAnnotationDrivenConfiguration} for the equivalent code of the Spring Boot auto-configuration
     */
    @Bean("jmsListenerContainerFactory")
    public DefaultJmsListenerContainerFactory jmsListenerContainerFactory(ConnectionFactory connectionFactory,
                                                                          DefaultJmsListenerContainerFactoryConfigurer configurer,
                                                                          IdentityHolder identityHolder) { // this is one of our beans
        DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory() {
            @Override
            protected DefaultMessageListenerContainer createContainerInstance() {
                return new DefaultMessageListenerContainer() {
                    @Override
                    protected Message receiveMessage(MessageConsumer consumer) throws JMSException {
                        Message message = super.receiveMessage(consumer);
                        if (message != null) {
                            String identity = message.getStringProperty(Identity.JMS_PROPERTY_KEY);
                            identityHolder.setIdentity(identity)); // this stores the identity in a Thread-local variable
                        }
                        return message;
                    }
                };
            }
        };
        configurer.configure(factory, connectionFactory);
        return factory;
    }

[danieljohngomez]

I think you can use org.springframework.messaging.support.ChannelInterceptor for this. It lets you pre/post handle messages.

[destebanm]

@danieljohngomez I have tried this approach but I have not been able to get it working :_(

@jnizet thanks!!!! For the listening part I think I will use aspect approach. For the JmsTemplate part, how are you wrapping the jmsTemplate? I would like to override this method

protected void doSend(MessageProducer producer, Message message) throws JMSException {
        if (this.deliveryDelay >= 0) {
            producer.setDeliveryDelay(this.deliveryDelay);
        }
        if (isExplicitQosEnabled()) {
            producer.send(message, getDeliveryMode(), getPriority(), getTimeToLive());
        }
        else {
            producer.send(message);
        }
    }

modifying the message with my properties.

Thanks!

[jnizet]

@destebanm I just send all the messages using my own bean which itself delegates to JmsTemplate to send the message and set the appropriate message properties.

[danieljohngomez]

@destebanm Check out AbstractMessageBrokerConfiguration.brokerChannel() for a reference.

[destebanm]

Thanks @danieljohngomez, I will take a look!

[bound2]

Did you get it to work? Can't find any place where to plug in the ChannelInterceptor

[danieljohngomez]

@bound2 In my case, I have overridden these methods on org.springframework.messaging.simp.config.AbstractMessageBrokerConfiguration:

• configureClientInboundChannel(ChannelRegistration registration)
• configureClientOutboundChannel(ChannelRegistration registration)

With the ChannelRegistration, you can then do interceptors(ChannelInterceptor... interceptors).

If you want to do it on the broker channel, override AbstractMessageBrokerConfiguration.brokerChannel() then do super.brokerChannel().addInterceptor();

[bclozel]

Hi there! Sorry about the radio silence. We are considering this issue for Spring Framework 6.2 and we'd like to get some feedback on our approach.

First, if you are still interested by this feature, can you add a reaction to this comment and maybe explain the use case you're trying to implement in your application? We are asking this because this issue predates the Observability support in JMS. If your use case is now covered by the Observability support, we might not need this after all.

If we get enough valid use cases for this, we can consider the following implementation with this contract:

/**
 * Intercept a {@code Message} during a JMS operation:
 * <ul>
 *   <li>for send operations, interceptors can mutate the message before it is sent, or ignore it
 *   and prevent it from being sent altogether.
 *   <li>for receive operations, interceptors can mutate the message before it is consumed
 *   by the application, or ignore it and prevent its processing altogether.
 * </ul>
 *
 * @author Brian Clozel
 * @since 6.2.0
 */
@FunctionalInterface
public interface MessageInterceptor {

    /**
     * Intercept the given message during a JMS operation.
     * @param destination the JMS destination where the message is going to be sent, or where it was received from
     * @param message the message being intercepted
     * @return {@code true} if the message should be further processed, or {@code false} if it should be dropped
     * @throws JMSException throws by {@link Message} methods
     */
    boolean intercept(Destination destination, Message message) throws JMSException;

}

You would be able to configure "send interceptors" and "receive interceptors" on JmsTemplate, as well as "receive interceptors" for the MessageListenerContainer. An interceptor can then mutate the Message and even prevent further processing.

If we're getting enough feedback on this, we can consider it for 6.2.0-M1 and get this prototype into your hands so you can give it a try.

Thanks!

[nkonev]

Although I'm not an author of the issue - I can add an usecase.

I needed some interceptor in order to get some data from headers/properties and construct Security Context, because my legacy code heavy relied on Spring Security.

From my prospective your interface is good for it, (having in mind that Message has some getters for headers/properties)

[bclozel]

@nkonev Thanks for your feedback. Indeed, Message is a mutable instance and we chose to not go with a functional approach (like Message intercept(Message msg) throws JMSException) because the style wouldn't really fit. The proposal here does prevent wrapping the message instance, but so far this doesn't seem to be a problem.

[jnizet]

Sorry, I don't have access to the source code of the application where I had this usecase anymore, so I'm sorry I won't be able to tell if that would completely suit the needs I had back then. But I fully trust your judgement.

[Saljack]

We would really appreciate these interceptors for JMS. We would like to use them for filling and extracting security context as was already mentioned. We have also another use case and I am not sure if it fits to the current interceptors. We use Azure Service Bus JMS and this implementation is pretty crappy because it can happen if we do not send any message with a JmsTemplate for long time then the next submission of a message fails because of Azure Service Bus closed connection to them. There is no workaround how to reestablish this connection automatically and try to send the message again. See https://github.com/Azure/azure-sdk-for-java/issues/31966

So it would be nice to catch the connection exception and retry a submission again in interceptors.

[bclozel]

Delaying this until #32501 is implemented.