feat(rx): A new method of piping has been added #7229

benlesh · 2023-03-23T05:12:05Z

Adds rx a new method of piping observables. This allows any valid observable input to be passed as the first argument, and it will implicitly convert the argument to an Observable and pass it to the function that may be present in the second argument, the return value of THAT function is then passed an argument to the next function, and so on
Corrects the types of Observable#pipe such that anything can be returned at any step of the functional chain, however the first pipeable function must accept an Observable<T>.
Adds dtslint and runtime tests for rx and pipe.

NOTE: This does NOT deprecateObservable#pipe. That will be done in a follow up, and we need to define what timeline we'll take to remove that, as it's an API that is broadly used - could be v9, could be v10, could be never. At this point, this is alpha/beta functionality.

BREAKING CHANGE: Observable#pipe now allows any pipeable unary function as an argument, just so long as the types properly compose, this means in some cases it will now return unknown instead of Observable<unknown> the workaround is just to cast the result for those cases. (or you can break your operator piping into smaller pipe sets)

Addresses #7203

Note for later: Make sure to update squashed commit message to reflect `rx` and not `r`.

+ Adds `r` a new method of piping observables. This allows any valid observable input to be passed as the first argument, and it will implicitly convert the argument to an Observable and pass it to the function that may be present in the second argument, the return value of THAT function is then passed an argument to the next function, and so on + Corrects the types of `Observable#pipe` such that anything can be returned at any step of the functional chain, however the first pipeable function must accept an `Observable<T>`. + Adds dtslint and runtime tests for `r` and `pipe`. NOTE: This does NOT deprecate`Observable#pipe`. That will be done in a follow up, and we need to define what timeline we'll take to remove that, as it's an API that is broadly used - could be v9, could be v10, could be never. At this point, this is alpha/beta functionality. BREAKING CHANGE: `Observable#pipe` now allows any pipeable unary function as an argument, just so long as the types properly compose, this means in some cases it will now return `unknown` instead of `Observable<unknown>` the workaround is just to cast the result for those cases. (or you can break your operator piping into smaller pipe sets) Related ReactiveX#7203

src/internal/util/r.ts

benlesh · 2023-03-23T20:11:28Z

After some Core Team discussion, and some testing of the syntax, we've decided to move the implementation to where it takes an array of operators as the second argument. One of the example cases that compelled this change was this:

r(
  combineLatest([a, b]),
  map(([a, b]) => a + b),
  filter(n => n > 10),
)
.subscribe(console.log)

In the above scenario, muscle-memory for RxJS users leads them to believe that combineLatest is an operator, where it is in fact creating an observable.

The following was deemed to be more readable:

r(
  combineLatest([a, b]),
  [
    map(([a, b]) => a + b),
    filter(n => n > 10),
  ],
)
.subscribe(console.log)

This PR is currently broken

I knowingly pushed it up this way so it can be addressed later. There's a problem with the TypeScript types when we do this. TypeScript can't seem to figure out how to type the operators as they go through the array.

Actions

Explore alternative APIs? like: r({ source, operators }), this is too verbose and ends up with the same issues the array does.
Engage the TypeScript team to see if there's a better way to do this. Possibly some recursive type shenanigans?
Roll back to the rest arguments implementation as "good enough".

benlesh · 2023-03-23T20:18:39Z

cc @DanielRosenwasser 🙏❤️

jannispl · 2023-04-02T08:47:17Z

Personally I am not a huge fan of the new array notation. One of my main issues when dealing with RxJS operators is getting the damn brackets right when refactoring code. Introducing new ones won't help.

benlesh · 2023-04-10T16:10:29Z

After speaking offline with the TypeScript folks, there's just no way to properly type the array notation. So this is going back to the more "flat" approach.

benlesh · 2023-04-10T16:30:11Z

I know we talked about this @kwonoj ... but I'm really not sure I love r. lol... 😬

cartant

LGTM, just a few questions about any usage and a nitpick.

spec/util/pipe-spec.ts

cartant · 2023-04-25T04:56:58Z

