Using this document as a reference: https://blog.rabbitmq.com/posts/2021/07/rabbitmq-streams-first-application I have created a stream in RabbitMQ and added 1 million messages to it.
try (Environment environment = Environment.builder().uri("rabbitmq-stream://guest:guest@localhost:5552")
.build()) {
environment.streamCreator().stream("tenth-application-stream").create();
Producer producer = environment.producerBuilder().stream("tenth-application-stream")
.build();
int messageCount = 1_000_000;
CountDownLatch confirmLatch = new CountDownLatch(messageCount);
IntStream.range(0, messageCount).forEach(i -> {
Message message = producer.messageBuilder().properties().creationTime(System.currentTimeMillis())
.messageId(i).messageBuilder().properties().groupId(String.valueOf(i)).messageBuilder()
.addData(String.valueOf(i).getBytes(StandardCharsets.UTF_8)).build();
producer.send(message, confirmationStatus -> confirmLatch.countDown());
});
try {
boolean done = confirmLatch.await(10, TimeUnit.MINUTES);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
} catch (StreamException e) {
System.out.println(e);
}
I'm trying to read back all of the messages in the stream by doing the following:
Consumer consumer = environment.consumerBuilder().stream("tenth-application-stream")
.offset(OffsetSpecification.first()).messageHandler((context, message) -> {
System.out.println(message.getBody().toString());
}).build();
However, this will only print up to message 499. How do i see all of the messages in the stream?
CountDownLatch
in the listener as well, to avoid exiting the program before all the expected messages made it to the application. The complete code of the consumer program would be useful.CountDownLatch
trick solved your problem. To be fair @chad-knutson had the first hunch, so I let them make an answer if they want.