Qual é a maneira correta de compartilhar o resultado de uma chamada de rede Angular Http nos RxJs 5?

Usando o HTTP, chamamos um método que faz uma chamada de rede e retorna um http observável:

getCustomer() {
    return this.http.get('/someUrl').map(res => res.json());
}

Se considerarmos isso observável e adicionar vários assinantes:

let network$ = getCustomer();

let subscriber1 = network$.subscribe(...);
let subscriber2 = network$.subscribe(...);

O que queremos fazer é garantir que isso não cause várias solicitações de rede.

Pode parecer um cenário incomum, mas na verdade é bastante comum: por exemplo, se o chamador assina o observável para exibir uma mensagem de erro e a transmite ao modelo usando o canal assíncrono, já temos dois assinantes.

Qual é a maneira correta de fazer isso nos RxJs 5?

Ou seja, isso parece funcionar bem:

getCustomer() {
    return this.http.get('/someUrl').map(res => res.json()).share();
}

Mas essa é a maneira idiomática de fazer isso nos RxJs 5, ou deveríamos fazer outra coisa?

Nota: Conforme o novo Angular 5 HttpClient, a .map(res => res.json())parte em todos os exemplos agora é inútil, pois o resultado JSON agora é assumido por padrão.

Coloque em cache os dados e, se disponível, em cache, retorne isso, caso contrário, faça a solicitação HTTP.

import {Injectable} from '@angular/core';
import {Http, Headers} from '@angular/http';
import {Observable} from 'rxjs/Observable';
import 'rxjs/add/observable/of'; //proper way to import the 'of' operator
import 'rxjs/add/operator/share';
import 'rxjs/add/operator/map';
import {Data} from './data';

@Injectable()
export class DataService {
  private url: string = 'https://cors-test.appspot.com/test';

  private data: Data;
  private observable: Observable<any>;

  constructor(private http: Http) {}

  getData() {
    if(this.data) {
      // if `data` is available just return it as `Observable`
      return Observable.of(this.data); 
    } else if(this.observable) {
      // if `this.observable` is set then the request is in progress
      // return the `Observable` for the ongoing request
      return this.observable;
    } else {
      // example header (not necessary)
      let headers = new Headers();
      headers.append('Content-Type', 'application/json');
      // create the request, store the `Observable` for subsequent subscribers
      this.observable = this.http.get(this.url, {
        headers: headers
      })
      .map(response =>  {
        // when the cached data is available we don't need the `Observable` reference anymore
        this.observable = null;

        if(response.status == 400) {
          return "FAILURE";
        } else if(response.status == 200) {
          this.data = new Data(response.json());
          return this.data;
        }
        // make it shared so more than one subscriber can get the result
      })
      .share();
      return this.observable;
    }
  }
}

Exemplo de plunker

Este artigo https://blog.thoughtram.io/angular/2018/03/05/advanced-caching-with-rxjs.html é uma ótima explicação sobre como armazenar em cache shareReplay.

Por sugestão do @Cristian, esta é uma maneira que funciona bem para observáveis ​​HTTP, que emitem apenas uma vez e depois são concluídos:

getCustomer() {
    return this.http.get('/someUrl')
        .map(res => res.json()).publishLast().refCount();
}

ATUALIZAÇÃO: Ben Lesh diz que no próximo lançamento menor após a 5.2.0, você poderá chamar apenas shareReplay () para realmente armazenar em cache.

ANTERIORMENTE.....

Em primeiro lugar, não use share () ou publishReplay (1) .refCount (), eles são iguais e o problema é que ele só compartilha se as conexões forem feitas enquanto o observável estiver ativo, se você se conectar após a conclusão , ele cria um novo observável novamente, tradução, sem realmente armazenar em cache.

Birowski deu a solução correta acima, que é usar ReplaySubject. ReplaySubject armazenará em cache os valores que você atribuir (bufferSize) no nosso caso 1. Ele não criará um novo observável como share () quando refCount chegar a zero e você fizer uma nova conexão, que é o comportamento correto para o armazenamento em cache.

Aqui está uma função reutilizável

export function cacheable<T>(o: Observable<T>): Observable<T> {
  let replay = new ReplaySubject<T>(1);
  o.subscribe(
    x => replay.next(x),
    x => replay.error(x),
    () => replay.complete()
  );
  return replay.asObservable();
}

Veja como usá-lo

import { Injectable } from '@angular/core';
import { Http } from '@angular/http';
import { Observable } from 'rxjs/Observable';
import { cacheable } from '../utils/rxjs-functions';

