Dynamic Routing key on RabbitListener Annotation

Question

I need to create a queue linked to Direct Exchange for every user who has logged in to the application. The basement routing will be 'user_' + userId.

That is, every time I receive a message through the user management queue that a user is logged on. Instantiate a bean with scope 'prototype' that contains a method annotated with RabbitListener to declare its queue. To this bean, I passed the userId to be able to configure the name of the queue and routingKey. But I can not access this instance variable in the Spel expression due to a circular reference error.

Here I put the bean with which declares the queue:

@Component("usersHandler")
@Scope(value = "prototype")
public class UsersHandler {

    private static Logger logger = LoggerFactory.getLogger(UsersHandler.class);

    private Long userId;

    public UsersHandler(Long userId) {
        this.userId = userId;
    }

    public Long getUserId() {
        return userId;
    }

    public void setUserId(Long userId) {
        this.userId = userId;
    }

    @RabbitListener(bindings
            = @QueueBinding(
                    value = @Queue(
                            value = "#{'queue_'.concat(usersHandler.userId)}",
                            durable = "false",
                            autoDelete = "true",
                            arguments = {
                                @Argument(
                                        name = "x-message-ttl",
                                        value = "#{rabbitCustomProperties.directExchange.queueArguments['x-message-ttl']}",
                                        type = "java.lang.Integer"
                                )
                                ,
                                @Argument(
                                        name = "x-expires",
                                        value = "#{rabbitCustomProperties.directExchange.queueArguments['x-expires']}",
                                        type = "java.lang.Integer"
                                )
                                ,
                                @Argument(
                                        name = "x-dead-letter-exchange",
                                        value = "#{rabbitCustomProperties.directExchange.queueArguments['x-dead-letter-exchange']}",
                                        type = "java.lang.String"
                                )
                            }
                    ),
                    exchange = @Exchange(
                            value = "#{rabbitCustomProperties.directExchange.name}",
                            type = ExchangeTypes.DIRECT,
                            durable = "#{rabbitCustomProperties.directExchange.durable}",
                            autoDelete = "#{rabbitCustomProperties.directExchange.autoDelete}",
                            arguments = {
                                @Argument(
                                        name = "alternate-exchange",
                                        value = "#{rabbitCustomProperties.directExchange.arguments['alternate-exchange']}",
                                        type = "java.lang.String"
                                )
                            }
                    ),
                    key = "#{'user_'.concat(usersHandler.userId)}")
    )
    public void handleMessage(@Payload Notification notification) {
        logger.info("Notification Received : " + notification);
    }

}

This is the other bean in charge of creating as many UserHandler as users have logged in:

@Component("adminHandler")
public class AdminHandler implements UsersManadgementVisitor {

    @Autowired
    private ApplicationContext appCtx;

    private Map<Long, UsersHandler> handlers = new HashMap<Long, UsersHandler>();

    private static Logger logger = LoggerFactory.getLogger(AdminHandler.class);

    @RabbitListener(queues="#{rabbitCustomProperties.adminExchange.queues['users'].name}")
    public void handleMessage(@Payload UsersManadgementMessage message) {
        logger.info("Message -> " + message);
        message.getType().accept(this, message.getId());
    }

    @Override
    public void visitUserConnected(Long idUser) {
        logger.info("Declare new queue for user: " + idUser );
        UsersHandler userHandler = appCtx.getBean(UsersHandler.class, idUser);
        handlers.put(idUser, userHandler);
    }

    @Override
    public void visitUserDisconnected(Long idUser) {
        logger.info("Remove queue for user: " + idUser );
        handlers.remove(idUser);
    }
}

My question is this:

How can I make the variable userId available in the evaluation context of the SpEL expressions?

Answer 1

You could use a ThreadLocal and the T operator...

@SpringBootApplication
public class So43717710Application {

    public static void main(String[] args) throws Exception {
        ConfigurableApplicationContext context = SpringApplication.run(So43717710Application.class, args);
        UserHolder.setUser("someUser");
        context.getBean(Listener.class);
        UserHolder.clearUser();
        context.getBean(RabbitTemplate.class).convertAndSend("foo", "user_someUser", "bar");
        Thread.sleep(5000);
        context.close();
    }

    @Bean
    @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
    public Listener listener() {
        return new Listener();
    }


    public static class Listener {

        @RabbitListener(bindings = @QueueBinding(value = @Queue("#{'queue_' + T(com.example.UserHolder).getUser()}"),
                exchange = @Exchange(value = "foo"),
                key = "#{'user_' + T(com.example.UserHolder).getUser()}"))
        public void listen(String in) {
            System.out.println(in);
        }

    }

}

public class UserHolder {

    private static final ThreadLocal<String> user = new ThreadLocal<String>();

    public static void setUser(String userId) {
        user.set(userId);
    }

    public static String getUser() {
        return user.get();
    }

    public static void clearUser() {
        user.remove();
    }

 }

If the ThreadLocal is in a @Bean you can use a bean reference...

@SpringBootApplication
public class So43717710Application {

    public static void main(String[] args) throws Exception {
        ConfigurableApplicationContext context = SpringApplication.run(So43717710Application.class, args);
        UserHolder.setUser("someUser");
        context.getBean(Listener.class);
        UserHolder.clearUser();
        context.getBean(RabbitTemplate.class).convertAndSend("foo", "user_someUser", "bar");
        Thread.sleep(5000);
        context.close();
    }

    @Bean
    public UserHolder holder() {
        return new UserHolder();
    }

    @Bean
    @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
    public Listener listener() {
        return new Listener();
    }


    public static class Listener {

        @RabbitListener(bindings = @QueueBinding(value = @Queue("#{'queue_' + @holder.user}"),
                exchange = @Exchange(value = "foo"),
                key = "#{'user_' + @holder.user}"))
        public void listen(String in) {
            System.out.println(in);
        }

    }

}