Объединение операторов RxJS в новый оператор с помощью TypeScript

Я часто обнаруживаю, что добавляю одну и ту же последовательность операторов к наблюдаемым, например 

observable$
  .do(x => console.log('some text', x))
  .publishReplay()
  .refCount();
Я ищу способ объединить эти 3 оператора в небольшой многоразовый оператор (например, .cache('some text')), который я могу связать с любым наблюдаемым. Как я могу определить это в Typescript, чтобы я мог импортировать rxjs/Observable и этот оператор, как я делаю с операторами rxjs?
2 9
typescript rxjs rxjs5

2 ответа:

Чтобы реализовать описанный оператор, создайте файл cache.ts следующего содержания: 

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;
  }
}

И потреблять его вот так: 

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));
21

Ответ Картанта, приведенный выше, хорошо работает и отвечает на заданный вопрос.( Как я могу определить это в Typescript, чтобы я мог импортировать rxjs/Observable и этот оператор, как я делаю с операторами rxjs?)

Недавно я обнаружил оператор let, который, если вам на самом деле не нужно, чтобы функция была реализована как оператор, все равно позволит вам высушить ваш код.

Я начал внедрять услугу angular 2, чтобы интерфейс с моим бэкендом rails и знал, что большинство моих вызовов api будут выглядеть очень похоже, поэтому я хотел попробовать и поместить как можно больше общих вещей в функцию.

Почти все вызовы будут делать следующее:

  1. повторите попытку ошибки (моя функция ниже нуждается в дополнительной работе на этом фронте)
  2. отображение http-ответа в локально определенный класс typescript (через json-typescript-mapper )
  3. обрабатывать ошибки

Вот пример моего использования let оператор для моих http-ответов через общую функцию (handleResponse)через оператор RxJS let.

  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}));
  }
2