Apa cara yang benar untuk membagikan hasil panggilan jaringan Angular Http di RxJs 5?

Dengan menggunakan Http, kami memanggil metode yang melakukan panggilan jaringan dan mengembalikan http yang dapat diamati:

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

Jika kami menerima hal ini dan menambahkan beberapa pelanggan ke dalamnya:

let network$ = getCustomer();

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

Apa yang ingin kami lakukan adalah memastikan bahwa ini tidak menyebabkan beberapa permintaan jaringan.

Ini mungkin tampak seperti skenario yang tidak biasa, tetapi sebenarnya cukup umum: misalnya jika pemanggil berlangganan yang diamati untuk menampilkan pesan kesalahan, dan meneruskannya ke templat menggunakan pipa async, kami sudah memiliki dua pelanggan.

Apa cara yang benar untuk melakukan itu di RxJs 5?

Yaitu, ini tampaknya berfungsi dengan baik:

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

Tetapi apakah ini cara idiomatis untuk melakukan ini di RxJs 5, atau haruskah kita melakukan sesuatu yang lain?

Catatan: Sesuai sudut 5 baru HttpClient, .map(res => res.json())bagian dalam semua contoh sekarang tidak berguna, karena hasil JSON sekarang diasumsikan secara default.

Tembolok data dan jika tersedia di-cache, kembalikan ini jika tidak buat permintaan 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;
    }
  }
}

Contoh plunker

Artikel ini https://blog.thoughtram.io/angular/2018/03/05/advanced-caching-with-rxjs.html adalah penjelasan yang bagus tentang cara melakukan cache shareReplay.

Per saran Kristen, ini adalah salah satu cara yang bekerja dengan baik untuk HTTP yang dapat diamati, yang hanya memancarkan satu kali dan kemudian mereka menyelesaikan:

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

UPDATE: Ben Lesh mengatakan rilis minor berikutnya setelah 5.2.0, Anda akan dapat memanggil shareReplay () untuk benar-benar cache.

SEBELUMNYA.....

Pertama, jangan gunakan share () atau publishReplay (1) .refCount (), mereka adalah sama dan masalah dengannya, adalah bahwa ia hanya berbagi jika koneksi dilakukan saat yang diamati aktif, jika Anda terhubung setelah selesai. , itu menciptakan lagi diamati, terjemahan, tidak benar-benar caching.

Birowski memberikan solusi tepat di atas, yaitu menggunakan ReplaySubject. ReplaySubject akan mem-cache nilai-nilai yang Anda berikan (bufferSize) dalam kasus kami 1. Ini tidak akan membuat share yang dapat diamati seperti baru () setelah refCount mencapai nol dan Anda membuat koneksi baru, yang merupakan perilaku yang tepat untuk caching.

Inilah fungsi yang dapat digunakan kembali

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

Inilah cara menggunakannya

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

Di bawah ini adalah versi lebih lanjut dari fungsi cacheable. Yang satu ini memungkinkan memiliki tabel pencarian sendiri + kemampuan untuk menyediakan tabel pencarian kustom. Dengan cara ini, Anda tidak perlu memeriksa this._cache seperti pada contoh di atas. Perhatikan juga bahwa alih-alih meneruskan observable sebagai argumen pertama, Anda melewatkan fungsi yang mengembalikan observable, ini karena Http Angular dieksekusi segera, jadi dengan mengembalikan fungsi malas yang dieksekusi, kita dapat memutuskan untuk tidak memanggilnya jika sudah ada di cache kami.

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

Pemakaian:

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

rxjs 5.4.0 memiliki metode shareReplay baru .

• rx-book shareReplay ()
• Tidak ada dokumen di reactivex.io/rxjs

Penulis secara eksplisit mengatakan "ideal untuk menangani hal-hal seperti caching hasil AJAX"

rxjs PR # 2443 feat (shareReplay): menambahkan shareReplayvarianpublishReplay

shareReplay mengembalikan yang dapat diobservasi yang merupakan sumber multicasted di atas ReplaySubject. Subjek ulangan yang didaur ulang karena kesalahan dari sumber, tetapi tidak pada penyelesaian sumber. Ini membuat ShareReplay ideal untuk menangani hal-hal seperti caching hasil AJAX, karena dapat dicoba. Itu perilaku yang berulang, namun, berbeda dari bagian dalam hal itu tidak akan mengulangi sumber yang dapat diamati, melainkan akan mengulangi nilai-nilai sumber yang dapat diamati.

