Make RxJavaPlugins.reset() public

[Thorn1089]

Let me preface this by saying the RxJava library has made my life easier in just the short time I've been using it. However, there's one jagged edge that bit me today.

I'm attempting to use the RxJavaPlugins class to override the default schedulers via registerSchedulersHook. However, I want to only override the defaults for a single JUnit test. Ideally, I would call RxJavaPlugins.getInstance().reset() in my @After method...but I can't, because it is package private.

I could get at it via reflection, but it'd be easier if it was public -- unless there is a) a very good reason it is not or b) there is an obvious alternative I am missing (always a possibility!)

[benjchristensen]

I'm open to making it public. Do you want to submit the PR to make the change?

There are 2 reasons it is not public:

1. I avoid making things public until they need to be as once public it must be supported for a very long time.
2. The drawback ... resetting the plugins is not safe during application runtime and also bad code could invoke it in the middle of an application life-cycle and really break things. For example, a 3rd party transitive library that abusively calls reset() in order to always register its plugin automatically in place of anything else already registered.

Of course, if someone wants to do that bad thing we can't prevent it anyways, since all someone has to do is put a class inside the same package and invoke reset from there. So the "security drawback" isn't a strong enough reason to avoid it. And the non-safe issue is one developers will just have to understand and use it wisely. In other words, reset should never be called during normal application operation.

/cc @akarnokd @mattrjacobs @abersnaze for your thoughts on this

[Thorn1089]

Some good points. I definitely agree with "as private as possible" by default.

After looking into it some more, I see that my use case would not precisely be served by this -- even if I could re-register a different hook, the Scheduler class will have already assigned its instance fields for io, computation, etc.

For now I am solving my testing issue by injecting a Scheduler with Guice which is annotated with @Named("io"), and in the test substituting Schedulers.test() for that scheduler. I'm not sure I love that approach as the scheduler doesn't feel like a "real" dependency, and it's another thing which must be configured in the DI container... Another approach might be to sidestep the public API and swap out schedulers using something like PowerMock.

So, rambling aside -- making reset() public still would not let me tweak the default schedulers on a per-test basis. If I think of an alternative approach which doesn't seem to compromise the safety of the rest of the API I'll submit a PR.

[akarnokd]

Why don't you use operators which let you parametrize the scheduler and not rely on the defaults?

For the other hooks, you could set up a level of indirection, i.e., you register your own hook which then does allow setting/removing further hooks at will. You can even create it such a way that multiple hooks can work at the same time, but you have to implement a separate Observable-Observer and make careful decisions when these hooks return some value instead of just consuming.

[Axxiss]

As a work around for this what I have done by now is defining a RxJavaTestPlugins class on src/test/java/rx/plugins

public class RxJavaTestPlugins extends RxJavaPlugins {
    RxJavaTestPlugins() {
        super();
    }

    public static void resetPlugins(){
        getInstance().reset();
    }
}

And then invoke resetPlugins() on the tear down setup of my units test before registerSchedulersHook

If your run it on the tearDown there can be another test that uses RxJava then, hence default hook is already loaded and you can't load your custom hook.

[JakeWharton]

I very much would like something like this, too.

I do something similar to what you have but as a JUnit rule so it can be used as:

@Rule public final RxJavaPluginsResetRule pluginsReset = new RxJavaPluginsResetRule();

and it calls .reset() automatically both before and after each test.

[JakeWharton]

FYI RxAndroid shipped with a @Beta reset() method public on its RxAndroidPlugins class.

[pakerfeldt]

@JakeWharton Is there really a point of doing .reset() in your Rule? To me it doesn't seem to have the expected effect. Any subsequent call to Schedulers.* will still return the same original Scheduler even after calling .reset(). Since the hooks are only called once when the Schedulers instance is created.

I can't seem to find a valid use case for using RxJavaPlugins to register test schedulers in unit tests really. Maybe it was never intended for that anyway.

[pakerfeldt]

Here's an example of how you can accidently misuse TestScheduler to break unrelated tests.
https://github.com/pakerfeldt/rxjava-the-schedulers-pitfall

[pakerfeldt]

Actually, using a singleton TestScheduler wrapper we can more easily identify when we accidently initialize Schedulers before our hook is called (instead of just getting weird behavior in the test):
https://gist.github.com/pakerfeldt/db297764918e7d464884#file-testschedulerproxy-java

However, that would require all unit tests that happen to execute any code that would initialize Schedulers to first initialize our TestSchedulerProxy (which registers the hook). And possibly also advance / trigger the test scheduler I guess.

[mikebaum]

I ran into this exact problem and it was frustrating. I ended up creating the following TestWatcher, which is specific to verifying that there are no leaking subscriptions, but I suppose could be used more generally.

/**
 * This Verifier will assert that any subscriptions created during a test are unsubscribed after the
 * test has finished.<br>
 * 
 * NOTE: It is very important that the call {@link #initialize()} is done before any
 * {@link Observable} is created. Failing to do so, means that the {@link SusbcriptionCollector}
 * would fail to collect any subscriptions.
 */
public class RxTestCaseSubscriptionsVerifier extends TestWatcher
{
    @Override
    protected void starting( Description description )
    {
        SusbcriptionCollector subSusbcriptionCollector = SusbcriptionCollector.getInstance();

        checkState( ! subSusbcriptionCollector.isCollecting(),
                    "Cannot start collecting subscriptions while another SusbcriptionCollector is " +
                    "already collecting." );

        subSusbcriptionCollector.startCollecting();
    }