@Injectable()
export class SettingsService {
  _cache: Observable<any>;
  constructor(private _http: Http, ) { }

  refresh = () => {
    if (this._cache) {
      return this._cache;
    }
    return this._cache = cacheable<any>(this._http.get('YOUR URL'));
  }
}

Abaixo está uma versão mais avançada da função que pode ser armazenada em cache. Essa permite ter sua própria tabela de pesquisa + a capacidade de fornecer uma tabela de pesquisa personalizada. Dessa forma, você não precisa verificar isso._cache como no exemplo acima. Observe também que, em vez de passar o observável como o primeiro argumento, você passa uma função que retorna os observáveis; isso ocorre porque o Http do Angular é executado imediatamente; portanto, ao retornar uma função executada preguiçosamente, podemos decidir não chamá-lo se já estiver em nosso cache.

let cacheableCache: { [key: string]: Observable<any> } = {};
export function cacheable<T>(returnObservable: () => Observable<T>, key?: string, customCache?: { [key: string]: Observable<T> }): Observable<T> {
  if (!!key && (customCache || cacheableCache)[key]) {
    return (customCache || cacheableCache)[key] as Observable<T>;
  }
  let replay = new ReplaySubject<T>(1);
  returnObservable().subscribe(
    x => replay.next(x),
    x => replay.error(x),
    () => replay.complete()
  );
  let observable = replay.asObservable();
  if (!!key) {
    if (!!customCache) {
      customCache[key] = observable;
    } else {
      cacheableCache[key] = observable;
    }
  }
  return observable;
}

Uso:

getData() => cacheable(this._http.get("YOUR URL"), "this is key for my cache")

O rxjs 5.4.0 possui um novo método shareReplay .

• rx-book shareReplay ()
• Nenhum documento em reactivex.io/rxjs

O autor diz explicitamente "ideal para lidar com coisas como armazenar em cache resultados AJAX"

rxjs PR # 2443 feat (shareReplay): adiciona shareReplayvariante depublishReplay

shareReplay retorna um observável que é a origem com difusão seletiva em um ReplaySubject. Esse assunto de reprodução é reciclado por erro da fonte, mas não após a conclusão da fonte. Isso torna o shareReplay ideal para lidar com coisas como armazenar em cache os resultados do AJAX, pois é possível tentar novamente. Seu comportamento de repetição, no entanto, difere do compartilhamento, pois não repetirá a fonte observável, mas sim os valores da fonte observável.

de acordo com este artigo

Acontece que podemos adicionar facilmente cache ao observável adicionando publishReplay (1) e refCount.

tão dentro se as declarações apenas acrescentam

.publishReplay(1)
.refCount();

para .map(...)

A versão 5.4.0 do rxjs (09-05-2017) adiciona suporte ao shareReplay .

Por que usar o shareReplay?

Você geralmente deseja usar o shareReplay quando tiver efeitos colaterais ou cálculos tributários que não deseja executar entre vários assinantes. Também pode ser valioso em situações em que você sabe que terá assinantes atrasados ​​em um fluxo que precisa acessar valores emitidos anteriormente. Essa capacidade de reproduzir valores na assinatura é o que diferencia share e shareReplay.

Você pode modificar facilmente um serviço angular para usá-lo e retornar um observável com um resultado em cache que apenas fará a chamada http uma única vez (supondo que a primeira chamada tenha sido bem-sucedida).

Exemplo de serviço angular

Aqui está um serviço ao cliente muito simples que usa shareReplay.

customer.service.ts

import { shareReplay } from 'rxjs/operators';
import { Observable } from 'rxjs';
import { HttpClient } from '@angular/common/http';

@Injectable()
export class CustomerService {

    private readonly _getCustomers: Observable<ICustomer[]>;

    constructor(private readonly http: HttpClient) {
        this._getCustomers = this.http.get<ICustomer[]>('/api/customers/').pipe(shareReplay());
    }

    getCustomers() : Observable<ICustomer[]> {
        return this._getCustomers;
    }
}

export interface ICustomer {
  /* ICustomer interface fields defined here */
}

Observe que a atribuição no construtor pode ser movida para o método, getCustomersmas como os observáveis ​​retornados HttpClientsão "frios", isso é aceitável no construtor, pois a chamada http será feita apenas com a primeira chamada para subscribe.

Além disso, a suposição aqui é que os dados retornados iniciais não ficam obsoletos durante a vida útil da instância do aplicativo.

Eu estrelou a pergunta, mas tentarei tentar.

