Shared RxJS subject for bidirectional data binding in Angular 2

4
votes

I have a singleton service for app settings

class Setting {
  get foo() {
    return storage.get('foo');
  }

  set foo(val)
    storage.set('foo', val);
  }
}

that is bound in components' views as setting.foo.

Because storage calls may be costly and because it may be asynchronous, I would prefer to replace getter/setter with RxJS subject that could update and read storage whenever needed.

So I'm refactoring it to

class Setting {

  constructor() {
    this.fooSubject = new ReplaySubject(1);

    fooSubject.subscribe((val) => {
      storage.set('foo', val);
    });

    this.foo$ = this.fooSubject
    .map(() => storage.get('foo'))
    .publishReplay(1).refCount();
}

and using it like setting.foo$ | async and setting.fooSubject.next(newFoo). It looks like costly storage.get calls are cached now.

There are two problems.

The first one is that both fooSubject subject and foo$ observable should be publicly available to make this work, while a subject was chosen because it is supposed to be both an observable and an observer.

Can foo$ be made to be a single Subject property of Setting service, so it could be subscribed with subscribe(...) and updated with next(...)?

The second one is that the code is still synchronous.

How can this case be treated for storage.get and storage.set that return promises?

2
javascriptangulartypescriptrxjs

2 Answers

3
votes

There's not much to suggest in your code when it actually works as you want. I'm not sure I catch your use case exactly but I think you can simplify your code by not using this.foo$ at all.

The functionality of storing the latest value is provided by ReplaySubject and unless you really need to call storage.get('foo') every time after storage.set('foo', val); it's not necessary.

I put your code into a live demo: http://plnkr.co/edit/pxjRQr6Q6Q7LzYb1oode?p=preview

So I think it can be simplified to this.

class Setting {

  constructor() {
    var storage = new Storage();

    this.fooSubject = new ReplaySubject(1);
    this.fooSubject.subscribe((val) => {
      storage.set('foo', val);
    });
  }

  get observable() {
    return this.fooSubject.asObservable();
  };

  store(val) {
    this.fooSubject.next(val);
  }
}

I purposely hide the fact that I'm using a Subject with .asObservable() and by wrapping .next() call with store() method. A typical usage could look like:

let settings = new Setting();

settings.store('Hello');

settings.observable.subscribe(val => console.log(val));
settings.store('Hello 2');
settings.store('Hello 3');

settings.observable.subscribe(val => console.log(val));
settings.store('Hello 4');

Which prints to console:

Hello
Hello 2
Hello 3
Hello 3
Hello 4
Hello 4

Note, that you're not initializing the ReplaySubject with any value. Even if you called setting.fooSubject.next(newFoo) right after creating the ReplaySubject it'll be stored again right after subscribing with storage.set('foo', val);.

The issue about being synchronous. Well, your code is in fact asynchronous but sequential. Since JavaScript is single threaded then if storage.get('foo') does some synchronous time consuming operation then it'll block the execution thread and probably the only option is to move it to a Web Worker.

3
votes

It was really easy to get a subject with side effects by extending AnonymousSubject (a class for Subject.create(...) factory). Resulting subject gets destination and source properties that hold original subject and observable.

class FooSharedSubject extends AnonymousSubject {
    constructor() {
        const subject = new BehaviorSubject('');

        const observable = subject.asObservable()
        .mergeMap((value) => promisedStorage.get('foo'))
        .publishReplay(1)
        .refCount();

        super(subject, observable);
    }

    next(value): void {
        promisedStorage.set('foo', value)).then(
            () => this.destination.next(value),
            () => this.destination.error(value)
        );
    }
}