src/internal/Observable.ts

+    op6: UnaryFunction<E, F>,
+    op7: UnaryFunction<F, G>,
+    op8: UnaryFunction<G, H>,
+    op9: UnaryFunction<H, Observable<I>>,
    ...operations: OperatorFunction<any, any>[]


Can we replace the any usage here with unknown?

cartant · 2023-04-25T04:57:06Z

src/internal/Observable.ts

+    op7: UnaryFunction<F, G>,
+    op8: UnaryFunction<G, H>,
+    op9: UnaryFunction<H, I>,
+    ...operations: UnaryFunction<any, any>[]


Can we replace the any usage here with unknown?

cartant · 2023-04-25T04:57:14Z

src/internal/Observable.ts

@@ -344,7 +350,7 @@ export class Observable<T> implements Subscribable<T> {
   *   .subscribe(x => console.log(x));
   * ```
   */
-  pipe(...operations: OperatorFunction<any, any>[]): Observable<any> {
+  pipe(...operations: UnaryFunction<any, any>[]): unknown {


Can we replace the any usage here with unknown?

cartant · 2023-04-25T04:58:59Z

src/internal/util/rx.ts

+  fn7: UnaryFunction<F, G>,
+  fn8: UnaryFunction<G, H>,
+  fn9: UnaryFunction<H, I>,
+  ...fns: UnaryFunction<any, any>[]


Same question as with pipe: Can we replace the any usage here with unknown?

kolodny

🚀🚀🚀

kolodny · 2023-04-25T15:10:50Z

src/internal/Observable.ts

    ...operations: OperatorFunction<any, any>[]
  ): Observable<unknown>;
+  pipe<A, B, C, D, E, F, G, H, I>(


I really wish we either had a dedicated pipe operator in js or at least had some way to use recursive/conditional types powerful enough to support not needing to do this.

src/internal/util/rx.ts

kolodny · 2023-04-25T15:18:15Z

src/internal/util/rx.ts

+ * ```ts
+ * r(source, map(x => x + 1), filter(x => x % 2 === 0));
+ * pipe(from(source), map(x => x + 1), filter(x => x % 2 === 0));
+ * pipe(source, from, map(x => x + 1), filter(x => x % 2 === 0));


Probably also worth calling out that rx([1, 2, 3]); is an Observable<number> and not Observable<number[]> in the documentation (it's already in the type tests)

jakovljevic-mladen

A lot of minor issues found: r to rx rename should be updated in couple of places, some generics issues need to be addressed and docs needs more info and some updates.

src/internal/Observable.ts

jakovljevic-mladen · 2023-04-26T07:14:00Z

src/internal/util/rx.ts

+import { ObservableInput, UnaryFunction } from '../types';
+import { pipeFromArray } from './pipe';
+
+export function rx<T, A>(source: ObservableInput<A>): Observable<A>;


The T generics are unused in the entire file. Please remove them.

src/internal/util/rx.ts

jakovljevic-mladen · 2023-04-26T07:17:51Z

src/internal/util/rx.ts

+ * pipe(source, from, map(x => x + 1), filter(x => x % 2 === 0));
+ * ```
+ *
+ * Furthermore, `r` can be used to create an observable and pipe it in any number of ways. For example:


Suggested change

* Furthermore, `r` can be used to create an observable and pipe it in any number of ways. For example:

* Furthermore, `rx` can be used to create an observable and pipe it in any number of ways. For example:

jakovljevic-mladen · 2023-04-26T07:24:35Z

src/internal/util/rx.ts

+export function rx<T, A>(source: ObservableInput<A>): Observable<A>;
+export function rx<T, A, B>(source: ObservableInput<A>, fn2: UnaryFunction<Observable<A>, B>): B;
+export function rx<T, A, B, C>(source: ObservableInput<A>, fn2: UnaryFunction<Observable<A>, B>, fn3: UnaryFunction<B, C>): C;


The T generics are unused in the entire file. Please remove them.

Or use them ☝️ like this:

Suggested change

export function rx<T, A>(source: ObservableInput<A>): Observable<A>;

export function rx<T, A, B>(source: ObservableInput<A>, fn2: UnaryFunction<Observable<A>, B>): B;

export function rx<T, A, B, C>(source: ObservableInput<A>, fn2: UnaryFunction<Observable<A>, B>, fn3: UnaryFunction<B, C>): C;

export function rx<T>(source: ObservableInput<T>): Observable<T>;

export function rx<T, A>(source: ObservableInput<T>, fn1: UnaryFunction<Observable<T>, A>): A;

export function rx<T, A, B>(source: ObservableInput<T>, fn1: UnaryFunction<Observable<T>, A>, fn2: UnaryFunction<A, B>): B;

...and so on.

FWIW: starting with T and fn1 (instead of fn2) is much better IMO as users might wonder where fn1 is when reviewing the docs.

This is the docs as I suggested here:

vs how it is now:

jakovljevic-mladen · 2023-04-26T07:47:51Z

src/internal/util/rx.ts

+ *
+ * Furthermore, `r` can be used to create an observable and pipe it in any number of ways. For example:
+ *
+ * ```ts


In order for these examples to work on StackBlitz, I'd like to introduce imports:

Suggested change

* ```ts

* ```ts

* import { rx, of } from 'rxjs';

jakovljevic-mladen · 2023-04-26T07:49:48Z

src/internal/util/rx.ts

+ *   },
+ * });
+ * ````
+ */