//this will be the shared observable that 
//anyone can subscribe to, get the value, 
//but not cause an api request
let customer$ = new Rx.ReplaySubject(1);

getCustomer().subscribe(customer$);

//here's the first subscriber
customer$.subscribe(val => console.log('subscriber 1: ' + val));

//here's the second subscriber
setTimeout(() => {
  customer$.subscribe(val => console.log('subscriber 2: ' + val));  
}, 1000);

function getCustomer() {
  return new Rx.Observable(observer => {
    console.log('api request');
    setTimeout(() => {
      console.log('api response');
      observer.next('customer object');
      observer.complete();
    }, 500);
  });
}

Aqui está a prova :)

Há apenas um argumento: getCustomer().subscribe(customer$)

Não estamos assinando a resposta api de getCustomer(), estamos assinando um ReplaySubject que é observável, que também pode se inscrever em um Observable diferente e (e isso é importante) manter seu último valor emitido e republicá-lo em qualquer um deles (ReplaySubject's ) assinantes.

Encontrei uma maneira de armazenar o resultado do http em sessionStorage e usá-lo para a sessão, para que ele nunca chame o servidor novamente.

Eu usei para chamar a API do github para evitar o limite de uso.

@Injectable()
export class HttpCache {
  constructor(private http: Http) {}

  get(url: string): Observable<any> {
    let cached: any;
    if (cached === sessionStorage.getItem(url)) {
      return Observable.of(JSON.parse(cached));
    } else {
      return this.http.get(url)
        .map(resp => {
          sessionStorage.setItem(url, resp.text());
          return resp.json();
        });
    }
  }
}

Para sua informação, o limite de sessionStorage é de 5M (ou 4,75M). Portanto, não deve ser usado assim para um grande conjunto de dados.

------ editar -------------
Se você deseja atualizar os dados com o F5, que usa dados de memória em vez de sessionStorage;

@Injectable()
export class HttpCache {
  cached: any = {};  // this will store data
  constructor(private http: Http) {}

  get(url: string): Observable<any> {
    if (this.cached[url]) {
      return Observable.of(this.cached[url]));
    } else {
      return this.http.get(url)
        .map(resp => {
          this.cached[url] = resp.text();
          return resp.json();
        });
    }
  }
}

A implementação que você escolher dependerá se você deseja cancelar a assinatura () para cancelar sua solicitação HTTP ou não.

De qualquer forma, os decoradores TypeScript são uma ótima maneira de padronizar o comportamento. Este é o que eu escrevi:

  @CacheObservableArgsKey
  getMyThing(id: string): Observable<any> {
    return this.http.get('things/'+id);
  }

Definição do decorador:

/**
 * Decorator that replays and connects to the Observable returned from the function.
 * Caches the result using all arguments to form a key.
 * @param target
 * @param name
 * @param descriptor
 * @returns {PropertyDescriptor}
 */
export function CacheObservableArgsKey(target: Object, name: string, descriptor: PropertyDescriptor) {
  const originalFunc = descriptor.value;
  const cacheMap = new Map<string, any>();
  descriptor.value = function(this: any, ...args: any[]): any {
    const key = args.join('::');

    let returnValue = cacheMap.get(key);
    if (returnValue !== undefined) {
      console.log(`${name} cache-hit ${key}`, returnValue);
      return returnValue;
    }

    returnValue = originalFunc.apply(this, args);
    console.log(`${name} cache-miss ${key} new`, returnValue);
    if (returnValue instanceof Observable) {
      returnValue = returnValue.publishReplay(1);
      returnValue.connect();
    }
    else {
      console.warn('CacheHttpArgsKey: value not an Observable cannot publishReplay and connect', returnValue);
    }
    cacheMap.set(key, returnValue);
    return returnValue;
  };

  return descriptor;
}

Dados de resposta HTTP em cache usando Rxjs Observer / Observable + Caching + Subscription

Veja o código abaixo

* aviso de isenção de responsabilidade: sou novo no rxjs, portanto, lembre-se de que posso estar abusando da abordagem observável / observador. Minha solução é puramente um conglomerado de outras soluções que encontrei e é a conseqüência de não conseguir encontrar uma solução simples e bem documentada. Portanto, estou fornecendo minha solução completa de código (como gostaria de ter encontrado) na esperança de que ajude outras pessoas.

* observe que essa abordagem se baseia livremente no GoogleFirebaseObservables. Infelizmente, não tenho a experiência / tempo adequados para replicar o que eles fizeram sob o capô. Mas a seguir, é uma maneira simplista de fornecer acesso assíncrono a alguns dados capazes de cache.

