How to create an EventBus Message instance?

[flowt-au]

Hi. I am quite new to Java and hence Mutiny so this might be a dumb question, however, I am stuck.

I have a Uni pipeline and in one stage I return the result of an EventBus.request() like so:

return eventBus.<MessageOutBean>request(channel, messagePacket);

so the subsequent stage gets an item of type Message<MessageOutBean>. So far, so good.

Now, in the first stage I want to conditionally make that Eventbus.request() and if I do not make it I want to send a "manufactured" Message<MessageOutBean> item to the subsequent stage. eg:

if (someCondition) {
  // Make the request as normal
  return eventBus.<MessageOutBean>request(channel, messagePacket);
} else {
  // Dont make the request but return something like this:
  Message<MessageOutBean> messageOutBean = new Message<MessageOutBean>();
  // Maybe set some properties on MessageOutBean ...
  // Return that "manufactured" Message
  return Uni.createFrom().item(messageOutBean);
}    

So, my Java newbie question is: io.vertx.mutiny.core.eventbus.Message is an interface. How can I create an instance as per the else clause above? Do I need to make a class that implements Message? Or?

Thanks,
Murray

[jponge]

You don't implement or create Message instances yourself. They are created internally by Vert.x to wrap a body and headers in a message that's sent over the event-bus.

Here's a quick JBang script you might use to see how it works with Vert.x + Mutiny:

///usr/bin/env jbang "$0" "$@" ; exit $?
//DEPS io.smallrye.reactive:smallrye-mutiny-vertx-core:2.27.0

import io.vertx.mutiny.core.Vertx;

public class Check {

    public static void main(String... args) {
        var vertx = Vertx.vertx();
        var bus = vertx.eventBus();

        bus.<String>consumer("a.b.c").handler(message -> {
            System.out.println(message.address());
            System.out.println(message.replyAddress());
            System.out.println(message.body());
            message.reply("Hello " + message.body());
        });

        // Note that this is a blocking call!
        var response = bus.<String>requestAndAwait("a.b.c", "you");
        System.out.println(response.body());
    }
}

[flowt-au]

Ahhh, yes, I see. My workaround was to send the message over the bus anyway then check in the class that received it then send back an "empty" message. What you show is the way to do that in that "locally".

Cool. Merci beaucoup, Julien.

[flowt-au]

@jponge So, I have implemented your example in my code and the requestAndAwait results in The current thread cannot be blocked: vert.x-eventloop-thread-4.

If I use just request with a subscriber, as below, I eventually get a timeout error because the request never returns. The consumer handler is printing the values, so it is hearing the request.

bus.<String>request("a.b.c", "My message string")
  .subscribe().with(res -> {
      System.out.println("test request res=" + res.toString());
  });

Timed out after waiting 30000(ms) for a reply. address: __vertx.reply.3, repliedAddress: a.b.c

I tried returning a Uni from the handler but it expects a void.

Sorry for my noob status here. Maybe I just need to go with my workaround in this situation.

Thanks,
Murray

[jponge]

Yes that's why I said it was blocking, if you block from a Vert.x event-loop (and not from the main thread as I did in the snippet) it will cause problems.

request gives you a Uni, you can perform various transformations such as ignoring the result, and eventually subscribe to trigger the processing.

The handler is called for each event bus message, the handler definition method returns void, which is normal.

[jponge]

Perhaps share a snippet and I can tell you what's still wrong.

[flowt-au]

Thanks @jponge.
I have included a simplified version of the code below. I understand this might be an edge-case and I have a successful workaround (as shown towards the end) so if you don't have time for this that is totally ok. There is a lot of code below.

Having said that, working this through with you has taught me a lot so it might be valuable for others as well.

What I was originally hoping I could do in the chain() stage was:

.chain(messagePacket2 -> {
  if (!messagePacket2.getContent().isEmpty()) {
    // content is not empty so we have data.
    // Make a request to the correct service
    // using that data to fetch the "child" data we need for the next stage.
    return eventBus.<MessageOutBean>request("bus070", messagePacket2);
  } else {
    // No data was returned. Return a messageOutBean with content.data = "[]"

    // Don't make the request but return a Message, something like this PSEUDO-code:
    Message<MessageOutBean> messageOutBean = new Message<MessageOutBean>();
    // Build the messageContent with data = "[]"
    // Set the data to the messageOutBean
    messageOutBean.setContentDataAsJson("[]");

    // Return the bean
    return Uni.createFrom().item(messageOutBean);
  }
})