Please also add @param args ... and @return ... here. Docs seem empty for now:

I fixed the types up, which should resolve that a little.

I also added @param and @return docs.

jakovljevic-mladen · 2023-04-26T08:20:13Z

src/internal/util/rx.ts

+ *
+ * ```ts
+ * r(source, map(x => x + 1), filter(x => x % 2 === 0));
+ * pipe(from(source), map(x => x + 1), filter(x => x % 2 === 0));


This doesn't seem to be valid. This does:

Suggested change

* pipe(from(source), map(x => x + 1), filter(x => x % 2 === 0));

* pipe(() => from(source), map(x => x + 1), filter(x => x % 2 === 0));

Fixed. It was a little different than what that said.

jakovljevic-mladen · 2023-04-26T08:25:27Z

src/internal/util/rx.ts

+ *
+ * The following are equivalent:
+ *
+ * ```ts


Adding the source would help a lot. E.g.

const source = Promise.resolve(1);

or

const source = [1, 2, 3];

I added more context.

Co-authored-by: Mladen Jakovljević <jakovljevic.mladen@gmail.com>

Co-authored-by: Nicholas Jamieson <nicholas@cartant.com>

Co-authored-by: Mladen Jakovljević <jakovljevic.mladen@gmail.com>

+ Adds some more tests around `Observable#pipe`, ensuring arbitrary unary functions work + Adds more dtslint tests for `rx`. + Fixes some comments + Improves types and runtime for args in `rx` function

jakovljevic-mladen · 2023-05-16T12:29:19Z

src/internal/util/rx.ts

+import { pipeFromArray } from './pipe';
+
+export function rx<A>(source: ObservableInput<A>): Observable<A>;
+export function rx<A, B>(source: ObservableInput<A>, fn2: UnaryFunction<Observable<A>, B>): B;


I'm still in favor of using fn1 instead of fn2 here.

jakovljevic-mladen

LGTM overall, there are just some nitpicks that I'd like to be considered fixing.

benlesh force-pushed the r branch from 0f06cb7 to 70dc95a Compare March 23, 2023 05:35

benlesh added the 8.x Issues and PRs for version 8.x label Mar 23, 2023

benlesh force-pushed the r branch from 70dc95a to 4da2e83 Compare March 23, 2023 05:38

LcsGa reviewed Mar 23, 2023

View reviewed changes

src/internal/util/r.ts Outdated Show resolved Hide resolved

benlesh force-pushed the r branch from ca18f20 to 4da2e83 Compare April 10, 2023 16:12

morlay mentioned this pull request Apr 14, 2023

RFC: A new way of "piping" #7203

Open