Situação : Um componente 'lista de produtos' é encarregado de exibir uma lista de produtos. O site é um aplicativo da web de página única com alguns botões de menu que filtram os produtos exibidos na página.

Solução : O componente "assina" um método de serviço. O método de serviço retorna uma matriz de objetos do produto, que o componente acessa através do retorno de chamada da assinatura. O método de serviço agrupa sua atividade em um Observador recém-criado e retorna o observador. Dentro desse observador, ele procura dados em cache e os repassa para o assinante (o componente) e retorna. Caso contrário, emite uma chamada http para recuperar os dados, assina a resposta, onde você pode processar esses dados (por exemplo, mapear os dados para o seu próprio modelo) e depois repassá-los ao assinante.

O código

product-list.component.ts

import { Component, OnInit, Input } from '@angular/core';
import { ProductService } from '../../../services/product.service';
import { Product, ProductResponse } from '../../../models/Product';

@Component({
  selector: 'app-product-list',
  templateUrl: './product-list.component.html',
  styleUrls: ['./product-list.component.scss']
})
export class ProductListComponent implements OnInit {
  products: Product[];

  constructor(
    private productService: ProductService
  ) { }

  ngOnInit() {
    console.log('product-list init...');
    this.productService.getProducts().subscribe(products => {
      console.log('product-list received updated products');
      this.products = products;
    });
  }
}

product.service.ts

import { Injectable } from '@angular/core';
import { Http, Headers } from '@angular/http';
import { Observable, Observer } from 'rxjs';
import 'rxjs/add/operator/map';
import { Product, ProductResponse } from '../models/Product';

@Injectable()
export class ProductService {
  products: Product[];

  constructor(
    private http:Http
  ) {
    console.log('product service init.  calling http to get products...');

  }

  getProducts():Observable<Product[]>{
    //wrap getProducts around an Observable to make it async.
    let productsObservable$ = Observable.create((observer: Observer<Product[]>) => {
      //return products if it was previously fetched
      if(this.products){
        console.log('## returning existing products');
        observer.next(this.products);
        return observer.complete();

      }
      //Fetch products from REST API
      console.log('** products do not yet exist; fetching from rest api...');
      let headers = new Headers();
      this.http.get('http://localhost:3000/products/',  {headers: headers})
      .map(res => res.json()).subscribe((response:ProductResponse) => {
        console.log('productResponse: ', response);
        let productlist = Product.fromJsonList(response.products); //convert service observable to product[]
        this.products = productlist;
        observer.next(productlist);
      });
    }); 
    return productsObservable$;
  }
}

product.ts (o modelo)

export interface ProductResponse {
  success: boolean;
  msg: string;
  products: Product[];
}

export class Product {
  product_id: number;
  sku: string;
  product_title: string;
  ..etc...

  constructor(product_id: number,
    sku: string,
    product_title: string,
    ...etc...
  ){
    //typescript will not autoassign the formal parameters to related properties for exported classes.
    this.product_id = product_id;
    this.sku = sku;
    this.product_title = product_title;
    ...etc...
  }



  //Class method to convert products within http response to pure array of Product objects.
  //Caller: product.service:getProducts()
  static fromJsonList(products:any): Product[] {
    let mappedArray = products.map(Product.fromJson);
    return mappedArray;
  }

  //add more parameters depending on your database entries and constructor
  static fromJson({ 
      product_id,
      sku,
      product_title,
      ...etc...
  }): Product {
    return new Product(
      product_id,
      sku,
      product_title,
      ...etc...
    );
  }
}

Aqui está um exemplo da saída que vejo quando carrego a página no Chrome. Observe que, no carregamento inicial, os produtos são buscados a partir de http (chame meu serviço de descanso de nó, que está sendo executado localmente na porta 3000). Quando clico para navegar para uma exibição 'filtrada' dos produtos, eles são encontrados no cache.

Meu registro do Chrome (console):

core.es5.js:2925 Angular is running in the development mode. Call enableProdMode() to enable the production mode.
app.component.ts:19 app.component url: /products
product.service.ts:15 product service init.  calling http to get products...
product-list.component.ts:18 product-list init...
product.service.ts:29 ** products do not yet exist; fetching from rest api...
product.service.ts:33 productResponse:  {success: true, msg: "Products found", products: Array(23)}
product-list.component.ts:20 product-list received updated products

... [clicou em um botão de menu para filtrar os produtos] ...

