How to populate a Multi from the contents of an actively changing Queue?

[VestOfHolding]

Here's my puzzle that I'm having trouble solving: I have a BlockingQueue that I'm interacting with that is periodically being added to by a very large and expensive operation with the result of that operation. Each time a result is added to this BlockingQueue, I'd like to also publish that result through an endpoint. Then my web page can asynchronously update the page based on each result as we get it.

In general, it looks like the solution I want for this is to use Server-Sent Events. Sure, no problem. Except I'm not sure how to populate my Multi based on periodically checking if anything's been added to my BlockingQueue on another thread yet.

Is this something that's possible? Or do I need to emit the results stored in this Queue some other way? Either way, my goal is to emit each result as it comes in, rather than waiting for the potentially hundreds or thousands of results to slowly fill the Queue and returning the massive list of results at the end.

Thanks ahead of time.

[VestOfHolding]

Follow-up: I think I'm a lot closer. For now I've tried this:

private float[] attemptQueueRead() throws CustomQueueReadException {
    // Attempt to read from the queue. If it has nothing in it at the moment, throw an exception. 
    // This can happen either because the current operation isn't done yet, so there's no result yet, or everything is actually complete.
}

@GET
@Path("/queueRead")
@RestStreamElementType(MediaType.APPLICATION_JSON)
public Multi<float[]> queueRead() {
    return Multi.createBy().repeating()
            .uni(() -> Uni.createFrom()
                    .item(attemptReadQueue())
                    .onFailure(CustomQueueReadException.class).retry()
                    .withBackOff(Duration.ofMillis(100), Duration.ofSeconds(1))
                    .atMost(10)
                )
            // Hmm, but what to put in this until() statement to say "until when all the retries mentioned above fail and we still hit this point with an exception"?
            .until();
}

My logic is that it appears onFailure() is my key to doing multiple retries, which means I need to throw an exception if it didn't find anything to read. Unfortunately that leaves the predicate needed for until() in an awkward place.

Or, maybe I replace the until() with:

.indefinitely()
.onFailure(CustomQueueReadException.class)
.recoverWithCompletion();

Apologies for the kind of slow stream-of-consciousness as I work through this problem. I didn't see a discussion of someone going through this, and I'm hoping my experience and any other feedback that's provided will prove helpful to someone else in the future.

[VestOfHolding]

Alright, so this still isn't working, and I don't know why. Here's my current code:

public Multi<RestResponse<MyResponseObject>> queueRead() {
      return Multi.createBy().repeating()
              .uni(() -> Uni.createFrom()
                      .item(attemptReadQueue())
                      .onFailure(CustomQueueReadException.class)
                          .invoke(failure -> LOG.info("No data found in queue."))
                      .onItem()
                          .transform(this::buildOkResponse)
                  )
              .indefinitely()
              .onFailure(CustomQueueReadException.class)
              .recoverWithCompletion();
  }

private MyResponseObject attemptQueueRead() throws CustomQueueReadException {
   // "myBlockingQueue" is a BlockingQueue that is being populated in another function on another thread.
   if (myBlockingQueue == null) {
      throw new CustomQueueReadException("Queue doesn't exist yet.");
   }

   float[] data;
   try {
      data = myBlockingQueue.poll(2, SECONDS);
   }
   catch (InterruptedException e) {
        throw new CustomQueueReadException(e);
    }

    if (data == null) {
        throw new CustomQueueReadException("Could not read from queue.");
    }

   // Otherwise, do the rest of the function.
}

Here's the problem. I never see that No data found in queue. log, so apparently it never fails with the specified exception? I can't figure out what's going on. I see the results I want on the other end of the pipeline, so it's not like no data is going through, but for some reason I can't get it to stop when I want.

[weltonrodrigo]

Hi. I suppose you could use reactive messaging here? It would probably simplify the problem. On the component that does the expensive operation, you inject an @Channel("results") MutinyEmitter emitter and call emitter.emit(item). On the controller, you inject a @Channel("results") Multi<float[]> resultsStream.

Then:

@GET
@Path("/queueRead")
@RestStreamElementType(MediaType.APPLICATION_JSON)
public Multi<float[]> queueRead() {
    return resultsStream;
}

A problem arises in the mismatch of the imperative producer and the async consumption. Ie: the producer keeps producing when maybe no one is reading yet.

You solve that with an overflow strategy. Take a look at https://smallrye.io/smallrye-reactive-messaging/4.3.0/concepts/emitter/#overflow-management

If the messaging paradigm is too much, you can have a service (lets call it QueueService) acting as a connector between the producer (the one filling the queue) and the consumers (the controller) and use this strategy: https://smallrye.io/smallrye-mutiny/latest/guides/integrate-a-non-reactive-source/