    @Override
    protected void succeeded( Description description )
    {
        SusbcriptionCollector subSusbcriptionCollector = SusbcriptionCollector.getInstance();

        checkState( subSusbcriptionCollector.isCollecting(), 
                    "Cannot assert all subscriptions are unsubscribed if the SusbcriptionCollector " +
                    "was not collecting subscriptions" );

        subSusbcriptionCollector.assertAllSubscriptionsAreUnsubscribed();
    }

    @Override
    protected void finished( Description description )
    {
        SusbcriptionCollector subSusbcriptionCollector = SusbcriptionCollector.getInstance();

        subSusbcriptionCollector.reset();

        checkState( ! subSusbcriptionCollector.isCollecting(), 
                    "The SusbcriptionCollector should not be collecting once reset" );
    }

    /**
     * See NOTE in the class documentation.
     */
    public static void initialize()
    {
        SusbcriptionCollector.getInstance();
    }

    private static class SusbcriptionCollector
    {
        private static SusbcriptionCollector INSTANCE = null;

        private final Map<Subscription, String> mSubscriptions =  Collections.synchronizedMap( Maps.newHashMap() );
        private volatile boolean mShouldCollect = false;

        private SusbcriptionCollector()
        {
            RxJavaPlugins.getInstance().registerObservableExecutionHook( createSubscriptionCollector() );
        }

        static synchronized SusbcriptionCollector getInstance()
        {
            if ( INSTANCE == null )
                INSTANCE = new SusbcriptionCollector();

            return INSTANCE;
        }

        void startCollecting()
        {
            mShouldCollect = true;
        }

        public void assertAllSubscriptionsAreUnsubscribed()
        {
            synchronized( mSubscriptions )
            {
                for ( Map.Entry<Subscription, String> entry : mSubscriptions.entrySet() )
                {
                    if ( ! entry.getKey().isUnsubscribed() )
                    {
                        fail( "There was an non-unsubscribed subscription detected " +
                              "which was created at the following place:\n" +
                              "<-------------------- start unsubscribed stack -------------------->\n" + 
                              entry.getValue() +
                              "<--------------------- end unsubscribed stack --------------------->" );
                    }
                }
            }
        }

        private void reset()
        {
            synchronized( mSubscriptions )
            {                
                mSubscriptions.clear();
                mShouldCollect = false;
            }
        }

        private boolean isCollecting()
        {
            return mShouldCollect;
        }

        private RxJavaObservableExecutionHook createSubscriptionCollector()
        {
            return new RxJavaObservableExecutionHook()
            {
                @Override
                public <T extends Object> Subscription onSubscribeReturn( Subscription subscription )
                {
                    if ( ! mShouldCollect )
                        return subscription;

                    String stackTrace = Debug.getStackTrace( new Exception( "Subscription not unsubscribed." ) );
                    mSubscriptions.put( subscription, stackTrace );
                    return subscription;
                };
            };
        }
    }
}

then in the test you need to add the following:

@Rule public final TestRule mSubscriptionsVerifier = new RxTestCaseSubscriptionsVerifier();

or in a Test Suite class you could do this if you want to use the rule on all tests.

    @BeforeClass
    public static void initializeRxVerifier()
    {
        RxTestCaseSubscriptionsVerifier.initialize();
    }

[ivacf]

I also think making reset() public would be very useful for unit testing. At the moment, there isn't a way of calling registerSchedulersHook during local unit tests that doesn't end up throwing java.lang.IllegalStateException: Another strategy was already registered

The closes I got was from the constructor of a custom test Runner but this only works for the first test case, the rest would throw the exception above. So the two option I have is catch the IllegalStateException so once my hook is registered it will live for the rest of the tests or use reflection to call reset. I'm using reflection for now but there should be a better way of achieving this.

[mikebaum]

@ivacf Not sure if you are referring to my previous comment, but it's very important that you add:

    @BeforeClass
    public static void initializeRxVerifier()
    {
        RxTestCaseSubscriptionsVerifier.initialize();
    }

to whatever class groups all you're tests. If you don't have a suite that groups tests, then it's possible this might have to be done before each test. Anyway, this was tricky and frustrating for the reasons you already mentioned.

[shivangshah]

👍
I was trying to contribute back to spring-cloud-sleuth to have distributed tracing capabilities across RxJava threadpools as well (spring-cloud/spring-cloud-sleuth#235). I am following their current Hystrix implementation strategy : https://github.com/spring-cloud/spring-cloud-sleuth/blob/master/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/hystrix/SleuthHystrixConcurrencyStrategy.java#L66
You could reset HystrixPlugins but the same cannot be done right now with RxjavaPlugins. I have to create a wrapper (as mentioned above) in the same rx.plugins folder as a workaround (testing as we speak if it's working or not).
It'd be good to not have to do this workaround.

[marcingrzejszczak]

Hi! @benjchristensen @akarnokd @mattrjacobs @abersnaze - any news on this? It doesn't seem like a big change and would certainly be helpful in Spring Cloud Sleuth's intergration with RxJava.

[akarnokd]

I don't use the plugin API so can't tell if its worth it or what consequences it holds. PR welcome.

[akarnokd]

Closing via #3820

[Alisunov]

I'm getting problem when I use:
@ClassRule public static final RxJavaPluginsResetRule pluginsReset = new RxJavaPluginsResetRule();
instead of:
@Rule public final RxJavaPluginsResetRule pluginsReset = new RxJavaPluginsResetRule();

@ClassRule causes (singleExecutionHook.get() == null) to be true in RxJavaPlugins.getSingleExecutionHook() which leads to default hook for RxJavaPlugins.

How can it be possible?
Thanks!

[afjoseph]

Not sure if its related to this issue, but just wanted to say that RxJava2 changed a few things about how to override schedulers. I found another way outlined here: http://stackoverflow.com/a/43320828/3870025