Как правильно передать результат сетевого вызова Angular Http в RxJs 5?

Используя Http, мы вызываем метод, который выполняет сетевой вызов и возвращает наблюдаемый http:

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

Если мы возьмем это наблюдаемое и добавим к нему несколько подписчиков:

let network$ = getCustomer();

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

Что мы хотим сделать, так это убедиться, что это не вызывает многократных сетевых запросов.

Это может показаться необычным сценарием, но на самом деле он довольно распространен: например, если вызывающий абонент подписывается на наблюдаемое для отображения сообщения об ошибке и передает его в шаблон с помощью асинхронного канала, у нас уже есть два подписчика.

Как правильно сделать это в RxJs 5?

А именно, это, кажется, работает нормально:

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

Но это идиоматический способ сделать это в RxJs 5, или мы должны вместо этого сделать что-то еще?

Примечание: согласно Angular 5 new HttpClient, .map(res => res.json())часть во всех примерах теперь бесполезна, так как результат JSON теперь принят по умолчанию.

Кэшируйте данные и, если доступно, кешируйте, возвращайте это, иначе сделайте 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;
    }
  }
}

Пример плунжера

Эта статья https://blog.thoughtram.io/angular/2018/03/05/advanced-caching-with-rxjs.html является отличным объяснением того, как кэшировать shareReplay.

Согласно предложению @Cristian, это один из способов, который хорошо работает для наблюдаемых HTTP, которые излучают только один раз, а затем завершают:

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

ОБНОВЛЕНИЕ: Бен Леш говорит, что в следующем небольшом выпуске после 5.2.0 вы сможете просто вызвать функцию shareReplay () для кеширования.

РАНЕЕ .....

Во-первых, не используйте share () или publishReplay (1) .refCount (), они одинаковы, и проблема с ним в том, что он разделяет, только если установлены соединения, когда наблюдаемая активна, если вы подключаетесь после ее завершения , он снова создает новый наблюдаемый, перевод, а не кеширование.

Birowski дал правильное решение выше, которое заключается в использовании ReplaySubject. ReplaySubject будет кэшировать значения, которые вы ему даете (bufferSize) в нашем случае 1. Он не создаст новую наблюдаемую, например, share (), когда refCount достигнет нуля, и вы создадите новое соединение, что является правильным поведением для кэширования.

Вот многоразовая функция

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

Вот как это использовать

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

Ниже приведена более продвинутая версия кешируемой функции. Эта позволяет иметь собственную таблицу поиска + возможность предоставить пользовательскую таблицу поиска. Таким образом, вам не нужно проверять this._cache, как в примере выше. Также обратите внимание, что вместо передачи наблюдаемой в качестве первого аргумента, вы передаете функцию, которая возвращает наблюдаемые, это потому, что Http Angular выполняет сразу, поэтому, возвращая лениво выполненную функцию, мы можем решить не вызывать ее, если она уже находится в наш кеш.

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

Использование:

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

В rxjs 5.4.0 появился новый метод shareReplay .

• rx-book shareReplay ()
• Нет документов на сайте реактив.

Автор прямо говорит, что «идеально подходит для обработки таких вещей, как кэширование результатов AJAX»

rxjs PR # 2443 feat (shareReplay): добавляет shareReplayвариантpublishReplay

shareReplay возвращает наблюдаемое, являющееся источником многоадресной передачи через ReplaySubject. Эта тема воспроизведения перерабатывается при ошибке источника, но не при завершении источника. Это делает shareReplay идеальным для обработки таких вещей, как кэширование результатов AJAX, так как оно повторяется. Это повторное поведение, однако, отличается от общего тем, что оно не будет повторять наблюдаемый источник, а будет повторять значения наблюдаемой источника.

в соответствии с этой статьей

Оказывается, мы можем легко добавить кеширование в наблюдаемое, добавив publishReplay (1) и refCount.

так внутри, если заявления просто добавить

.publishReplay(1)
.refCount();

в .map(...)

В версии 5.4.0 rxjs (2017-05-09) добавлена ​​поддержка shareReplay .

Зачем использовать shareReplay?

Как правило, вы хотите использовать shareReplay, когда у вас есть побочные эффекты или налоговые вычисления, которые вы не хотите выполнять среди нескольких подписчиков. Это также может быть полезно в ситуациях, когда вы знаете, что у вас будут поздние подписчики на поток, которому необходим доступ к ранее выданным значениям. Эта способность воспроизводить значения в подписке - это то, что отличает share и shareReplay.

Вы можете легко изменить угловой сервис, чтобы использовать его и возвращать наблюдаемое с кэшированным результатом, который только когда-либо сделает вызов http один раз (при условии, что первый вызов был успешным).

Пример угловой службы

Вот очень простое обслуживание клиентов, которое использует 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 */
}

Обратите внимание, что присваивание в конструкторе может быть перемещено в метод, getCustomersно поскольку наблюдаемые, возвращаемые из HttpClient"холодного", делать это в конструкторе допустимо, поскольку вызов http будет выполняться только каждый раз при первом вызове subscribe.

