# RxJS BehaviorSubject with asynchronous initial value

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

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

subscribe(...args) {
.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);
}
});
}

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

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