refactor: rename to rx

5b196ef

cartant approved these changes Apr 25, 2023

View reviewed changes

kolodny approved these changes Apr 25, 2023

View reviewed changes

jakovljevic-mladen requested changes Apr 26, 2023

View reviewed changes

jakovljevic-mladen changed the title ~~feat(r): A new method of piping has been added~~ feat(rx): A new method of piping has been added Apr 26, 2023

benlesh and others added 5 commits May 15, 2023 16:57

docs: fix r to rx reference.

17c0d6c

Co-authored-by: Mladen Jakovljević <jakovljevic.mladen@gmail.com>

chore: fix description

de44750

Co-authored-by: Nicholas Jamieson <nicholas@cartant.com>

chore: fix return type of part of signature.

6d0519a

Co-authored-by: Mladen Jakovljević <jakovljevic.mladen@gmail.com>

chore: correct docs to say rx and not r.

8d92cad

Co-authored-by: Mladen Jakovljević <jakovljevic.mladen@gmail.com>

chore: address comments

72f9b47

+ Adds some more tests around `Observable#pipe`, ensuring arbitrary unary functions work + Adds more dtslint tests for `rx`. + Fixes some comments + Improves types and runtime for args in `rx` function

jakovljevic-mladen reviewed May 16, 2023

View reviewed changes

jakovljevic-mladen approved these changes May 16, 2023

View reviewed changes

benlesh merged commit c2cf0c4 into ReactiveX:master May 16, 2023
3 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat(rx): A new method of piping has been added #7229

feat(rx): A new method of piping has been added #7229

benlesh commented Mar 23, 2023 •

edited

benlesh commented Mar 23, 2023 •

edited

benlesh commented Mar 23, 2023

jannispl commented Apr 2, 2023

benlesh commented Apr 10, 2023

benlesh commented Apr 10, 2023

cartant left a comment

cartant Apr 25, 2023

cartant Apr 25, 2023

cartant Apr 25, 2023

cartant Apr 25, 2023

kolodny left a comment

kolodny Apr 25, 2023

kolodny Apr 25, 2023

jakovljevic-mladen left a comment

jakovljevic-mladen Apr 26, 2023

jakovljevic-mladen Apr 26, 2023

jakovljevic-mladen Apr 26, 2023

jakovljevic-mladen Apr 26, 2023 •

edited

jakovljevic-mladen Apr 26, 2023

benlesh May 15, 2023

benlesh May 15, 2023

jakovljevic-mladen Apr 26, 2023

benlesh May 15, 2023

jakovljevic-mladen Apr 26, 2023

benlesh May 15, 2023

jakovljevic-mladen May 16, 2023

jakovljevic-mladen left a comment

	* Furthermore, `r` can be used to create an observable and pipe it in any number of ways. For example:
	* Furthermore, `rx` can be used to create an observable and pipe it in any number of ways. For example:

	* pipe(from(source), map(x => x + 1), filter(x => x % 2 === 0));
	* pipe(() => from(source), map(x => x + 1), filter(x => x % 2 === 0));

feat(rx): A new method of piping has been added #7229

feat(rx): A new method of piping has been added #7229

Conversation

benlesh commented Mar 23, 2023 • edited

Note for later: Make sure to update squashed commit message to reflect rx and not r.

benlesh commented Mar 23, 2023 • edited

This PR is currently broken

Actions

benlesh commented Mar 23, 2023

jannispl commented Apr 2, 2023

benlesh commented Apr 10, 2023

benlesh commented Apr 10, 2023

cartant 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

Choose a reason for hiding this comment

kolodny left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jakovljevic-mladen 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

jakovljevic-mladen Apr 26, 2023 • edited

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

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

Choose a reason for hiding this comment

jakovljevic-mladen left a comment

Choose a reason for hiding this comment

benlesh commented Mar 23, 2023 •

edited

Note for later: Make sure to update squashed commit message to reflect `rx` and not `r`.

benlesh commented Mar 23, 2023 •

edited

jakovljevic-mladen Apr 26, 2023 •

edited