app.component.ts:19 app.component url: /products/chocolatechip
product-list.component.ts:18 product-list init...
product.service.ts:24 ## returning existing products
product-list.component.ts:20 product-list received updated products

Conclusão: Essa é a maneira mais simples que eu encontrei (até agora) de implementar dados de resposta http em cache. No meu aplicativo angular, cada vez que navego para uma visualização diferente dos produtos, o componente da lista de produtos é recarregado. O ProductService parece ser uma instância compartilhada; portanto, o cache local de 'products: Product []' no ProductService é mantido durante a navegação e as chamadas subseqüentes a "GetProducts ()" retornam o valor em cache. Uma observação final, eu li comentários sobre como as observáveis ​​/ assinaturas precisam ser fechadas quando você terminar para evitar 'vazamentos de memória'. Eu não incluí isso aqui, mas é algo a ter em mente.

Suponho que @ ngx-cache / core possa ser útil para manter os recursos de armazenamento em cache para as chamadas http, especialmente se a chamada HTTP for feita nas plataformas de navegador e servidor .

Digamos que temos o seguinte método:

getCustomer() {
  return this.http.get('/someUrl').map(res => res.json());
}

Você pode usar o Cacheddecorador de @ NGX-cache / núcleo para armazenar o valor retornado do método de fazer a chamada HTTP na cache storage( o storagepode ser configurável, verifique a implementação em ng-semente / universal ) - à direita na primeira execução. Nas próximas vezes que o método for chamado (não importa no navegador ou na plataforma do servidor ), o valor será recuperado do cache storage.

import { Cached } from '@ngx-cache/core';

...

@Cached('get-customer') // the cache key/identifier
getCustomer() {
  return this.http.get('/someUrl').map(res => res.json());
}

Há também a possibilidade de métodos de utilização de armazenamento em cache ( has, get, set) usando o cache API .

anyclass.ts

...
import { CacheService } from '@ngx-cache/core';

@Injectable()
export class AnyClass {
  constructor(private readonly cache: CacheService) {
    // note that CacheService is injected into a private property of AnyClass
  }

  // will retrieve 'some string value'
  getSomeStringValue(): string {
    if (this.cache.has('some-string'))
      return this.cache.get('some-string');

    this.cache.set('some-string', 'some string value');
    return 'some string value';
  }
}

Aqui está a lista de pacotes, tanto para o cache do lado do cliente quanto do lado do servidor:

• @ ngx-cache / core : utilitário de cache
• @ ngx-cache / platform-browser : implementação da plataforma SPA / Browser
• @ ngx-cache / platform-server : implementação da plataforma do servidor
• @ ngx-cache / fs-storage : utilitário de armazenamento (necessário para a plataforma do servidor)

rxjs 5.3.0

Eu não fui feliz com .map(myFunction).publishReplay(1).refCount()

Com vários assinantes, .map()executa myFunctionduas vezes em alguns casos (espero que seja executado apenas uma vez). Uma correção parece serpublishReplay(1).refCount().take(1)

Outra coisa que você pode fazer é simplesmente não usar refCount()e aquecer o Observable imediatamente:

let obs = this.http.get('my/data.json').publishReplay(1);
obs.connect();
return obs;

Isso iniciará a solicitação HTTP, independentemente dos assinantes. Não tenho certeza se o cancelamento da inscrição antes da conclusão do HTTP GET o cancelará ou não.

O que queremos fazer é garantir que isso não cause várias solicitações de rede.

Meu favorito pessoal é usar asyncmétodos para chamadas que fazem solicitações de rede. Os métodos em si não retornam um valor; em vez disso, atualizam um BehaviorSubjectno mesmo serviço, no qual os componentes se inscreverão.

Agora, por que usar um em BehaviorSubjectvez de um Observable? Porque,

• Na assinatura, BehaviorSubject retorna o último valor, enquanto A observável regular somente é acionado quando recebe um onnext.
• Se você deseja recuperar o último valor do BehaviorSubject em um código não observável (sem uma assinatura), você pode usar o getValue()método

Exemplo:

customer.service.ts

public customers$: BehaviorSubject<Customer[]> = new BehaviorSubject([]);

public async getCustomers(): Promise<void> {
    let customers = await this.httpClient.post<LogEntry[]>(this.endPoint, criteria).toPromise();
    if (customers) 
        this.customers$.next(customers);
}

Então, sempre que necessário, podemos apenas assinar customers$.

public ngOnInit(): void {
    this.customerService.customers$
    .subscribe((customers: Customer[]) => this.customerList = customers);
}

