Reworking Multicasting: share, connect, and makeConnectable by benlesh · Pull Request #5634 · ReactiveX/rxjs

benlesh · 2020-08-05T20:45:20Z

I'd like to do this instead of what I was doing over at #5432 ..

Overview

Make share configurable, such that it can be used to create things like shareReplay or people can configure the behavior however they like. If they want a shareBehavior or shareLast, they can easily make it work however they like.
Add a newconnect operator that does the "selector" version of multicast, publish, et al. (this also fixes a broken error behavior in publishReplay).
Add a new makeConnectable function that creates a ConnectableObservableLike (instead of ConnectableObservable).

What this sets us up for:

We'd have

share (NOTE: I'll leave shareReplay as just a wrapper around this, as seen in this PR)
connect
connectable

Instead of

share
multicast
publish
publishLast
publishReplay
publishBehavior
refCount
shareReplay
ConnectableObservable

All "operators" at that point would actually only return Observable and never ConnectableObservable, nor would any require ConnectableObservable as a source (such as with refCount).

TODO:

Additional documentation
Additional tests
Add deprecation details to deprecated operators and paths

Other thoughts

It's possible that a better API for makeConnectable might be to return a tuple with a connect function. Like so:

const [ result$, connect ] = makeConnectable(source);
result$.subscribe(console.log);
connect();

// instead of
const result$ = makeConnectable(source);
result$.subscribe(console.log);
result$.connect();

How this "replaces" certain things

Well, the I find the prevalence of publishReplay(1), refCount() very disturbing. For one thing, I doubt many folks using this are aware of what this does completely. It's not retryable, it's not replayable, etc.

a$.pipe(publishReplay(1), refCount())

// is

a$.pipe(
  share({
    connector: () => new ReplaySubject(1),
    resetOnError: false,
    resetOnComplete: false,
    resetOnUnsubscribe: false
  });
)

"But Ben, that's more code!"... yes, and it's explicit code that tells you exactly what you're doing there. That's not retryable, it's not repeatable... it will never reset internally. It also allows the author more control without having to know implementation details of publishReplay. Needing to know implementation details of our multicasting operators has been a sore point of this library, IMO.

NO WORRIES: a$.pipe(share()) will work exactly as it does now. (Notice those tests didn't change)

benlesh · 2020-08-05T21:14:51Z

Related: #3833

benlesh · 2020-08-06T14:34:54Z

Rebased

benlesh · 2020-08-12T20:20:48Z

Core team to review... revisit next meeting.

benlesh · 2020-09-19T23:47:05Z

This is blocked on #5729

cartant

LGTM. Just some questions, a bunch of nitpicks and some names of which I am not a fan. 😅

Also, after this is rebased, the skipped firehose tests should be unskipped and made to pass. A contributor opened a PR to address the skipped tests, but I blocked the PR, as the changes were based on the old multicast infrastructure - see #5834

cartant · 2020-10-20T01:26:00Z

src/internal/operators/connect.ts

+    return lift(source, function (this: Subscriber<R>, source: Observable<T>) {
+      const subscriber = this;
+      let subject: Subject<T>;
+      try {
+        subject = connector();
+      } catch (err) {
+        subscriber.error(err);
+        return;
+      }
+
+      let result: Observable<R>;
+      try {
+        result = from(setup(subject.asObservable()));
+      } catch (err) {
+        subscriber.error(err);
+        return;
+      }
+
+      const subscription = result.subscribe(subscriber);
+      subscription.add(source.subscribe(subject));
+      return subscription;
+    });


nitpick: This PR predates the smallification, so it should be updated to use operate. And the source.subscribe(subject) subscription should be added to the subscriber, etc.

cartant · 2020-10-20T01:29:21Z

src/internal/operators/connect.ts

+  setup,
+}: {
+  connector?: () => Subject<T>;
+  setup: (shared: Observable<T>) => ObservableInput<R>;


nitpick: I'm not a fan of the setup property's name. TBH, I preferred selector. My main beef with setup is that, IMO, it doesn't really explain what the parameter does. Elsewhere in the API functions that are used in this manner are called selectors and I am a fan of consistency and precedent.

how about project?

selector, IMO, also doesn't explain what the parameter does, either. I mean, this sets up the multicast as much as it selects the output. Happy to bikeshed the name, but overall, I guess I'm just not sure what to name this. project is used elsewhere to mean a synchronous map-type function, and I'm not thrilled with its use in things like mergeMap.

Maybe cast? or multicast? or??

(I'm open to moving to back to the selector, though). I just don't like that it doesn't completely imply what's going on here. catchError is a little different, because it's only called when an error happens, and you have to "select" what to observe after that error shuts things down.

This function is required to setup the multicast. It also selects what sort of values are emitted from the result. I guess I'm now unhappy with either setup or select haha. But I do know I want it to be terse.

This function is required to setup the multicast. It also selects what sort of values are emitted from the result.

Well, the function is "required to setup" the multicast, but it doesn't actually set it up. It receives the shared/multicast observable and then "selects what sort of values are emitted from the result".

That said, cast or multicast are okay, I suppose, but my preference is still for selector.

cartant · 2020-10-20T01:37:29Z

src/internal/observable/makeConnectable.ts

+
+  result.connect = function () {
+    if (!connection) {
+      connection = defer(() => source).subscribe(connector);


question: Is there a reason for using defer here instead of from? source is a parameter and is never reassigned, so I don't see why this needs to be lazy.

cartant · 2020-10-20T01:41:40Z

src/internal/observable/makeConnectable.ts

+ * @returns A "connectable" observable, that has a `connect()` method, that you must call to
+ * connect the source to all consumers through the subject provided as the connector.
+ */
+export function makeConnectable<T>(source: ObservableInput<T>, connector: Subject<T> = new Subject<T>()): ConnectableObservableLike<T> {


nitpick: Not a fan of the makeConnectable name. IMO, this seems odd, when looking at the other parts of the API. I mean, it's timer, not makeTimer or createTimer. My preference would be for it to be named connectable.

cartant · 2020-10-20T01:46:07Z

src/internal/operators/share.ts


-function shareSubjectFactory() {
-  return new Subject<any>();
+interface ShareOptions<T, R> {


(very minor) nitpick: In other parts of the API, these types/parameters are referred to as 'config'. I think it should be one or the other and should be consistent throughout the API, but - again - this is very minor.

cartant · 2020-10-20T01:47:25Z

src/internal/operators/share.ts

+      hasCompleted = hasErrored = false;
+    };
+
+    return lift(source, function (this: Subscriber<T | R>, source: Observable<T>) {


nitpick: With the changes made in the smallification, this should now used operate.

cartant · 2020-10-20T01:49:45Z

src/internal/operators/share.ts

+        subject = connector!();
+      }
+
+      const castSubscription = subject.subscribe(subscriber);


nitpick: This seems like a weird variable name, IMO. Where is the cast?

backbone87 · 2020-11-18T09:02:10Z

src/internal/operators/share.ts

+    let { connector, resetOnComplete = true, resetOnError = true, resetOnUnsubscribe = true } = options;
+    if (!connector) {
+      connector = () => new Subject<T>();
+    }


isnt this more consistent?

const { connector = () => new Subject<T>(), resetOnComplete = true, resetOnError = true, resetOnUnsubscribe = true, } = options || {};

also shouldnt this be moved into the share functions block insteadof the operator function block? maybe even:

export function share<T, R>({ connector = () => new Subject<T>(), resetOnComplete = true, resetOnError = true, resetOnUnsubscribe = true, }: ShareOptions<T, R> = {}): OperatorFunction<T, T | R> { // ... }

also the type param R doesnt seem to be used really?

backbone87 · 2020-11-18T09:09:13Z

src/internal/operators/share.ts

+   * If false, when the number of subscribers to the resulting observable reaches zero due to unsubscription, the subject
+   * will remain connected to the source, and new subscriptions to the result will be connected through that same subject.
+   */
+  resetOnUnsubscribe?: boolean;


i was under the assumption that a complete/error will always result in an implicit unsubscribe afterwards. its not reflected in this description that an unsubscribe that follows a complete/error, will not result in a reset.

This means that the next time the resulting observable is subscribed to, a new subject will be created and the source will be subscribed to again.

This should also be added to the resetOnError/resetOnComplete descriptions. Maybe its worth to describe a reset independantly of these 3 configs in the share operator itself, so that the description of the reset params can be reduced to actual differences between them and therefore causing less mental friction.

backbone87 · 2020-11-18T09:26:37Z

src/internal/operators/share.ts


-function shareSubjectFactory() {
-  return new Subject<any>();
+interface ShareOptions<T, R> {


is it worth to add a resetNotifier$ in order to have more fine grained control over when resets happen?

You could technically do that by composing takeUntil

source$.pipe( takeUntil(resetNotifier$), share(), repeat(), )

backbone87 · 2020-11-18T09:35:16Z

It's possible that a better API for makeConnectable might be to return a tuple with a connect function. Like so:

just throwing in other alternatives:

const notifier = new Subject();

makeConnectable(source, notifier);
notifier.subscribe(connects(source));

Adds a feature that allows two observables to be tested for equality of output.

…Like`. BREAKING CHANGE: The TypeScript type `Subscribable` now only supports what is a valid return for `[Symbol.observable]()`. BREAKING CHANGE: The TypeScript type `Observer` no longer incorrectly has an optional `closed` property.

benlesh · 2020-11-28T02:13:50Z

Okay, I've made a few changes to this.

Per @cartant's request, makeConnectable is now connectable.

OTHER VERY IMPORTANT THINGS:

I've added a new TestScheduler helper: expectObservable(a$).toEqual(b$). I needed this so I could test that the deprecation advice I was giving people in the deprecation messages was proven, not just theoretical. It's publicly exposed, and I think we should ship it. It's useful for this exact scenario and it follows the jest and jasmine pattern of toBe and toEqual, IMO.
I've fixed the Observer interface to remove the optional closed property, which shouldn't have been there. (That may be breaking for TypeScript users).
I've fixed the Subscribable interface to just be what is valid as a return value of [Symbol.observable](). That's because I was using with SubjectLike.
I'm now exporting a new type SubjectLike, it's exactly what it sounds like.

TODO

We need to discuss SubjectLike. We might want it to be SubjectLike<In, Out> and not just SubjectLike<T>. This relates to things like WebSocketSubject, etc.
We need to bikeshed the setup function as discussed above.

cc @cartant @kwonoj

cartant

Some things that might or might not be issues and some nitpicks and comments/thoughts, etc. No major issues, AFAICT.

src/internal/types.ts

cartant · 2020-12-01T05:08:09Z

src/internal/operators/connect.ts

+  setup,
+}: {
+  connector?: () => Subject<T>;
+  setup: (shared: Observable<T>) => ObservableInput<R>;


This function is required to setup the multicast. It also selects what sort of values are emitted from the result.

Well, the function is "required to setup" the multicast, but it doesn't actually set it up. It receives the shared/multicast observable and then "selects what sort of values are emitted from the result".

That said, cast or multicast are okay, I suppose, but my preference is still for selector.

src/internal/operators/share.ts

spec/schedulers/TestScheduler-spec.ts

src/internal/observable/connectable.ts

src/internal/operators/multicast.ts

spec/operators/share-spec.ts

benlesh · 2020-12-06T01:24:13Z

Requesting rereview.

Note that the change to make Subscribable accept Partial<Observer<T>> also meant that the type for Observable subscribe needed to take Partial<Observer<T>> instead of PartialObserver<T>, the latter being a type that forced at least one callback (for better or worse). I'm totally fine with this, and I'm not sure that bit was helpful anyhow. If anything, I'd bet it was confusing for users at times.

cartant

LGTM ❤

src/internal/operators/publishLast.ts

cartant · 2020-12-14T03:35:17Z

🎉

benlesh requested a review from cartant August 5, 2020 20:48

benlesh added the AGENDA ITEM Flagged for discussion at core team meetings label Aug 5, 2020

benlesh mentioned this pull request Aug 5, 2020

WIP: Add static multicasters #5432

Closed

7 tasks

benlesh requested review from kolodny and kwonoj August 5, 2020 21:19

benlesh force-pushed the spike/multicast-do-over branch from 90a9b23 to a4c11a3 Compare August 6, 2020 14:34

benlesh added the blocked label Sep 19, 2020

niklas-wortmann mentioned this pull request Oct 5, 2020

[WIP] Roadmap V7 #5795

Closed

8 tasks

cartant mentioned this pull request Oct 16, 2020

fix(multicast,share,refCount,shareReplay): enable synchronous firehose #5834

Closed

cartant approved these changes Oct 20, 2020

View reviewed changes

cartant mentioned this pull request Nov 16, 2020

Change behavior of shareReplay #5894

Closed

backbone87 reviewed Nov 18, 2020

View reviewed changes

benlesh force-pushed the spike/multicast-do-over branch from a4c11a3 to b43d28f Compare November 27, 2020 23:35

benlesh added 6 commits November 27, 2020 18:22

feat(TestScheduler): add expectObservable(a$).toEqual(b$).

e7ce95a

Adds a feature that allows two observables to be tested for equality of output.

feat(connect): Adds new connect operator.

553f7a7

feat(connectable): Adds connectable creation method

b86f574

chore(deprecations): Adds documentation and tests for deprecations

ac1e39f

chore: update golden files

d760a6c

benlesh force-pushed the spike/multicast-do-over branch from b43d28f to d760a6c Compare November 28, 2020 02:08

benlesh requested a review from cartant November 28, 2020 02:13

chore: Fix types to pass dtslint

f55fd35

cartant reviewed Dec 1, 2020

View reviewed changes

chore: Address comments

affeb50

benlesh requested a review from cartant December 6, 2020 01:22

cartant approved these changes Dec 9, 2020

View reviewed changes

backbone87 reviewed Dec 9, 2020

View reviewed changes

src/internal/operators/publishLast.ts Show resolved Hide resolved

benlesh merged commit 8c13e0a into ReactiveX:master Dec 14, 2020

benlesh mentioned this pull request Jan 12, 2021

Memory leaks caused by shareReplay when shared observable is synchronous or continuous #5931

Closed

cartant mentioned this pull request Mar 13, 2021

Split multicast, publish, et al into static functions and operators? #3833

Closed

Conversation

benlesh commented Aug 5, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Overview

What this sets us up for:

We'd have

Instead of

TODO:

Other thoughts

How this "replaces" certain things

Uh oh!

benlesh commented Aug 5, 2020

Uh oh!

benlesh commented Aug 6, 2020

Uh oh!

benlesh commented Aug 12, 2020

Uh oh!

benlesh commented Sep 19, 2020

Uh oh!

cartant left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

benlesh Nov 27, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

backbone87 Nov 18, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

backbone87 commented Nov 18, 2020

Uh oh!

benlesh commented Nov 28, 2020

OTHER VERY IMPORTANT THINGS:

TODO

Uh oh!

cartant left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

benlesh commented Dec 6, 2020

benlesh commented Aug 5, 2020 •

edited

Loading

cartant left a comment •

edited

Loading

benlesh Nov 27, 2020 •

edited

Loading

backbone87 Nov 18, 2020 •

edited

Loading