toMap - prevent multiple terminal events, support backpressure

[davidmoten]

As per discussion in #4242, if an operator maps an onNext emission to an onError emission downstream then it needs be defensive about an onCompleted being sent from upstream even if upstream has been unsubscribed.

Includes three unit tests that failed on the original code. The fix also has the side effect of enabling gc of map when the factory fails (onError was called on the child, not this so map was not set to null).

[akarnokd]

I thought it supported backpressure. Could you rewrite it by using DeferredScalarSubscription?

[davidmoten]

unrelated test failure:

rx.schedulers.ComputationSchedulerTests > testHandledErrorIsNotDeliveredToThreadHandler FAILED
    java.lang.AssertionError: Handler should not have received anything expected:<0> but was:<1>
        at org.junit.Assert.fail(Assert.java:88)
        at org.junit.Assert.failNotEquals(Assert.java:834)
        at org.junit.Assert.assertEquals(Assert.java:645)
        at rx.schedulers.SchedulerTests.testHandledErrorIsNotDeliveredToThreadHandler(SchedulerTests.java:102)
        at rx.schedulers.ComputationSchedulerTests.testHandledErrorIsNotDeliveredToThreadHandler(ComputationSchedulerTests.java:153)

[davidmoten]

I thought it supported backpressure. Could you rewrite it by using DeferredScalarSubscription?

Sure.

[davidmoten]

What I might do is fix multiple terminal emissions in OnSubscribeCollect first and make necessary done changes to DeferredScalarSubscriber in that PR. I'll follow that with the backpressure fix for OperatorToMap.

[akarnokd]

This needs fixing as well.

[davidmoten]

@akarnokd

In terms of releasing a value so can be gc'd I thought to modify DeferredScalarSubscriber in this method so that the field this.value is set to null just before the call to a.onNext():

protected final void complete(R value) {
        Subscriber<? super R> a = actual;
        for (;;) {
            int s = state.get();

            if (s == NO_REQUEST_HAS_VALUE || s == HAS_REQUEST_HAS_VALUE || a.isUnsubscribed()) {
                return;
            }
            if (s == HAS_REQUEST_NO_VALUE) {
                R v = value; // <--------------------------------
                value = null; // <--------------------------------
                a.onNext(v);
                if (!a.isUnsubscribed()) {
                    a.onCompleted();
                }
                state.lazySet(HAS_REQUEST_HAS_VALUE);
                return;
            }
            this.value = value;
            if (state.compareAndSet(NO_REQUEST_NO_VALUE, NO_REQUEST_HAS_VALUE)) {
                return;
            }
        }
    }

Is that ok?

[davidmoten]

don't review latest change, just committed so could work on it from another location

[akarnokd]

I'm not sure if it's worth it. Many operators don't really do that because it is also likely the whole chain gets forgotten and GC claims all of them on its own. If you want to make sure there is no leak, use onTerminateDetach.

[akarnokd]

Could you rebase the whole thing? Somehow, you picked up changes to master and they show up in the diff.

[codecov-io]

Current coverage is 84.39% (diff: 100%)

Merging #4251 into 1.x will increase coverage by 0.01%

@@                1.x      #4251   diff @@
==========================================
  Files           267        267          
  Lines         17460      17460          
  Methods           0          0          
  Messages          0          0          
  Branches       2660       2662     +2   
==========================================
+ Hits          14732      14735     +3   
- Misses         1865       1869     +4   
+ Partials        863        856     -7   

Powered by Codecov. Last update 0577b4c...37da430

[davidmoten]

• Rebased
• Rewrote to use DeferredScalarSubscriberSafe
• use singleton of DefaultMapFactory
• moved to OnSubscribe to save allocations
• added backpressure test

[akarnokd]

👍

[akarnokd]

A small trick: make the OnSubscribeToMap also implement Func0<Map<K, V>> and assign this to the map factory function - of course you can't use this() constructor delegation but assign to the final fields explicitly. This way, there is no need for an extra class.

[davidmoten]

Good idea, I've updated the PR.

[akarnokd]

👍