Stream across node projects prevent process from exiting

[cdanielw]

Bug Report

Current Behavior
Creating a stream in one project, and using it in another can prevent the process from exiting.

Reproduction

• REPL or Repo link: https://github.com/cdanielw/rxjs-issue

Note that this use a forked rxjs version. I just packaged HEAD and included dist in the repo, to easily add the dependency in package.json.

1. npm install both a and b
2. node a/main.js

Project a:

const {from, of, interval} = require('rxjs')
const {switchMap, take} = require('rxjs/operators')
const {
    create$,
    // Re-exporting rxjs also allows it to exit properly
    // rxjs: {from, of, interval}, rxjsOperators: {switchMap, take}
} = require('b/main')

const b$ = create$() // Process doesn't exit
// const b$ = from(create$()) // Exits properly
// const b$ = of('b') // Exits properly

b$.pipe(
    switchMap(b => interval(100)),
    take(3)
).subscribe(
    next => console.log('NEXT', next),
    error => console.log('ERROR', error),
    () => console.log('COMPLETE'),
)

Project b:

const {of} = require('rxjs')

module.exports = {
    create$: () => of('b'),
     // Re-export rxjs as a workaround
    rxjs: require('rxjs'),
    rxjsOperators: require('rxjs/operators')
}

Expected behavior
Process exit after some console output.

Environment

• Runtime: Node v13.10.1
• RxJS version: HEAD, 7.0.0-beta.0, 6.5.5

Possible Solution
I’ve found two workarounds, but only one really practical for my purposes.

1. Wrap the stream from the other project with from()
2. Share the same rxjs library, by having one project re-export it.

The first option is very unpractical. This is a simple example, where it’s clear where to wrap it. In real life, things are very different. This would make it terrible difficult to refactor shared code into a library.

The second option is what I’m using now. It’s pretty ugly, but it can be implemented with a simple find/replace.

Additional context/Screenshots
I wonder if this is not related to #5237

[cartant]

I wonder if this is not related to #5237

Yes. The subscribeWith util that was added in #5333 needs to be used in all operators.

You should be able to confirm that the problem is what you think it is by using from(b$).pipe(... - as, in your snippet, from is imported from the same module as the operators.

[cartant]

It seems that there is more to this than just using subscribeWith in each operator.

There is also an interop subscriber problem that's due to lift. The interop source observable is lifted, so every subscribe call is an interop call and every operator subscriber receives an interop destination subscriber. And that hits the SafeSubscriber problem that this PR deals with: #5311

---

Actually, it's more subtle. The PR referenced above deals with effecting unsubscriptions in safe subscribers that are effected by notifications. In this issue, the unsubscription is effected by an explicit unsubscription - via the take operator - so, AFAICT, the chaining introduced in the referenced PR won't resolve the problem. The subscribers need to be chained in the opposite direction, too - see this line, where the non-interop source and destination subscribers are chained together.