feat(publishReplay): add selector function to publishReplay #2885

martinsik · 2017-10-01T14:27:56Z

Description:

This PR adds selector function to the publishReplay operator similarly to publish.

I didn't modify any existing tests. However, this could be probably a breaking change because I added the parameter before scheduler (I'm not sure this is a problem or not).

Related issue (if exists):

#2844

2844

rxjs-bot · 2017-10-01T14:34:33Z

	Messages
📖

CJS: 1345.9KB, global: 745.4KB (gzipped: 140.0KB), min: 145.3KB (gzipped: 30.9KB)

Generated by 🚫 dangerJS

coveralls · 2017-10-01T14:35:54Z

Coverage increased (+0.0003%) to 97.477% when pulling 6d1742e on martinsik:2844-add-selector-to-publishreplay into e8d8c08 on ReactiveX:master.

benlesh

A change to the one test, and a request that more test coverage be added.

benlesh · 2017-10-02T19:58:18Z

spec/operators/publishReplay-spec.ts

@@ -409,4 +409,15 @@ describe('Observable.prototype.publishReplay', () => {

    published.connect();
  });
+
+  it('should mirror a simple source Observable with selector', () => {
+    const selector = observable => observable.map(v => String.fromCharCode(96 + parseInt(v)));


this one is a little too clever. :)

Can you use something more straightforward? A hash lookup or something?

benlesh · 2017-10-02T20:04:44Z

spec/operators/publishReplay-spec.ts

+
+    expectObservable(published).toBe(expected);
+    expectSubscriptions(source.subscriptions).toBe(sourceSubs);
+  });
 });


Need to add tests for:

errors thrown in selector directly.