Ou talvez você queira usá-lo diretamente em um modelo

<li *ngFor="let customer of customerService.customers$ | async"> ... </li>

Portanto, agora, até que você faça outra chamada getCustomers, os dados são retidos no customers$BehaviorSubject.

E daí se você deseja atualizar esses dados? basta ligar paragetCustomers()

public async refresh(): Promise<void> {
    try {
      await this.customerService.getCustomers();
    } 
    catch (e) {
      // request failed, handle exception
      console.error(e);
    }
}

Usando esse método, não precisamos reter explicitamente os dados entre as chamadas de rede subsequentes, pois elas são manipuladas pelo BehaviorSubject.

PS: Normalmente, quando um componente é destruído, é uma boa prática se livrar das assinaturas, para isso você pode usar o método sugerido nesta resposta.

Ótimas respostas.

Ou você pode fazer isso:

Esta é a versão mais recente do rxjs. Eu estou usando 5.5.7 versão do RxJS

import {share} from "rxjs/operators";

this.http.get('/someUrl').pipe(share());

Basta ligar para share () após o mapa e antes de qualquer assinatura .

No meu caso, eu tenho um serviço genérico (RestClientService.ts) que está fazendo a chamada restante, extraindo dados, verificando erros e retornando observável a um serviço de implementação concreto (por exemplo: ContractClientService.ts), finalmente essa implementação concreta retorna observável para ContractComponent.ts e este se inscreve para atualizar a exibição.

RestClientService.ts:

export abstract class RestClientService<T extends BaseModel> {

      public GetAll = (path: string, property: string): Observable<T[]> => {
        let fullPath = this.actionUrl + path;
        let observable = this._http.get(fullPath).map(res => this.extractData(res, property));
        observable = observable.share();  //allows multiple subscribers without making again the http request
        observable.subscribe(
          (res) => {},
          error => this.handleError2(error, "GetAll", fullPath),
          () => {}
        );
        return observable;
      }

  private extractData(res: Response, property: string) {
    ...
  }
  private handleError2(error: any, method: string, path: string) {
    ...
  }

}

ContractService.ts:

export class ContractService extends RestClientService<Contract> {
  private GET_ALL_ITEMS_REST_URI_PATH = "search";
  private GET_ALL_ITEMS_PROPERTY_PATH = "contract";
  public getAllItems(): Observable<Contract[]> {
    return this.GetAll(this.GET_ALL_ITEMS_REST_URI_PATH, this.GET_ALL_ITEMS_PROPERTY_PATH);
  }

}

ContractComponent.ts:

export class ContractComponent implements OnInit {

  getAllItems() {
    this.rcService.getAllItems().subscribe((data) => {
      this.items = data;
   });
  }

}

Eu escrevi uma classe de cache,

/**
 * Caches results returned from given fetcher callback for given key,
 * up to maxItems results, deletes the oldest results when full (FIFO).
 */
export class StaticCache
{
    static cachedData: Map<string, any> = new Map<string, any>();
    static maxItems: number = 400;

    static get(key: string){
        return this.cachedData.get(key);
    }

    static getOrFetch(key: string, fetcher: (string) => any): any {
        let value = this.cachedData.get(key);

        if (value != null){
            console.log("Cache HIT! (fetcher)");
            return value;
        }

        console.log("Cache MISS... (fetcher)");
        value = fetcher(key);
        this.add(key, value);
        return value;
    }

    static add(key, value){
        this.cachedData.set(key, value);
        this.deleteOverflowing();
    }

    static deleteOverflowing(): void {
        if (this.cachedData.size > this.maxItems) {
            this.deleteOldest(this.cachedData.size - this.maxItems);
        }
    }

    /// A Map object iterates its elements in insertion order — a for...of loop returns an array of [key, value] for each iteration.
    /// However that seems not to work. Trying with forEach.
    static deleteOldest(howMany: number): void {
        //console.debug("Deleting oldest " + howMany + " of " + this.cachedData.size);
        let iterKeys = this.cachedData.keys();
        let item: IteratorResult<string>;
        while (howMany-- > 0 && (item = iterKeys.next(), !item.done)){
            //console.debug("    Deleting: " + item.value);
            this.cachedData.delete(item.value); // Deleting while iterating should be ok in JS.
        }
    }

    static clear(): void {
        this.cachedData = new Map<string, any>();
    }

}

É tudo estático por causa de como o usamos, mas fique à vontade para torná-lo uma classe e um serviço normais. Não tenho certeza se angular mantém uma única instância o tempo todo (novo no Angular2).

