Allow use of the returned subscription to cancel periodic scheduling

[samueltardieu]

The documentation for schedulePeriodically indicates that the returned
subscription can be used to unsubscribe from the periodic action, or to
unschedule it if it has not been scheduled yet. That was the case only
before the first action took place, and it was then impossible to
unsubscribe using the given subscription, although unsubscribing the
worker did work.

This fixes #1344.

[cloudbees-pull-request-builder]

RxJava-pull-requests #1257 SUCCESS
This pull request looks good

[akarnokd]

This will cancel the entire worker, not just the recursive action. If one uses a scheduler for general scheduling tasks (instead of an Executor) this will disrupt the worker.

This is one step better, but for the 100% correct version, a new subscription type is required.

        public Subscription schedulePeriodically(final Action0 action, long initialDelay, long period, TimeUnit unit) {
            final long periodInNanos = unit.toNanos(period);
            final long startInNanos = TimeUnit.MILLISECONDS.toNanos(now()) + unit.toNanos(initialDelay);

            final MultipleAssignmentSubscription mas = new MultipleAssignmentSubscription();
            final Action0 recursiveAction = new Action0() {
                long count = 0;
                @Override
                public void call() {
                    if (!mas.isUnsubscribed()) {
                        action.call();
                        long nextTick = startInNanos + (++count * periodInNanos);
                        mas.set(schedule(this, nextTick - TimeUnit.MILLISECONDS.toNanos(now()), TimeUnit.NANOSECONDS));
                    }
                }
            };
            mas.set(schedule(recursiveAction, initialDelay, unit));
            return mas;
        }

[samueltardieu]

Wouldn't it be correct if we just updated the test to read

if (!mas.isUnsubscribed() && !isUnsubscribed()) {

?

[akarnokd]

I thought about this double-check for a minute; if the worker is unsubscribed, it may or may not happen before or after these checks in general, and if so, the next call to schedule won't do anything anyway. I.e., such eagerness is not really necessary here.

[samueltardieu]

That is my understanding as well, although I wondered why the if (!isUnsubscribed()) was here in the actual source code.

Shall I update the PR with your proposed changes, or do you want to do it yourself?

[akarnokd]

I can't do PRs right now, so be my guest.

[samueltardieu]

This new version contains a fixed fix by @akarnokd.

[cloudbees-pull-request-builder]

RxJava-pull-requests #1258 SUCCESS
This pull request looks good

[benjchristensen]

Thank you.