Combine RxJS operators into new operator using TypeScript

Question

I frequently find my self adding the same sequence of operators to observables, e.g.

observable$
  .do(x => console.log('some text', x))
  .publishReplay()
  .refCount();

I'm looking for a way to combine these 3 operators in a small reusable operator (e.g. .cache('some text')) that I can chain to any observable. How can I define this in Typescript, so that I could import rxjs/Observable and this operator, like I do with rxjs operators?

cartant · Accepted Answer · 2017-02-08 07:05:02Z

25

To implement the operator you have described, create a cache.ts file with the following content:

import { Observable } from "rxjs/Observable";
import "rxjs/add/operator/do";
import "rxjs/add/operator/publishReplay";

// Compose the operator:

function cache<T>(this: Observable<T>, text: string): Observable<T> {
  return this
    .do(x => console.log(text, x))
    .publishReplay()
    .refCount();
}

// Add the operator to the Observable prototype:

Observable.prototype.cache = cache;

// Extend the TypeScript interface for Observable to include the operator:

declare module "rxjs/Observable" {
  interface Observable<T> {
    cache: typeof cache;
  }
}

And consume it like this:

import { Observable } from "rxjs/Observable";
import "rxjs/add/observable/of";
import "./cache";

let cached = Observable.of(1).cache("some text");
cached.subscribe(x => console.log(x));

edited Feb 8, 2017 at 7:05

answered Feb 4, 2017 at 0:43

cartant

58.6k17 gold badges169 silver badges201 bronze badges

Sign up to request clarification or add additional context in comments.

6 Comments

nPn Over a year ago

I had a similar question, this looks like a good solution. Since this is just a chain of operators, does it handle errors and complete by default? Seems like it should, but I am new to rxjs.

cartant Over a year ago

@nPn Yep, errors and completions will be received by the composed observable's subscriber(s). There's nothing that needs to be done here.

Dustin Cleveland Over a year ago

When I try this approach, I get this error:

EXCEPTION: Uncaught (in promise): Error: Error in ./DashboardComponent class DashboardComponent - inline template:2:4 caused by: __WEBPACK_IMPORTED_MODULE_2_rxjs_Observable__.Observable.of(...).restrictToCommand is not a function TypeError: __WEBPACK_IMPORTED_MODULE_2_rxjs_Observable__.Observable.of(...).restrictToCommand is not a functio

. Any ideas?

cartant Over a year ago

@Dustin There really isn't enough space for context in comments. If you have a problem, ask a question. You can always reference this answer in the question.

Dustin Cleveland Over a year ago

@cartant: If you're willing to take a look, here's the question.

|

nPn · Accepted Answer · 2017-02-22 03:41:06Z

cartant's answer above works well, and answers the question that was asked (How can I define this in Typescript, so that I could import rxjs/Observable and this operator, like I do with rxjs operators?)

I recently discovered the let operator which if you don't actually need to have the function implemented as an operator, will still let you DRY up your code.

I was starting on implementing an angular 2 service to interface with my rails backend and knew that most of my api calls would look very similar so I wanted to try and put as much of the common stuff in a function.

Almost all the calls will do the following:

retry on an error (my function below needs more work on that front)
map the http response into a typescript locally defined class (via json-typescript-mapper)
handle errors

Here is an example of my use the let operator to my http responses through a common function (handleResponse) via the rxjs let operator.

  handleResponse<T>({klass, retries=0} :{klass:any,retries?:number }) : (source: Observable<Response>) => Observable<T> {
    return (source: Observable<Response>)  : Observable<T> => {
      return source.retry(retries)
            .map( (res) => this.processResponse(klass,res))
            .catch( (res) => this.handleError(res));
    } 
  } 

  processResponse(klass, response: Response) {
    return deserialize(klass, response.json());
  }

  handleError(res: Response) {
    const error = new RailsBackendError(res.status, res.statusText);
    return  Observable.throw(error);
  }

  getUserList({page=1,perPage=30,retry=0}: { page?:number, perPage?:number, retry?:number }={}) : Observable<UserList> {
    const requestURL = `/api/v1/users/?${this.apiTokenQueryString}&page=${page}&per_page=${perPage}`;
    return this.http.get(requestURL).let(this.handleResponse<UserList>({klass: UserList}));
  }

Collectives™ on Stack Overflow

Combine RxJS operators into new operator using TypeScript

2 Answers 2

6 Comments

Comments

Your Answer

Linked

Hot Network Questions

Collectives™ on Stack Overflow

2 Answers 2

6 Comments

Comments

Your Answer

Sign up or log in

Post as a guest

Linked

Related