[v7] share throws an exception when using observables with circular references

[voliva]

Bug Report

Current Behavior
When you have an observable with circular references, something like:

const deferredCount = defer(() => source).pipe(
   startWith(0)
);

const source: Observable<string> = from([1, 2, 3]).pipe(
  withLatestFrom(deferredCount),
  map(([a, b]) => String(Number(a) + Number(b))),
  share()
);

(Sorry for the contrived example, this could be easily done with a scan, but shows the essential shape of a circular reference in observables)

Share will throw an error. I've seen two different errors depending on the environment I'm running this test, but both of them look similar. When testing with rxjs-marbles:

    TypeError: Cannot read property 'complete' of null

      at Object.complete (node_modules/rxjs/src/internal/operators/share.ts:138:10)
      at Object.complete (node_modules/rxjs/src/internal/Subscriber.ts:188:14)
      at SafeSubscriber.Object.<anonymous>.Subscriber._complete (node_modules/rxjs/src/internal/Subscriber.ts:126:24)
      at SafeSubscriber.Object.<anonymous>.Subscriber.complete (node_modules/rxjs/src/internal/Subscriber.ts:100:12)
      at OperatorSubscriber.Object.<anonymous>.Subscriber._complete (node_modules/rxjs/src/internal/Subscriber.ts:126:24)
      at OperatorSubscriber.Object.<anonymous>.Subscriber.complete (node_modules/rxjs/src/internal/Subscriber.ts:100:12)
      at OperatorSubscriber.Object.<anonymous>.Subscriber._complete (node_modules/rxjs/src/internal/Subscriber.ts:126:24)
      at OperatorSubscriber.Object.<anonymous>.Subscriber.complete (node_modules/rxjs/src/internal/Subscriber.ts:100:12)
      at Object.observeNotification (node_modules/rxjs/src/internal/Notification.ts:238:102)
      at VirtualAction.subscriber.add.scheduler.schedule.message (node_modules/rxjs/src/internal/testing/ColdObservable.ts:43:13)
      at VirtualAction.Object.<anonymous>.AsyncAction._execute (node_modules/rxjs/src/internal/scheduler/AsyncAction.ts:116:12)
      at VirtualAction.Object.<anonymous>.VirtualAction._execute (node_modules/rxjs/src/internal/scheduler/VirtualTimeScheduler.ts:111:28)
      at VirtualAction.Object.<anonymous>.AsyncAction.execute (node_modules/rxjs/src/internal/scheduler/AsyncAction.ts:91:24)
      at TestScheduler.Object.<anonymous>.VirtualTimeScheduler.flush (node_modules/rxjs/src/internal/scheduler/VirtualTimeScheduler.ts:52:26)
      at TestScheduler.Object.<anonymous>.TestScheduler.flush (node_modules/rxjs/src/internal/testing/TestScheduler.ts:212:16)
      at TestScheduler.Object.<anonymous>.TestScheduler.run (node_modules/rxjs/src/internal/testing/TestScheduler.ts:678:12)
      at Object.wrapper (node_modules/rxjs-marbles/marbles.js:49:36)

In a playground:

Uncaught TypeError: Cannot read property 'next' of null
    at Object.next (share.js:31)
    at Object.eval [as next] (Subscriber.js:133)
    at SafeSubscriber.Subscriber._next (Subscriber.js:77)
    at SafeSubscriber.Subscriber.next (Subscriber.js:50)
    at eval (map.js:13)
    at OperatorSubscriber._this._next (OperatorSubscriber.js:22)
    at OperatorSubscriber.Subscriber.next (Subscriber.js:50)
    at eval (withLatestFrom.js:42)
    at OperatorSubscriber._this._next (OperatorSubscriber.js:22)
    at OperatorSubscriber.Subscriber.next (Subscriber.js:50)
    at Observable.eval [as _subscribe] (from.js:59)
    at Observable._trySubscribe (Observable.js:52)
    at Observable.subscribe (Observable.js:38)
    at eval (withLatestFrom.js:39)
    at OperatorSubscriber.eval (lift.js:17)
    at Observable.subscribe (Observable.js:38)
    at eval (map.js:12)
    at SafeSubscriber.eval (lift.js:17)
    at Observable.subscribe (Observable.js:38)
    at eval (share.js:29)
    at OperatorSubscriber.eval (lift.js:17)
    at Observable.subscribe (Observable.js:38)
    at Observable.eval [as _subscribe] (defer.js:11)
    at Observable._trySubscribe (Observable.js:52)
    at Observable.subscribe (Observable.js:38)
    at mergeInternals (mergeInternals.js:55)
    at eval (mergeMap.js:26)
    at OperatorSubscriber.eval (lift.js:17)
    at Observable.subscribe (Observable.js:38)
    at doInnerSub (mergeInternals.js:26)
    at outerNext (mergeInternals.js:20)
    at OperatorSubscriber._this._next (OperatorSubscriber.js:22)
    at OperatorSubscriber.Subscriber.next (Subscriber.js:50)
    at Observable.eval [as _subscribe] (from.js:59)
    at Observable._trySubscribe (Observable.js:52)
    at Observable.subscribe (Observable.js:38)
    at mergeInternals (mergeInternals.js:55)
    at eval (mergeMap.js:26)
    at OperatorSubscriber.eval (lift.js:17)
    at Observable.subscribe (Observable.js:38)
    at eval (startWith.js:17)
    at OperatorSubscriber.eval (lift.js:17)
    at Observable.subscribe (Observable.js:38)
    at _loop_1 (withLatestFrom.js:28)
    at eval (withLatestFrom.js:37)
    at OperatorSubscriber.eval (lift.js:17)
    at Observable.subscribe (Observable.js:38)
    at eval (map.js:12)
    at SafeSubscriber.eval (lift.js:17)
    at Observable.subscribe (Observable.js:38)

Expected behavior

In v6.x it worked as expected, without throwing any error.

Reproduction

https://codesandbox.io/s/stupefied-cdn-x92sk?file=/src/index.ts

Environment

• Runtime: any
• RxJS version: 7.0.0-beta.13

Additional context/Screenshots
It looks like subject at this point is null, so it breaks when calling .next, .error or .complete... but by reading the code I can't seem to understand how it's possible that it has a null value, as it seems like it's only set in here?

I tried adding a delay(x) on the deferredCount to check if it's because it's something synchronous, but that doesn't seem to help either.

Also note these kind of circular references in v6 always keep a subscriber open, which I guess it's discouraged. If I break this circular reference with a subject:

const countSubject = new Subject<string>();

const source: Observable<string> = from([1, 2, 3]).pipe(
  withLatestFrom(countSubject.pipe(startWith(0))),
  map(([a, b]) => String(Number(a) + Number(b))),
  tap(countSubject),
  share()
);

It doesn't throw any exception.

[cartant]

This is a bug, but IDK what behaviour you would expect in this contrived repro. AFAICT, no notifications would be received by any subscribers to source. That's because the subscription to the same, shared observable that occurs internally - within the withLatestFrom - completes and resetOnComplete defaults to true.

The fix would be for the subject and dest calls made with the share implementation's observer's handlers - e.g. here - to take into account that the subject could be null - e.g. next: (value) => subject?.next(value)

[cartant]

Actually, the problem is here:

rxjs/src/internal/operators/share.ts

Line 121 in 105bf99

connection = from(source).subscribe({

connection isn't assigned until the subscribe call returns, but that subscribe effects another - reentrant - call to the shared observable's subscribe. The implementation needs to use a Subscriber instance instead of an anonymous observer.