How do I cancel a ticker

[wind57]

by far not an experienced mutiny user here, so my question is rather basic. Having some code like this (taken from https://quarkus.io/blog/mutiny-back-pressure/):

    private static Long canOnlyConsumeOneItemPerSecond(long x) {
        try {
            Thread.sleep(TimeUnit.SECONDS.toMillis(1));
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }

        return Long.valueOf(x);
    }

and

  public static void main(String[] args) throws Exception {

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

        Thread.sleep(2_000);
        System.out.println("done");
        x.cancel();
 } 

I indeed get a "Got Failure" message, but then the program "hangs". How do I cancel/stop you name it, that created ticker and free all of its resources. The above code with : Cancellable::cancel does not work. May be its a bug? May be my version matters?

Thank you.

[cescoffier]

Well, you gets the failure(sending more item than the capacity), so no need to cancel. It does not hang, it is just 'done'.

When there is no failure, cancel would stop everything. You can also use .select().first(x) to cancel after x items.

[wind57]

May be we have a different meaning of "hang". What I mean is that the program does not end: the VM is not terminated.

If I issue a jstack against that running process, I see threads that are blocking the VM to exit. Unfortunately, their names (pool-2-thread-1) do not tell me what code does that for me to dig deeper.

Do I make sense now?

[jponge]

This is because the default executor is a Executors.newCachedThreadPool() and is used by emitOn and is still active and prevents a JVM shutdown.

If you used another executor, or if you wait long enough it will start to prune internal threads and your JVM will exit.

[jponge]

Moving to a discussion, this is not an issue.

[wind57]

This is exactly what I thought too, but:

import io.smallrye.mutiny.Multi;

import java.time.Duration;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class MyExample {

    private static final ExecutorService exec = Executors.newSingleThreadExecutor();

    private static Long canOnlyConsumeOneItemPerSecond(long x) {
        try {
            Thread.sleep(TimeUnit.SECONDS.toMillis(1));
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }

        return Long.valueOf(x);
    }

    public static void main(String[] args) throws Exception {

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

        Thread.sleep(2_000);
        System.out.println("shutting down");
        exec.shutdownNow();
        boolean done = exec.awaitTermination(10, TimeUnit.SECONDS);
        System.out.println("executor terminated? " + done);

    }

}

still does not end the JVM.

[wind57]

@jponge ping you here in case you missed my last comment. thank you.

[jponge]

Try to capture the subscription with onSubscription (e.g., extract it to an AtomicReference) then you can later use it to cancel.