menurut artikel ini

Ternyata kita dapat dengan mudah menambahkan caching ke observable dengan menambahkan publishReplay (1) dan refCount.

jadi di dalam jika pernyataan hanya menambahkan

.publishReplay(1)
.refCount();

untuk .map(...)

rxjs versi 5.4.0 (2017-05-09) menambahkan dukungan untuk shareReplay .

Mengapa menggunakan shareReplay?

Anda umumnya ingin menggunakan shareReplay ketika Anda memiliki efek samping atau perhitungan pajak yang Anda tidak ingin dieksekusi di antara banyak pelanggan. Mungkin juga bermanfaat dalam situasi di mana Anda tahu Anda akan memiliki pelanggan yang terlambat ke aliran yang membutuhkan akses ke nilai yang dipancarkan sebelumnya. Kemampuan untuk memutar ulang nilai-nilai pada berlangganan adalah apa yang membedakan share dan shareReplay.

Anda dapat dengan mudah memodifikasi layanan sudut untuk menggunakan ini dan mengembalikan yang dapat diamati dengan hasil yang di-cache yang hanya akan membuat panggilan http sekali saja (dengan asumsi panggilan pertama berhasil).

Contoh Layanan Sudut

Berikut adalah layanan pelanggan yang sangat sederhana yang digunakan 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 */
}

Perhatikan bahwa penugasan dalam konstruktor dapat dipindahkan ke metode getCustomerstetapi karena dapat diamati kembali dari HttpClientyang "dingin" melakukan hal ini dalam konstruktor dapat diterima karena panggilan http hanya akan dibuat dengan setiap panggilan pertama ke subscribe.

Juga asumsi di sini adalah bahwa data yang dikembalikan awal tidak basi dalam masa hidup contoh aplikasi.

Saya membintangi pertanyaan, tetapi saya akan mencoba dan mencoba ini.

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

Inilah buktinya :)

Hanya ada satu takeaway: getCustomer().subscribe(customer$)

