Tutorial: Producer Cache and Asynchronous Message Publishing

Follow along with the Tutorial Video.  It is part of your Hands-on Exercise.

We will start building our order processing application by:

1. Creating producer OrderProducer.java

2. Writing String messages asynchronously to topics orderBackLogChina and orderBackLogUS.  These represent incoming orders from different countries.

The above diagram is a depiction of the java files and Apache Pulsar topics in our e-commerce application.  We will continue to update this diagram throughout the course.

You can find OrderProducer.java in the AdditionalFiles folder.

OrderProducer.java extends AsyncRunner.  We are responsible for implementing setup(PulsarClient client) and workLoop().

In OrderProducer class, create the producerFactor and messageGenerator objects:

DynamicProducerFactory producerFactory;
MessageGenerator messageGenerator;

In setup(PulsarClient client), we initialize producerFactory and messageGenerator using the following:

producerFactory = new DynamicProducerFactory(client);
messageGenerator = new MessageGenerator();

In workloop(), we use messageGenerator to produce a new message and send the message asynchronously.

Use producerFactory to get an existing producer based on the topic name.  If it does not exist, a new producer will be created.  Use thenCompose to unwrap the CompletableFuture returned by getProducer.  We return the CompletableFuture returned from sendAsync().

MessageGenerator.StringMessage message = messageGenerator.nextStringMessage();
String topic = "persistent://student31/developer/orderBackLogChina";
System.out.println("Will publish message " + message.getValue() + " asynchronously to topic " + topic);

return producerFactory.getProducer(topic).thenCompose((producer) -> {
     return producer.newMessage().value(message).sendAsync();
});

If you are unfamiliar with CompletableFutures and asynchronous Java methods like thenCompose, the following is a good primer.  However, it is not required you become an expert in asynchronous Java code to complete this course.

https://www.baeldung.com/java-completablefuture