Также предполагается, что исходные возвращаемые данные не устаревают в течение времени жизни экземпляра приложения.

Я пометил вопрос, но постараюсь попробовать.

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

Вот доказательство :)

Есть только один вынос: getCustomer().subscribe(customer$)

Мы не подписываемся на ответ API getCustomer(), мы подписываемся на ReplaySubject, который является наблюдаемым, который также может подписаться на другой Observable и (и это важно) удерживать его последнее переданное значение и повторно опубликовать его на любом из его (ReplaySubject's ) подписчики.

Я нашел способ сохранить результат http get в sessionStorage и использовать его для сеанса, чтобы он никогда больше не вызывал сервер.

Я использовал его для вызова github API, чтобы избежать ограничения использования.

@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();
        });
    }
  }
}

К вашему сведению, предел сессии - 5 млн. (Или 4,75 млн.). Таким образом, он не должен использоваться таким образом для большого набора данных.

------ edit -------------
Если вы хотите обновить данные с помощью F5, которая использует данные памяти вместо 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();
        });
    }
  }
}

Реализация, которую вы выберете, будет зависеть от того, хотите ли вы отменить подписку (), чтобы отменить ваш HTTP-запрос или нет.

В любом случае, декораторы TypeScript - хороший способ стандартизировать поведение. Это тот, который я написал:

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

Определение декоратора:

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

Кэшируемые данные ответов HTTP с использованием Rxjs Observer / Observable + Caching + Subscription

Смотрите код ниже

* Отказ от ответственности: я новичок в rxjs, так что имейте в виду, что я могу неправильно использовать подход наблюдаемого / наблюдателя. Мое решение - это просто конгломерат других решений, которые я нашел, и является следствием того, что не удалось найти простое, хорошо документированное решение. Таким образом, я предоставляю свое полное решение для кода (как я хотел бы найти) в надежде, что оно поможет другим.

* обратите внимание, что этот подход основан на GoogleFirebaseObservables. К сожалению, мне не хватает надлежащего опыта / времени, чтобы повторить то, что они сделали под капотом. Но ниже приведен упрощенный способ предоставления асинхронного доступа к некоторым данным в кэше.

Ситуация : компоненту «список товаров» поручено отображение списка товаров. Сайт представляет собой одностраничное веб-приложение с некоторыми кнопками меню, которые будут «фильтровать» продукты, отображаемые на странице.

Решение : компонент «подписывается» на сервисный метод. Метод service возвращает массив объектов продукта, к которым компонент обращается через обратный вызов подписки. Сервисный метод оборачивает свою активность во вновь созданном Обозревателе и возвращает наблюдателя. Внутри этого наблюдателя он ищет кэшированные данные и передает их обратно подписчику (компоненту) и возвращает. В противном случае он выполняет http-вызов для извлечения данных, подписывается на ответ, где вы можете обработать эти данные (например, сопоставить данные с вашей собственной моделью), а затем передать данные обратно подписчику.

Код

продакт-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 (модель)

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

Вот пример вывода, который я вижу, когда загружаю страницу в Chrome. Обратите внимание, что при первоначальной загрузке продукты выбираются из http (обратитесь к моей службе отдыха узла, которая работает локально на порту 3000). Когда я нажимаю, чтобы перейти к «фильтрованному» представлению товаров, товары находятся в кеше.

Мой журнал Chrome (консоль):

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

... [нажал кнопку меню для фильтрации продуктов] ...

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

Вывод: это самый простой способ, который я нашел (на данный момент) для реализации кэшируемых данных ответов http. В моем угловом приложении каждый раз, когда я перехожу к другому виду продуктов, компонент списка продуктов перезагружается. ProductService, по-видимому, является общим экземпляром, поэтому локальный кэш «products: Product []» в ProductService сохраняется во время навигации, а последующие вызовы «GetProducts ()» возвращают кэшированное значение. В заключение я прочитал комментарии о том, как наблюдаемые / подписки должны быть закрыты, когда вы закончите, чтобы предотвратить «утечки памяти». Я не включил это здесь, но об этом нужно помнить.

Я предполагаю, что @ ngx-cache / core может быть полезен для поддержки функций кэширования для http-вызовов, особенно если HTTP-вызов выполняется как на браузерной, так и на серверной платформах.

Допустим, у нас есть следующий метод:

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

Вы можете использовать Cachedдекоратор @ NGX-кэш / ядер для хранения возвращаемого значения из метода , делающего HTTP заходящих в cache storage( может быть настраиваемыми, пожалуйста , проверьте реализацию на нг семечек / универсальные ) - справа на первое исполнении. В следующий раз, когда метод вызывается (независимо от браузера или серверной платформы), значение извлекается изstoragecache storage .

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

...

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

Там также возможность использования методов кэширования ( has, get, set) с помощью кэширования 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';
  }
}

