Question about Observable.using's resourceFactory & disposeAction

[yshrsmz]

I have a resource which I want to release upon unsubscribing of the Observable, and release action must be run on the same thread which I created the resource.

So I found Observable.using and assumed that resourceFactory & disposeAction will run on a same thread which I specified with subscribeOn.

At first it seems working as I expected, but I realized that sometimes disposeAction runs on different thread than I specified with subscribeOn

public class UsingResourceSample {
    public static void main(String[] args) throws InterruptedException {
    for (int i = 0; i < 10; i++) {
      Observable.using(() -> {
            String factoryThread = "Getting @" + Thread.currentThread().toString();
            System.out.println(factoryThread);
            return factoryThread;
          },
          Observable::just,
          (factoryThread) -> System.out.println("Closing @" + Thread.currentThread().toString() + ", " + factoryThread))
          .subscribeOn(Schedulers.io())
          .subscribe()
          .unsubscribe();
    }
  }
}

If I run this code multiple times, most of the time it shows Closing @ThreadA, Getting @ThreadA, but sometimes it shows Closing @ThreadB, Getting @ThreadA.

So, here are questions:

1. is this behavior intentional?
2. is there any way to run resourceFactory & disposeAction to run on the same thread which I specified with subscribeOn?

[akarnokd]

Hi.

1. Yes. The operator doesn't deal with scheduling.
2. Not with subscribeOn; try unsubscribeOn but you need a single-threaded Scheduler as all the default ones will give you different threads most likely.

[yshrsmz]

thank you for the clarification!
will try single-thread approach, but I'm wondering if it becomes performance overhead.
Though it may not suit the design of RxJava, it would be great if there is a method for setting both subcribeOn/unsubscribeOn as same thread at once

[akarnokd]

Once upon a time our subscribeOn did unsubscribe on the same thread but it was split into unsubscribeOn; I can't remember why. @zsxwing ?

[yshrsmz]

ok, I created this simple single-threaded scheduler pool.
hope this works well

public class SingleThreadSchedulerPool {
    private final static AtomicLong COUNT = new AtomicLong(0L);
    private final static List<Scheduler> SCHEDULERS = new ArrayList<>(){{
        add(Schedulers.from(Executors.newSingleThreadExecutor(new RxThreadFactory("SingleThreadScheduler-1-"))));
        add(Schedulers.from(Executors.newSingleThreadExecutor(new RxThreadFactory("SingleThreadScheduler-2-"))));
        add(Schedulers.from(Executors.newSingleThreadExecutor(new RxThreadFactory("SingleThreadScheduler-3-"))));
    }};

    private SingleThreadSchedulerPool() {
        // no-op
    }

    public static Scheduler get() {
        long current = COUNT.getAndIncrement();
        return SCHEDULERS.get((int) (current % 3));
    }
}

[akarnokd]

I'm closing this issue due to inactivity. If you have further input on the issue, don't hesitate to reopen this issue or post a new one.