when selector returns an errored Observable (--a--b--#)

when selector returns empty Observable

when selector returns never

when selector returns Observable.throw

benlesh · 2017-10-04T16:57:34Z

Also, @martinsik, if you can explain your cool pixel art profile pic, that would be good too. What is it?

benlesh

Last thing, we'll need to add function signature overrides for both operator/publishReplay and operators/publishReplay... they're slightly different. I've linked examples.

Thanks again!

benlesh · 2017-10-06T02:37:14Z

src/operator/publishReplay.ts

 * @param scheduler
- * @return {ConnectableObservable<T>}
+ * @return {Observable<T> | ConnectableObservable<T>}
 * @method publishReplay
 * @owner Observable
 */
 export function publishReplay<T>(this: Observable<T>, bufferSize: number = Number.POSITIVE_INFINITY,


Like below, we'll need to add some function overloads for TypeScript users to keep this change from being breaking. Here's what the publish counterpart to this file is doing...

rxjs/src/operator/publish.ts

Lines 7 to 9 in accbcd0

export function publish<T>(this: Observable<T>): ConnectableObservable<T>;

export function publish<T>(this: Observable<T>, selector: (source: Observable<T>) => Observable<T>): Observable<T>;

export function publish<T, R>(this: Observable<T>, selector: (source: Observable<T>) => Observable<R>): Observable<R>;

benlesh · 2017-10-06T02:37:58Z

src/operators/publishReplay.ts

@@ -7,6 +7,8 @@ import { UnaryFunction } from '../interfaces';

 export function publishReplay<T>(bufferSize: number = Number.POSITIVE_INFINITY,


I think we'll need some function signature overloads for this so we don't break compatability for people expecting ConnectableObservable<T> to be the only return type...

See what publish is doing here for a better idea:

rxjs/src/operators/publish.ts

Lines 6 to 8 in accbcd0

export function publish<T>(): MonoTypeOperatorFunction<T>;

export function publish<T>(selector: MonoTypeOperatorFunction<T>): MonoTypeOperatorFunction<T>;

export function publish<T, R>(selector: OperatorFunction<T, R>): OperatorFunction<T, R>;

coveralls · 2017-10-06T09:34:25Z

Coverage decreased (-0.03%) to 97.44% when pulling 2a157ef on martinsik:2844-add-selector-to-publishreplay into 0939f61 on ReactiveX:master.

benlesh · 2017-10-06T18:01:20Z

LGTM.... @kwonoj?

martinsik · 2017-10-06T18:06:01Z

@benlesh Sorry, it's work in progress I just pushed it because It was too late yesterday :). I'm not sure about a few things there.

btw, my avatar is a screenshot from this very old game https://en.wikipedia.org/wiki/Moonstone:_A_Hard_Days_Knight

martinsik · 2017-10-06T18:07:37Z

spec/operators/publishReplay-spec.ts

+    const published = source.publishReplay(1, Number.POSITIVE_INFINITY, selector);
+
+    // The exception is thrown outside Rx chain (not as an error notification).
+    expect(() => published.subscribe()).to.throw(error);


This exception is thrown outside the chain so I need to check it like this. This situations isn't tested in neither publish nor multicast (if I didn't miss anything) so I'm not sure I'm doing it right.

OH GEEZ! I missed this... So what should happen is you should catch the error and send it down the error path with observer.error(err). There are all sorts of examples of how to handle this around teh library.

Basically any user-supplied function needs to have it's errors caught and sent down the error channel of observation.

But I would expect multicast to catch this error for me (for example like the map() operator does) because that's where the selector is called: https://github.com/ReactiveX/rxjs/blob/master/src/operators/multicast.ts#L64

martinsik · 2017-10-06T18:09:04Z

src/operator/publishReplay.ts

+                                    selectorOrScheduler?: IScheduler | OperatorFunction<T, R>,
+                                    scheduler?: IScheduler): Observable<R> | ConnectableObservable<R> {
+
+  return higherOrder(bufferSize, windowTime, selectorOrScheduler as any, scheduler)(this);


I couldn't find any other sane way to call higherOrder. TypeScript didn't allow me just using:

higherOrder(bufferSize, windowTime, selectorOrScheduler, scheduler)

.. but I think this should work.

This will do, the other thing you can do is duplicate the logic that figures out what was passed from the other method.

I understand, I didn't want to write the same logic twice so I'll leave it as it is.

martinsik · 2017-10-06T18:12:27Z

src/operators/publishReplay.ts

-                                 scheduler?: IScheduler): UnaryFunction<Observable<T>, ConnectableObservable<T>> {
-  return (source: Observable<T>) => multicast(new ReplaySubject<T>(bufferSize, windowTime, scheduler))(source) as ConnectableObservable<T>;
+/* tslint:disable:max-line-length */
+export function publishReplay<T>(bufferSize?: number, windowTime?: number, scheduler?: IScheduler): UnaryFunction<Observable<T>, ConnectableObservable<T>>;


The publish operator defines MonoTypeOperatorFunction return type but I don't think that's correct because this results into Observable<T> while publish should return ConnectableObservable<T>.

https://github.com/ReactiveX/rxjs/blob/master/src/operators/publish.ts#L6-L8

This is fine for now... we might need to address this more generally across the library.

martinsik · 2017-10-06T18:13:47Z

src/operators/publishReplay.ts

+export function publishReplay<T, R>(bufferSize?: number, windowTime?: number, selector?: OperatorFunction<T, R>, scheduler?: IScheduler): OperatorFunction<T, R>;
+/* tslint:enable:max-line-length */
+
+export function publishReplay<T, R>(bufferSize?: number,


I removed all default values here and left it up to ReplaySubject to decide.

benlesh

Add appropriate error handling.

benlesh · 2017-10-06T18:23:48Z

spec/operators/publishReplay-spec.ts

+    const published = source.publishReplay(1, Number.POSITIVE_INFINITY, selector);
+
+    // The exception is thrown outside Rx chain (not as an error notification).
+    expect(() => published.subscribe()).to.throw(error);


OH GEEZ! I missed this... So what should happen is you should catch the error and send it down the error path with observer.error(err). There are all sorts of examples of how to handle this around teh library.

Basically any user-supplied function needs to have it's errors caught and sent down the error channel of observation.

benlesh · 2017-10-06T18:25:20Z

src/operator/publishReplay.ts

+                                    selectorOrScheduler?: IScheduler | OperatorFunction<T, R>,
+                                    scheduler?: IScheduler): Observable<R> | ConnectableObservable<R> {
+
+  return higherOrder(bufferSize, windowTime, selectorOrScheduler as any, scheduler)(this);


This will do, the other thing you can do is duplicate the logic that figures out what was passed from the other method.

benlesh · 2017-10-06T18:27:24Z

src/operators/publishReplay.ts

-                                 scheduler?: IScheduler): UnaryFunction<Observable<T>, ConnectableObservable<T>> {
-  return (source: Observable<T>) => multicast(new ReplaySubject<T>(bufferSize, windowTime, scheduler))(source) as ConnectableObservable<T>;
+/* tslint:disable:max-line-length */
+export function publishReplay<T>(bufferSize?: number, windowTime?: number, scheduler?: IScheduler): UnaryFunction<Observable<T>, ConnectableObservable<T>>;


This is fine for now... we might need to address this more generally across the library.

martinsik · 2017-10-16T08:23:00Z

@benlesh I added a comment above that I'd expect multicast to catch the error because that's where the selector is called https://github.com/ReactiveX/rxjs/blob/master/src/operators/multicast.ts#L64.

benlesh · 2017-10-18T15:04:01Z

Merged! Thanks @martinsik!

lock · 2018-06-06T06:12:51Z

This thread has been automatically locked since there has not been any recent activity after it was closed. Please open a new issue for related bugs.

feat(publishReplay): add selector function to publishReplay

6d1742e

2844

benlesh requested changes Oct 2, 2017

View reviewed changes

martinsik added 3 commits October 5, 2017 23:30

fix(publishReplay): make the selector optional

6daa0a6

test(publishReplay): add more test for the selector function

a0182ce

Merge branch 'master' into 2844-add-selector-to-publishreplay

684ec1e

benlesh requested changes Oct 6, 2017

View reviewed changes

fix(publishReplay): fix publishReplay function signatures

2a157ef

benlesh approved these changes Oct 6, 2017

View reviewed changes

martinsik commented Oct 6, 2017

View reviewed changes

benlesh requested changes Oct 6, 2017

View reviewed changes

benlesh merged commit e0efd13 into ReactiveX:master Oct 18, 2017

lock bot locked as resolved and limited conversation to collaborators Jun 6, 2018

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat(publishReplay): add selector function to publishReplay #2885

feat(publishReplay): add selector function to publishReplay #2885

martinsik commented Oct 1, 2017

rxjs-bot commented Oct 1, 2017 •

edited

Loading

coveralls commented Oct 1, 2017

benlesh left a comment

benlesh Oct 2, 2017

benlesh Oct 2, 2017

benlesh commented Oct 4, 2017

benlesh left a comment

benlesh Oct 6, 2017

benlesh Oct 6, 2017

coveralls commented Oct 6, 2017

benlesh commented Oct 6, 2017

martinsik commented Oct 6, 2017 •

edited

Loading

martinsik Oct 6, 2017

benlesh Oct 6, 2017

martinsik Oct 7, 2017 •

edited

Loading

martinsik Oct 6, 2017

benlesh Oct 6, 2017

martinsik Oct 7, 2017

martinsik Oct 6, 2017 •

edited

Loading

benlesh Oct 6, 2017

martinsik Oct 6, 2017

benlesh left a comment

benlesh Oct 6, 2017

benlesh Oct 6, 2017

benlesh Oct 6, 2017

martinsik commented Oct 16, 2017

benlesh commented Oct 18, 2017

lock bot commented Jun 6, 2018

	export function publish<T>(this: Observable<T>): ConnectableObservable<T>;
	export function publish<T>(this: Observable<T>, selector: (source: Observable<T>) => Observable<T>): Observable<T>;
	export function publish<T, R>(this: Observable<T>, selector: (source: Observable<T>) => Observable<R>): Observable<R>;

		@@ -7,6 +7,8 @@ import { UnaryFunction } from '../interfaces';

		export function publishReplay<T>(bufferSize: number = Number.POSITIVE_INFINITY,

	export function publish<T>(): MonoTypeOperatorFunction<T>;
	export function publish<T>(selector: MonoTypeOperatorFunction<T>): MonoTypeOperatorFunction<T>;
	export function publish<T, R>(selector: OperatorFunction<T, R>): OperatorFunction<T, R>;

feat(publishReplay): add selector function to publishReplay #2885

feat(publishReplay): add selector function to publishReplay #2885

Conversation

martinsik commented Oct 1, 2017

rxjs-bot commented Oct 1, 2017 • edited Loading

coveralls commented Oct 1, 2017

benlesh left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

benlesh commented Oct 4, 2017

benlesh left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

coveralls commented Oct 6, 2017

benlesh commented Oct 6, 2017

martinsik commented Oct 6, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

martinsik Oct 7, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

martinsik Oct 6, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

benlesh left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

martinsik commented Oct 16, 2017

benlesh commented Oct 18, 2017

lock bot commented Jun 6, 2018

rxjs-bot commented Oct 1, 2017 •

edited

Loading

martinsik commented Oct 6, 2017 •

edited

Loading

martinsik Oct 7, 2017 •

edited

Loading

martinsik Oct 6, 2017 •

edited

Loading