Вот список пакетов, как для кэширования на стороне клиента, так и на стороне сервера:

• @ NGX-кэш / ядро : утилита кеша
• @ NGX-кэш / платформа браузер : реализация платформы SPA / браузера
• @ NGX-кэш / платформы сервера : реализация серверной платформы
• @ ngx-cache / fs-storage : утилита хранения (требуется для серверной платформы)

rxjs 5.3.0

Я не был доволен .map(myFunction).publishReplay(1).refCount()

С несколькими подписчиками, .map()выполняется myFunctionдважды в некоторых случаях (я ожидаю, что он будет выполняться только один раз). Кажется, одно исправлениеpublishReplay(1).refCount().take(1)

Еще одна вещь, которую вы можете сделать, это просто не использовать refCount()и сразу же сделать Observable горячим:

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

Это запустит HTTP-запрос независимо от подписчиков. Я не уверен, отменит ли отмена подписки до завершения HTTP GET или нет.

Что мы хотим сделать, так это убедиться, что это не вызывает многократных сетевых запросов.

Мой личный фаворит - использовать asyncметоды для звонков, которые делают сетевые запросы. Сами методы не возвращают значение, вместо этого они обновляют BehaviorSubjectвнутри одной и той же службы, на которую будут подписываться компоненты.

Теперь зачем использовать BehaviorSubjectвместо Observable? Так как,

• При подписке BehaviorSubject возвращает последнее значение, тогда как обычная наблюдаемая срабатывает только при получении onnext .
• Если вы хотите получить последнее значение BehaviorSubject в ненаблюдаемом коде (без подписки), вы можете использовать getValue()метод.

Пример:

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

Тогда, где требуется, мы можем просто подписаться на customers$.

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

Или, может быть, вы хотите использовать его непосредственно в шаблоне

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

Так что теперь, пока вы не сделаете еще один вызов getCustomers, данные сохраняются в customers$BehaviorSubject.

Так что, если вы хотите обновить эти данные? просто позвонитеgetCustomers()

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

Используя этот метод, нам не нужно явно сохранять данные между последующими сетевыми вызовами, поскольку они обрабатываются BehaviorSubject.

PS: Обычно, когда компонент уничтожается, хорошей практикой является избавление от подписок, для этого вы можете использовать метод, предложенный в этом ответе.

Отличные ответы.

Или вы могли бы сделать это:

Это из последней версии rxjs. Я использую 5.5.7 версию RxJS

import {share} from "rxjs/operators";

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

Просто вызовите share () после map и перед любой подпиской .

В моем случае у меня есть универсальный сервис (RestClientService.ts), который выполняет вызов rest, извлекает данные, проверяет на наличие ошибок и возвращает наблюдаемый в конкретную службу реализации (например, ContractClientService.ts), наконец, эту конкретную реализацию. возвращает наблюдаемый в de ContractComponent.ts, и этот подписывается, чтобы обновить представление.

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

}

Я написал кеш-класс,

/**
 * 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>();
    }

}

Это все статично из-за того, как мы его используем, но не стесняйтесь сделать его обычным классом и сервисом. Я не уверен, что Angular хранит один экземпляр за все время (новичок в Angular2).

И вот как я это использую:

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

Я предполагаю, что мог бы быть более умный способ, который использовал бы некоторые Observableуловки, но это было просто хорошо для моих целей.

Просто используйте этот слой кеша, он делает все, что вам нужно, и даже управляйте кешем для запросов ajax.

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

Это очень легко использовать

@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"}]};
}

Слой (как инъекционный угловой сервис)

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();или .publishLast().refCount();так как Angular Http наблюдаемые завершены после запроса.

Этот простой класс кэширует результат, поэтому вы можете подписаться на .value много раз и делает только 1 запрос. Вы также можете использовать .reload (), чтобы сделать новый запрос и опубликовать данные.

Вы можете использовать его как:

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

и источник:

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

}

Вы можете создать простой класс Cacheable <>, который помогает управлять данными, полученными с http-сервера несколькими подписчиками:

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

}

использование

Объявите объект Cacheable <> (предположительно, как часть службы):

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

и обработчик:

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

Звонок из компонента:

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

Вы можете подписаться на несколько компонентов.

Более подробная информация и пример кода находятся здесь: http://devinstance.net/articles/20171021/rxjs-cacheable

Вы можете просто использовать ngx-cacheable ! Это лучше подходит для вашего сценария.

Преимущество использования этого

• Он вызывает rest API только один раз, кэширует ответ и возвращает то же самое для следующих запросов.
• Может вызывать API по мере необходимости после создания / обновления / удаления.

Итак, ваш класс обслуживания будет что-то вроде этого -

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

Вот ссылка для получения дополнительной ссылки.

Вы пробовали запустить код, который у вас уже есть?

Поскольку вы создаете Observable из обещания, полученного в результате getJSON(), сетевой запрос выполняется до того, как кто-либо подпишется. И полученное обещание разделяют все подписчики.

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
// ...