1
\$\begingroup\$

I am using RxJS to fetch a data structure from a server and cache it, but I also need to be able to overwrite the value locally (specifically, a user operation can change the server state, and the change takes effect locally without waiting for a round-trip communication)

Eventually I came up with this solution, where I am creating a pseudo-Subject object (AjaxSubject) which wraps a ReplaySubject with custom subscribe behaviour (specifically: loading the data from the server if no data exists locally yet).

I'm not particularly happy with it. Specifically I'm wondering if there is a better way to handle the use case of a BehaviorSubject with an asynchronous initial value (where as usual the request is not triggered until something subscribes).

I think the error handling also isn't great; I wrap data and errors in an object so that subscribers can get successful values after an initial failure (e.g. after a retry or setting it locally with next), but this seems clunky.

Can this be done using the RxJS standard operators? Or at least are there ways I can tidy it up?

import { ReplaySubject, of } from 'rxjs';
import { map, catchError } from 'rxjs/operators';
import { ajax } from 'rxjs/ajax';

function makeHttpError({ status }) {
  return (status >= 400) ? 'error ' + status : 'connection failed';
}

class AjaxSubject {
  constructor(url) {
    this.url = url;
    this.subject = new ReplaySubject(1);
    this.loaded = false;
  }

  next(data) {
    this.subject.next({ data, error: null });
    this.loaded = true;
  }

  subscribe(...args) {
    if (!this.loaded) {
      this.loaded = ajax(this.url)
        .pipe(
          map((data) => ({ data: data.response, error: null })),
          catchError((data) => of({ data: null, error: makeHttpError(data) })),
        )
        .subscribe((state) => {
          if (!state.error || this.loaded !== true) {
            this.subject.next(state);
            this.loaded = !state.error;
          }
        });
    }

    return this.subject.subscribe(...args);
  }
}

Update:

I found an alternative way to implement (almost) the same thing using RxJS builtins, but I don't know how I would (externally) trigger a retry with this one if loading from the server fails. I'm also not happy about the double-buffering (use of a BehaviorSubject and shareReplay).

import { BehaviorSubject, of } from 'rxjs';
import { switchMap, shareReplay, map, catchError } from 'rxjs/operators';
import { ajax } from 'rxjs/ajax';

function makeHttpError({ status }) {
  return (status >= 400) ? 'error ' + status : 'connection failed';
}

function load(url) {
  return ajax(url).pipe(
    map((data) => ({ data: data.response, error: null })),
    catchError((data) => of({ data: null, error: makeHttpError(data) })),
  );
}

function ajaxSubject(url) {
  const subject = new BehaviorSubject(undefined);
  const observable = subject.pipe(
    switchMap((v) => ((v === undefined) ? load(url) : of(v))),
    shareReplay(1),
  );
  return { subject, observable };
}
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.