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";
}
}
}
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:
@Service
annotation off RedisMessageSubscriber
RedisConfig
like this:@Bean
public RedisMessageSubscriber redisMessageSubscriber() {
return new RedisMessageSubscriber();
}
@Bean
MessageListenerAdapter messageListener() {
return new MessageListenerAdapter(redisMessageSubscriber());
}