Kami tidak berlangganan respons api dari getCustomer(), kami berlangganan ReplaySubject yang dapat diobservasi yang juga dapat berlangganan ke Observable berbeda dan (dan ini penting) menyimpan nilai yang dipancarkan terakhir dan mempublikasikannya ke salah satu dari itu (ReplaySubject's ) pelanggan.

Saya menemukan cara untuk menyimpan hasil http ke sessionStorage dan menggunakannya untuk sesi, sehingga tidak akan pernah memanggil server lagi.

Saya menggunakannya untuk memanggil API github untuk menghindari batas penggunaan.

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

FYI, batas sessionStorage adalah 5M (atau 4,75M). Jadi, tidak boleh digunakan seperti ini untuk set data yang besar.

------ sunting -------------
Jika Anda ingin memiliki data yang disegarkan dengan F5, yang menggunakan data memori bukan sesiStorage;

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

Implementasi yang Anda pilih akan bergantung pada apakah Anda ingin berhenti berlangganan () untuk membatalkan permintaan HTTP Anda atau tidak.

Bagaimanapun, dekorator TypeScript adalah cara yang bagus untuk perilaku standardisasi. Ini yang saya tulis:

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

Definisi dekorator:

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

Data Respons HTTP Cacheable menggunakan Rxjs Observer / Observable + Caching + Subscription

Lihat Kode Di Bawah Ini

* Penafian: Saya baru mengenal rxjs, jadi ingatlah bahwa saya mungkin menyalahgunakan pendekatan observable / observer. Solusi saya murni konglomerasi dari solusi lain yang saya temukan, dan merupakan konsekuensi dari kegagalan untuk menemukan solusi sederhana yang terdokumentasi dengan baik. Jadi saya memberikan solusi kode lengkap saya (seperti yang ingin saya temukan) dengan harapan dapat membantu orang lain.

* perhatikan, pendekatan ini secara longgar didasarkan pada GoogleFirebaseObservables. Sayangnya saya kurang pengalaman / waktu yang tepat untuk meniru apa yang mereka lakukan di bawah tenda. Tetapi berikut ini adalah cara sederhana untuk menyediakan akses tidak sinkron ke beberapa data yang bisa di-cache.

Situasi : Komponen 'daftar produk' bertugas menampilkan daftar produk. Situs ini adalah aplikasi web satu halaman dengan beberapa tombol menu yang akan 'memfilter' produk yang ditampilkan pada halaman.

Solusi : Komponen "berlangganan" ke metode layanan. Metode layanan mengembalikan array objek produk, yang diakses komponen melalui panggilan balik berlangganan. Metode layanan membungkus aktivitasnya dalam Pengamat yang baru dibuat dan mengembalikan pengamat. Di dalam pengamat ini, ia mencari data yang di-cache dan meneruskannya kembali ke pelanggan (komponen) dan kembali. Kalau tidak, ia mengeluarkan panggilan http untuk mengambil data, berlangganan respons, di mana Anda dapat memproses data itu (misalnya memetakan data ke model Anda sendiri) dan kemudian meneruskan data kembali ke pelanggan.

Kode

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 (model)

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

Ini adalah contoh dari output yang saya lihat ketika saya memuat halaman di Chrome. Perhatikan bahwa pada muatan awal, produk diambil dari http (panggilan ke layanan simpul saya, yang berjalan secara lokal pada port 3000). Ketika saya klik untuk menavigasi ke tampilan 'disaring' dari produk, produk ditemukan dalam cache.

Log Chrome (konsol) saya:

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

... [mengklik tombol menu untuk memfilter produk] ...

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

Kesimpulan: Ini adalah cara paling sederhana yang saya temukan (sejauh ini) untuk mengimplementasikan data respons http yang dapat di-cache. Di aplikasi sudut saya, setiap kali saya menavigasi ke tampilan produk yang berbeda, komponen daftar produk memuat ulang. ProductService tampaknya merupakan contoh bersama, jadi cache lokal 'produk: Produk []' di ProductService dipertahankan selama navigasi, dan panggilan berikutnya ke "GetProducts ()" mengembalikan nilai yang di-cache. Satu catatan terakhir, saya sudah membaca komentar tentang bagaimana observable / langganan perlu ditutup ketika Anda selesai untuk mencegah 'kebocoran memori'. Saya belum memasukkan ini di sini, tapi itu sesuatu yang perlu diingat.

Saya berasumsi bahwa @ ngx-cache / core dapat berguna untuk memelihara fitur caching untuk panggilan http, terutama jika panggilan HTTP dilakukan pada platform browser dan server .

Katakanlah kita memiliki metode berikut:

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

Anda dapat menggunakan Cacheddekorator @ ngx-cache / core untuk menyimpan nilai yang dikembalikan dari metode membuat panggilan HTTP di cache storage( yang storagedapat dikonfigurasi, silakan periksa implementasinya di ng-seed / universal ) - tepat pada eksekusi pertama. Kali berikutnya metode dipanggil (tidak masalah di browser atau platform server ), nilainya diambil dari 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());
}

Ada juga kemungkinan untuk metode penggunaan caching ( has, get, set) menggunakan caching 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';
  }
}

Berikut adalah daftar paket, baik untuk caching sisi klien dan sisi server:

• @ ngx-cache / core : utilitas cache
• @ ngx-cache / platform-browser : SPA / implementasi platform Browser
• @ ngx-cache / platform-server : implementasi platform server
• @ ngx-cache / fs-storage : utilitas penyimpanan (diperlukan untuk platform server)

rxjs 5.3.0

Saya belum senang dengan itu .map(myFunction).publishReplay(1).refCount()

Dengan banyak pelanggan, .map()dijalankan myFunctiondua kali dalam beberapa kasus (saya berharap ini hanya dijalankan sekali). Tampaknya satu perbaikanpublishReplay(1).refCount().take(1)

Hal lain yang dapat Anda lakukan adalah tidak menggunakan refCount()dan langsung membuat Observable hot:

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

Ini akan memulai permintaan HTTP terlepas dari pelanggan. Saya tidak yakin apakah berhenti berlangganan sebelum HTTP GET selesai akan membatalkannya atau tidak.

Apa yang ingin kami lakukan adalah memastikan bahwa ini tidak menyebabkan beberapa permintaan jaringan.

Favorit pribadi saya adalah menggunakan asyncmetode untuk panggilan yang membuat permintaan jaringan. Metode itu sendiri tidak mengembalikan nilai, sebaliknya mereka memperbarui BehaviorSubjectdalam layanan yang sama, komponen mana yang akan berlangganan.

Sekarang Mengapa menggunakan BehaviorSubjectbukan Observable? Karena,