E é assim que eu uso:

            let httpService: Http = this.http;
            function fetcher(url: string): Observable<any> {
                console.log("    Fetching URL: " + url);
                return httpService.get(url).map((response: Response) => {
                    if (!response) return null;
                    if (typeof response.json() !== "array")
                        throw new Error("Graph REST should return an array of vertices.");
                    let items: any[] = graphService.fromJSONarray(response.json(), httpService);
                    return array ? items : items[0];
                });
            }

            // If data is a link, return a result of a service call.
            if (this.data[verticesLabel][name]["link"] || this.data[verticesLabel][name]["_type"] == "link")
            {
                // Make an HTTP call.
                let url = this.data[verticesLabel][name]["link"];
                let cachedObservable: Observable<any> = StaticCache.getOrFetch(url, fetcher);
                if (!cachedObservable)
                    throw new Error("Failed loading link: " + url);
                return cachedObservable;
            }

Suponho que poderia haver uma maneira mais inteligente, que usaria alguns Observabletruques, mas isso era bom para meus propósitos.

Basta usar essa camada de cache, ele faz tudo o que você precisa e até gerencia o cache para solicitações de ajax.

http://www.ravinderpayal.com/blogs/12Jan2017-Ajax-Cache-Mangement-Angular2-Service.html

É muito fácil de usar

@Component({
    selector: 'home',
    templateUrl: './html/home.component.html',
    styleUrls: ['./css/home.component.css'],
})
export class HomeComponent {
    constructor(AjaxService:AjaxService){
        AjaxService.postCache("/api/home/articles").subscribe(values=>{console.log(values);this.articles=values;});
    }

    articles={1:[{data:[{title:"first",sort_text:"description"},{title:"second",sort_text:"description"}],type:"Open Source Works"}]};
}

A camada (como um serviço angular injetável) é

import { Injectable }     from '@angular/core';
import { Http, Response} from '@angular/http';
import { Observable }     from 'rxjs/Observable';
import './../rxjs/operator'
@Injectable()
export class AjaxService {
    public data:Object={};
    /*
    private dataObservable:Observable<boolean>;
     */
    private dataObserver:Array<any>=[];
    private loading:Object={};
    private links:Object={};
    counter:number=-1;
    constructor (private http: Http) {
    }
    private loadPostCache(link:string){
     if(!this.loading[link]){
               this.loading[link]=true;
               this.links[link].forEach(a=>this.dataObserver[a].next(false));
               this.http.get(link)
                   .map(this.setValue)
                   .catch(this.handleError).subscribe(
                   values => {
                       this.data[link] = values;
                       delete this.loading[link];
                       this.links[link].forEach(a=>this.dataObserver[a].next(false));
                   },
                   error => {
                       delete this.loading[link];
                   }
               );
           }
    }

    private setValue(res: Response) {
        return res.json() || { };
    }

    private handleError (error: Response | any) {
        // In a real world app, we might use a remote logging infrastructure
        let errMsg: string;
        if (error instanceof Response) {
            const body = error.json() || '';
            const err = body.error || JSON.stringify(body);
            errMsg = `${error.status} - ${error.statusText || ''} ${err}`;
        } else {
            errMsg = error.message ? error.message : error.toString();
        }
        console.error(errMsg);
        return Observable.throw(errMsg);
    }

    postCache(link:string): Observable<Object>{

         return Observable.create(observer=> {
             if(this.data.hasOwnProperty(link)){
                 observer.next(this.data[link]);
             }
             else{
                 let _observable=Observable.create(_observer=>{
                     this.counter=this.counter+1;
                     this.dataObserver[this.counter]=_observer;
                     this.links.hasOwnProperty(link)?this.links[link].push(this.counter):(this.links[link]=[this.counter]);
                     _observer.next(false);
                 });
                 this.loadPostCache(link);
                 _observable.subscribe(status=>{
                     if(status){
                         observer.next(this.data[link]);
                     }
                     }
                 );
             }
            });
        }
}

É .publishReplay(1).refCount();ou .publishLast().refCount();desde que os observáveis ​​do Angular Http sejam concluídos após solicitação.

Essa classe simples armazena em cache o resultado para que você possa se inscrever em .value várias vezes e faz apenas uma solicitação. Você também pode usar .reload () para fazer nova solicitação e publicar dados.

Você pode usá-lo como:

let res = new RestResource(() => this.http.get('inline.bundleo.js'));

