Block GetMapping util get message from Redis in Spring boot

Question

I'm having two services, Service A and Service B, using Redis for inter-service communication.

Service A using Spring boot.

Service B using expressjs (it seems like a consumer, consume message then publish a message notify to service A that "Hey service A, I'm done")

Service A should expose two endpoints: /api-publish and /api-consume. When a user makes a request to the /api-publish endpoint, Service A should publish a message to a Redis topic named topic1.

Service B should be set up to consume messages from topic1. Upon successful processing of a message from topic1, Service B should then publish a message to another Redis topic named topic2.

All I want to make a request to the /api-consume endpoint on Service A, the service should wait until Service B has successfully published a message to topic2. Only then should Service A return a response to the user.

Here is my Redis configuration in service A:

@Configuration
public class RedisConfig {

    @Value("${spring.data.redis.host}")
    private String redisHost;

    @Value("${spring.data.redis.port}")
    private int redisPort;

    @Bean
    public RedisConnectionFactory lettuceConnectionFactory() {
        return new LettuceConnectionFactory(new RedisStandaloneConfiguration(redisHost, redisPort));
    }

    @Bean
    public RedisTemplate<?, ?> redisTemplate() {
        RedisTemplate<?, ?> template = new RedisTemplate<>();
        template.setConnectionFactory(lettuceConnectionFactory());
        template.setHashKeySerializer(new StringRedisSerializer());
        template.setKeySerializer(new StringRedisSerializer());
        template.setHashValueSerializer(new StringRedisSerializer());
        template.setValueSerializer(new StringRedisSerializer());
        return template;
    }

    @Bean
    MessageListenerAdapter messageListener() {
        return new MessageListenerAdapter(new RedisMessageSubscriber());
    }

    @Bean
    ChannelTopic topic() {
        return new ChannelTopic("topic2");
    }

    @Bean
    RedisMessageListenerContainer redisContainer() {
        RedisMessageListenerContainer container
                = new RedisMessageListenerContainer();
        container.setConnectionFactory(lettuceConnectionFactory());
        container.addMessageListener(messageListener(), topic());
        return container;
    }

}

Here is service that get message from topic2:

@Service
public class RedisMessageSubscriber implements MessageListener {
    private final BlockingQueue<String> queue = new LinkedBlockingQueue<>();

    @Override
    public void onMessage(Message message, byte[] pattern) {
        String receivedMessage = new String(message.getBody());
        try {
            queue.put(receivedMessage);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    public String waitForMessage(long timeout, TimeUnit unit) throws InterruptedException {
        return queue.poll(timeout, unit);
    }
}

Here is a simple controller:

@RestController
public class TestController {
    @Autowired
    private RedisTemplate<String, Object> template;
    @Autowired
    private RedisMessageSubscriber subscriber;

    @GetMapping("/api/v1/public/test")
    public String test() {
        try {
            String message = subscriber.waitForMessage(5, TimeUnit.SECONDS);
            if (message != null) {
                return "Received message from Service B: " + message;
            } else {
                return "Timeout waiting for message from Service B";
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            return "Failed to receive message";
        }
    }
}

Answer 1

I suspect you have 2 instances of RedisMessageSubscriber. Your controller injects @Service annotated one, but the messages come to the:

@Bean
MessageListenerAdapter messageListener() {
    return new MessageListenerAdapter(new RedisMessageSubscriber()); // <--- this one
}

So, in order to use only one instance you have to:

• Drop the @Service annotation off RedisMessageSubscriber
• Slightly adjust the RedisConfig like this:

@Bean
public RedisMessageSubscriber redisMessageSubscriber() {
    return new RedisMessageSubscriber();
} 

@Bean
MessageListenerAdapter messageListener() {
    return new MessageListenerAdapter(redisMessageSubscriber());
}