• Setelah berlangganan, BehaviorSubject mengembalikan nilai terakhir sedangkan A yang dapat diamati hanya memicu ketika menerima onnext.
• Jika Anda ingin mengambil nilai terakhir dari BehaviorSubject dalam kode yang tidak dapat diamati (tanpa berlangganan), Anda dapat menggunakan getValue()metode ini.

Contoh:

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

Lalu, di mana pun diperlukan, kita bisa berlangganan saja customers$.

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

Atau mungkin Anda ingin menggunakannya secara langsung dalam templat

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

Jadi sekarang, sampai Anda membuat panggilan lagi ke getCustomers, data disimpan di customers$BehaviorSubject.

Jadi bagaimana jika Anda ingin menyegarkan data ini? lakukan panggilan kegetCustomers()

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

Menggunakan metode ini, kami tidak perlu secara eksplisit menyimpan data antara panggilan jaringan berikutnya karena ditangani oleh Internet BehaviorSubject.

PS: Biasanya ketika komponen dihancurkan itu adalah praktik yang baik untuk menyingkirkan langganan, untuk itu Anda dapat menggunakan metode yang disarankan dalam jawaban ini .

Jawaban yang bagus

Atau Anda bisa melakukan ini:

Ini dari versi terbaru rxjs. Saya menggunakan versi 5.5.7 dari RxJS

import {share} from "rxjs/operators";

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

Panggil saja share () setelah peta dan sebelum berlangganan .

Dalam kasus saya, saya memiliki layanan generik (RestClientService.ts) yang melakukan panggilan sisanya, mengekstraksi data, memeriksa kesalahan dan mengembalikan yang dapat diamati ke layanan implementasi konkret (misalnya: ContractClientService.ts), akhirnya implementasi konkret ini mengembalikan diamati ke de ContractComponent.ts, dan yang satu ini berlangganan untuk memperbarui tampilan.

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

}

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

}

Semuanya statis karena bagaimana kita menggunakannya, tetapi merasa bebas untuk menjadikannya kelas normal dan layanan. Saya tidak yakin apakah angular menyimpan satu instance untuk sepanjang waktu (baru di Angular2).

Dan inilah cara saya menggunakannya:

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

Saya berasumsi mungkin ada cara yang lebih pintar, yang akan menggunakan beberapa Observabletrik tapi ini baik-baik saja untuk tujuan saya.

Cukup gunakan lapisan cache ini, ia melakukan semua yang Anda butuhkan, dan bahkan mengelola cache untuk permintaan ajax.

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

Ini sangat mudah digunakan

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

Lapisan (sebagai layanan sudut injeksi) adalah

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

Itu .publishReplay(1).refCount();atau .publishLast().refCount();karena Angular Http dapat diamati selesai setelah permintaan.

Kelas sederhana ini cache hasilnya sehingga Anda dapat berlangganan .nilai berkali-kali dan hanya membuat 1 permintaan. Anda juga dapat menggunakan .reload () untuk membuat permintaan baru dan menerbitkan data.

Anda bisa menggunakannya seperti:

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

dan sumbernya:

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

}

Anda dapat membangun kelas sederhana Cacheable <> yang membantu mengelola data yang diambil dari server http dengan beberapa pelanggan:

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

}

Pemakaian

Deklarasikan objek <> Cacheable (mungkin sebagai bagian dari layanan):

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

dan pawang:

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

Panggilan dari komponen:

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

Anda dapat memiliki beberapa komponen untuk berlangganan.

Lebih detail dan contoh kode ada di sini: http://devinstance.net/articles/20171021/rxjs-cacheable

Anda cukup menggunakan ngx-cacheable ! Ini lebih sesuai dengan skenario Anda.

Manfaat menggunakan ini

• Itu panggilan sisanya API hanya sekali, cache respons & mengembalikan sama untuk permintaan berikut.
• Dapat memanggil API sesuai kebutuhan setelah membuat / memperbarui / menghapus operasi.

Jadi, kelas layanan Anda akan menjadi seperti ini -

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

Inilah tautan untuk referensi lebih lanjut.

Sudahkah Anda mencoba menjalankan kode yang sudah Anda miliki?

Karena Anda membuat Observable dari janji yang dihasilkan dari getJSON(), permintaan jaringan dibuat sebelum siapa pun berlangganan. Dan janji yang dihasilkan dibagikan oleh semua pelanggan.

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