res.status.subscribe((loading)=>{
    console.log('STATUS=',loading);
});

res.value.subscribe((value) => {
  console.log('VALUE=', value);
});

e a fonte:

export class RestResource {

  static readonly LOADING: string = 'RestResource_Loading';
  static readonly ERROR: string = 'RestResource_Error';
  static readonly IDLE: string = 'RestResource_Idle';

  public value: Observable<any>;
  public status: Observable<string>;
  private loadStatus: Observer<any>;

  private reloader: Observable<any>;
  private reloadTrigger: Observer<any>;

  constructor(requestObservableFn: () => Observable<any>) {
    this.status = Observable.create((o) => {
      this.loadStatus = o;
    });

    this.reloader = Observable.create((o: Observer<any>) => {
      this.reloadTrigger = o;
    });

    this.value = this.reloader.startWith(null).switchMap(() => {
      if (this.loadStatus) {
        this.loadStatus.next(RestResource.LOADING);
      }
      return requestObservableFn()
        .map((res) => {
          if (this.loadStatus) {
            this.loadStatus.next(RestResource.IDLE);
          }
          return res;
        }).catch((err)=>{
          if (this.loadStatus) {
            this.loadStatus.next(RestResource.ERROR);
          }
          return Observable.of(null);
        });
    }).publishReplay(1).refCount();
  }

  reload() {
    this.reloadTrigger.next(null);
  }

}

Você pode criar uma classe simples Cacheable <> que ajude a gerenciar os dados recuperados do servidor http com vários assinantes:

declare type GetDataHandler<T> = () => Observable<T>;

export class Cacheable<T> {

    protected data: T;
    protected subjectData: Subject<T>;
    protected observableData: Observable<T>;
    public getHandler: GetDataHandler<T>;

    constructor() {
      this.subjectData = new ReplaySubject(1);
      this.observableData = this.subjectData.asObservable();
    }

    public getData(): Observable<T> {
      if (!this.getHandler) {
        throw new Error("getHandler is not defined");
      }
      if (!this.data) {
        this.getHandler().map((r: T) => {
          this.data = r;
          return r;
        }).subscribe(
          result => this.subjectData.next(result),
          err => this.subjectData.error(err)
        );
      }
      return this.observableData;
    }

    public resetCache(): void {
      this.data = null;
    }

    public refresh(): void {
      this.resetCache();
      this.getData();
    }

}

Uso

Declare o objeto Cacheable <> (presumivelmente como parte do serviço):

list: Cacheable<string> = new Cacheable<string>();

e manipulador:

this.list.getHandler = () => {
// get data from server
return this.http.get(url)
.map((r: Response) => r.json() as string[]);
}

Chamada de um componente:

//gets data from server
List.getData().subscribe(…)

Você pode ter vários componentes inscritos nele.

Mais detalhes e exemplo de código estão aqui: http://devinstance.net/articles/20171021/rxjs-cacheable

Você pode simplesmente usar o ngx-cacheable ! É melhor para o seu cenário.

O benefício de usar este

• Ele chama API rest apenas uma vez, armazena em cache a resposta e retorna o mesmo para as seguintes solicitações.
• Pode chamar a API conforme necessário após a operação de criação / atualização / exclusão.

Portanto, sua classe de serviço seria algo como isto -

import { Injectable } from '@angular/core';
import { Cacheable, CacheBuster } from 'ngx-cacheable';

const customerNotifier = new Subject();

@Injectable()
export class customersService {

    // relieves all its caches when any new value is emitted in the stream using notifier
    @Cacheable({
        cacheBusterObserver: customerNotifier,
        async: true
    })
    getCustomer() {
        return this.http.get('/someUrl').map(res => res.json());
    }

    // notifies the observer to refresh the data
    @CacheBuster({
        cacheBusterNotifier: customerNotifier
    })
    addCustomer() {
        // some code
    }

    // notifies the observer to refresh the data
    @CacheBuster({
        cacheBusterNotifier: customerNotifier
    })
    updateCustomer() {
        // some code
    }
}

Aqui está o link para mais referência.

Você já tentou executar o código que já possui?

Como você está construindo o Observável a partir da promessa resultante getJSON(), a solicitação de rede é feita antes que alguém assine. E a promessa resultante é compartilhada por todos os assinantes.

var promise = jQuery.getJSON(requestUrl); // network call is executed now
var o = Rx.Observable.fromPromise(promise); // just wraps it in an observable
o.subscribe(...); // does not trigger network call
o.subscribe(...); // does not trigger network call
// ...