How to complete a Multi.createFrom().ticks()?

[azurvii]

Multi.createFrom().ticks() is a nice utility. But I have not found a proper way to complete it / signal an end to the streaming.

I found a suggestion with .onSubscription(), which would give me a Flow.Subscription. Calling its .cancel() indeed stops ticking, however the Multi stream seems still left open. i.e. the client is still expecting new items.

Is there a similar thing to emitter.complete() for a Multi from .ticks()?

[cescoffier]

If you call cancel, the resources will be released.
If you really want a "complete" signal you need to combine ticks with select().whilst / until

[Loki-Afro]

@cescoffier could you provide an answer with an example? MultiSelect has where and when which to my understanding act as a filter, not as a way to complete the stream if certain event got emitted and also that particular event does not get send down the pipeline as far as i understood

        return Multi.createFrom()
                .ticks()
                .every(Duration.ofSeconds(2))
                .onItem()
                .transform(tick -> {
			someCondition = codeThatTakesTime();
                        if (someCondition) {
				// last message and end stream preemptively 
				complete("done!") <-----------------------
		        }
                        return "still alive, waiting";
                })
                .select()
                .first(60);

writing in a generator fashion:

return Multi.createFrom().<Boolean, String>generator(() -> true, Unchecked.function((n, emitter) -> {
                    someCondition = codeThatTakesTime();
                    if (n) {
                        emitter.emit("yaaay, still going");
                        return someCondition;
                    } else {
                        emitter.complete();
                        return false;
                    }
                }))
                .onItem().call(i -> {
                    // the complicated way of saying sleep for 2 seconds
                    return Uni.createFrom().nullItem().onItem().delayIt().by(Duration.ofSeconds(2));
                })
                .select()
                .first(10);

using this approach the code feels like it is all over the place

by further thinking and looking into more example code, what i am probably want to archive with the minimum effort possible:

polling a service every 2 seconds until some condition is met, however not the subscriber shall close the stream but the stream itself

last update on my attempt, the above mentioned code will sleep after the event got actually created which is rather stupid in my use case

here is another approach, this however needs one additional tick to reach the until condition, also meh, but good enough for my very particular use case, at least the code is somewhat readable

        AtomicBoolean atomicBoolean = new AtomicBoolean();

        return Multi.createBy()
                .repeating()
                .supplier(Unchecked.supplier(() -> getEvent()))
                .withDelay(Duration.ofSeconds(2))
                .until(e -> atomicBoolean.getAndSet(e.finalMessage()));

[wind57]

If you call cancel, the resources will be released.

I have a very simple example, even taking it from a page where you wrote it:

Multi.createFrom().ticks().every(Duration.ofMillis(10))
                .emitOn(Infrastructure.getDefaultExecutor())
                .onItem().transform(BackPressureExample::canOnlyConsumeOneItemPerSecond)
                .subscribe().with(
                        item -> System.out.println("Got item: " + item),
                        failure -> System.out.println("Got failure: " + failure)
                );

I indeed get a Got Failure message. The question is how in the world do I cancel that ticker now? because it "hangs".

And how do you : " combine ticks with select().whilst / until" if such methods are not there? A more verbose answer would be needed, but its not there :(