How to complete a Multi depending on the previously received items

[samodadela]

Hi,

I'm using an MQTT service which when sent a certain message will respond with multiple responses. Each response contains some 'paging' info - the total number of expected responses and the 'id' of the current response. The responses can come out of order; e.g. get 1/3, 3/3 and 2/3.

The condition to end the Multi is to have received 3 out of 3 responses.

The thread that receives the responses is calling emitter.emit() on each item. The responses are opaque to this thread - they are parsed only later in onItem().transform(). If this thread knew this is the last response, it could just call emitter.emit() and then emitter.complete() and complete the stream.

The problem is I don't know how to complete a Multi outside this thread, within the pipeline.

At first I thought to pass the emitter into transform() - but that would mean I'd call complete() before returning an item. Probably not a good idea.

Then I though of returning a failure, but that would mean the last element would not be emitted.

Or maybe I could create a LastElementException which would contain the last element and then recover with it. Doesn't look very good, though.

Here's some code, to illustrate the problem:

  public static class Chunk {
        int currentPage;
        int totalPages;

        String data;

        Chunk(int currentPage, int totalPages, String data) {
            this.currentPage = currentPage;
            this.totalPages = totalPages;
            this.data = data;
        }
    }

    public interface Callback {
        void onResponse(Chunk c);

        void onComplete();
    }

    public static class Request {
        Callback callback;

        Request(Callback callback) {
            this.callback = callback;
        }
    }

    @Test
    public void endMultiTest() throws InterruptedException {

        List<Request> requestList = new ArrayList<>();

        Multi<Chunk> emitter = Multi.createFrom().<Chunk>emitter(
            em -> {
                Callback callback = new Callback() {
                    @Override
                    public void onResponse(Chunk c) {
                        em.emit(c);
                    }

                    @Override
                    public void onComplete() {
                        em.complete();
                    }
                };
                requestList.add(new Request(callback));
                // make network request which triggers multiple responses
            }
        );

        Thread thread = new Thread(() -> {
            while (requestList.isEmpty()) {
                ThreadSleeper.sleep(1000);
            }

            Callback callback = requestList.get(0).callback;

            // responses come in random order; we are done when we receive 3 of 3 responses
            // chunks are opaque here - they are parsed only in onItem().transform()
            callback.onResponse(new Chunk(3, 3, "c"));
            callback.onResponse(new Chunk(1, 3, "a"));
            callback.onResponse(new Chunk(2, 3, "b"));
        });
        thread.start();

        List<Integer> receivedPages = new ArrayList<>(); // todo: synchronize?
        emitter
            .onItem().transform(chunk -> {
                receivedPages.add(chunk.currentPage);
                if (receivedPages.size() == chunk.totalPages) {
                    // that's the last element!
                    // emit it and then complete the stream - how?
                }
                return chunk.data;
            })
            .collect().asList()
            .await().indefinitely();
    }

Please suggest how this can be achieved? I'm stuck.

[cescoffier]

You can use:

• an emitter (as you do)
• a combination of select() and collect() groups
• a custom operator

[samodadela]

@cescoffier The problem with the emitter is that to know that I should call complete(), I first need to parse the response in transform() . The thread that calls onResponse() and onComplete() does not know when the stream is done. It would have to parse the response first.

As regards select() and collect().. can you please show how this can be achieved? I figure I should accumulate all the responses with collect() but don't know how to stop the collection when a condition is reached. I don't see how to terminate the stream using those operators. Won't they just wait forever for the stream to terminate?

As for custom operator, I'll look into it... I guess I need one that does all of this:

• pass through items unchanged as they come
• constructs/accumulates some state
• terminates the stream when the state satisfies some condition

My current workaround is to throw a custom exception which holds a reference to the last item. I then recoverWithItem() by extracting the item from the exception. This effectively terminates the stream and returns all the items.