As you say, it is not possible to manufacture a Message like that.
My implementation of your suggestion is below.
I solved the request() timeout issue I asked about above (my bad) and now have an issue where to make the following work I need a custom codec.

So, I stopped at that point because both your suggestion from above and my workaround below both involve making an eventBus request and my workaround code is simpler and does not need a custom codec.

Note: the following method is a generic one and the event bus requests could go to any of a number of consumers. For that reason I wanted to catch the "no data" condition in this method and not have to repeat my "no data" code in consumers.

// Take the incoming requestPacket, make the necessary event bus requests
// then return the requestPacket with an updated
// requestPacket.content.data (json string), which could be "[]".
Uni<RequestPacket> processReadChildrenByOwner(RequestPacket requestPacket) {

  // Build a message packet for the eventbus request
  // eventBus type is io.vertx.mutiny.core.eventbus.EventBus;
  MessagePacket messagePacket = new MessagePacket(moduleId, "search", "serviceRequestor");

  // Simplified and using single quotes for readability for this example:
  messagePacket.setContent("{ 'mkey': '" + requestPacket.getMkey() + "' }");

  // Create the request which the eventual subscriber will run
  return eventBus.<MessageOutBean>request("bus01F", messagePacket)
      .onItem()
      .transform(messageOutBean -> {
        // Build a message packet for the request in the next stage based on the 
        // messageOutBean we get here
        MessagePacket messagePacket2 = new MessagePacket(moduleId, "search", "serviceRequestor");

        // Process messageOutBean.body().getContent().getData()
        // and build messagePacket2 ...

        // Note: One possible valid outcome of processing this messageOutBean
        // is that messagePacket2.content is an empty string.

        return messagePacket2;
      })
      .chain(messagePacket2 -> {
        if (!messagePacket2.getContent().isEmpty()) {
          // content is not empty so we have data.
          // This code works correctly.
          // Make a request to the correct service
          // using that data to fetch the "child" data we need for the next stage.
          return eventBus.<MessageOutBean>request("bus070", messagePacket2);
        } else {

          // No data was returned. Return a messageOutBean with content.data = "[]"

          var vertx = Vertx.vertx();
          var bus = vertx.eventBus();

          // Add a consumer to the eventBus on address "a.b.c"
          bus.<MessageOutBean>consumer("a.b.c").handler(message -> {

            // Build the messageContent with data = "[]"
            MessageContent messageContent = new MessageContent();
            messageContent.setData("[]");
            messageContent.setSuccess(true);

            // Build the messageOutBean we wish to send to the next stage
            // with the "empty" messageContent.
            MessageOutBean messageOutBean = new MessageOutBean();
            messageOutBean.setContent(messageContent);

            // This throws:
            // No message codec for type: class org.flowt.shopping.messaging.MessageOutBean
            // For this to work I need a custom codec for messageOutBean
            message.reply(messageOutBean);
          });

          return bus.<MessageOutBean>request("a.b.c", "");
        }
      })
      .onItem()
      .transformToUni(messageOutBean -> {
        try {
          // Add the final data to the requestPacket that was passed into this method
          requestPacket.setResultJson(messageOutBean.body().getContent().getData());

          // Flag this packet as successful
          requestPacket.setSuccess(true);

          // Return the updated requestPacket as a Uni<RequestPacket>
          // for the eventual subscriber.
          return Uni.createFrom().item(requestPacket);
        } catch (Exception e) {
          return Uni.createFrom().failure(e);
        }
      });
}

My workaround is to create a fallback consumer class and make the request there instead.
That way I have a consistent flow and the "no data" code is in just one place rather than being repeated in each possible consumer service.

So my chain() stage looks like:

.chain(messagePacket2 -> {
  if (!messagePacket2.getContent().isEmpty()) {
    // content is not empty so we have data.

    // Make a request to the correct service
    // using that data to fetch the "child" data we need for the next stage.
    return eventBus.<MessageOutBean>request("bus070", messagePacket2);
  } else {
    // No data. Make a request to the fallback service
    // which returns a MessageOutBean Uni with content.data = "[]".
    return eventBus.<MessageOutBean>request("fallback", messagePacket2);
  }
})

As I was doing this I can see optimisations I can make above but just for completeness I wanted to close the loop on the original topic about manufacturing Messages and one way to handle that "use-case" since manufacturing is not possible.

Thanks again Julien, I appreciate your responses and help.

Go well,
Murray