How to synchronize multiple angular observable data services

Question

Suppose I structure my Angular app using observable data services, backed by serverside endpoints:

@Injectable()
export class TodoService {
  todos: Observable<Todo[]>
  private _todos: BehaviorSubject<Todo[]>;

  constructor(private http: Http) {
    this._todos = <BehaviorSubject<Todo[]>>new BehaviorSubject([]);
    this.todos = this._todos.asObservable();
  }

  loadData() {
    this.http.get(dataUrl).subscribe(data => this._todos.next(data));
  }
}

Now suppose my data has reference to some other model and exposed by some other service (some form of Many-to-Many relationship):

interface ToDo {
  title: string;
  status: StatusEnum;
  requiredResouces: Resource[];
}
interface Resource {
  name: string;
  todos: ToDo[];
}

@Injectable()
export class ResourcesService {
  resources: Observable<Resource[]>
  private _resources: BehaviorSubject<Resource[]>;

  ...
}

Now, suppose I add a method to either service that "links" a todo and a resource, that service will be able to push an updated state down the subject, but the other service will be unaware of the change. For example:

export class TodoService {

  ...

  addResourceRequirement(todo: ToDo, resource: Resource) {
    this.http.post(`${url}/${todo.id}/`, {addResource: resource.id})
      .subscribe(() => this.loadData());
  }
}

Would cause any "todos" observer to refresh, but any "resources" observer would still show the old state...

What design / pattern / architecture would you use to synchronize both services?

---

(I know there are architectures that avoid this difficulty as a whole - particularly flux based solutions - NgRX etc... but I'm interested in a solution specifically for the observable data services pattern)

Answer 1

I am not sure I understand the question.

But lets suppose you have both subjects with lists of resources and todos.

import {combineLatest} from "rxjs/observable/combineLatest";

mergeResourcesAndTodos() {
    return combineLatest(
        this.todoService.todos
        this.resourceService.resources
      ).map((data: any[]) => {
       const todos=data[0]
       const resources=data[1]
       //and here you can map anyway you want like a list of todos that have a list of resources or the other way around, here I will do an example with a list of todos with resources.
         return todos.map(todo=>{
              const resource= resources.find(resource=>resource.id===todo.resourceId)
              todo.resource=resource;
              return todo
          })
     })
  }

Answer 2

Your pattern is almost done, you just need to avoid creating the observable at boot time, also, keep in mind that BehaviorSubject extends Observable, so you can use it as it is, using an implicit getter if you want to get it easy to use.

@Injectable()
export class TodoService {
  private _todos: BehaviorSubject<Todo[]>;

  constructor() {
    this._todos = new BehaviorSubject<Todo[]>([]);
  }

  public get todos(): Observable<Todo[]>{
    return this._todos;
  }
}

Then, when you want to add data, simply emit a new value in your _todos subject, see RxJs: Incrementally push stream of data to BehaviorSubject<[]> for incremental array emission if you want to emit everything each time you add some data.

Edit

If a service has a dependency over todos data, you just have to build a new Observable using map, mergeMap, etc operators in order to build a new Observable with todos as source, allowing you to have a new value emitted if the root data changes.

Example:

@Injectable()
export class ResourcesService{

  private _resources: Observable<Resources[]>;

  constructor(private todoService: TodoService){
    this._resources = this.todoService.todos.map(todos => {
      // I suppose you want to aggreagate all resources, you can do it using reduce or anything, I'll use forEach in this example
      const resources: Resource[] = [];
      todos.forEach(todo => resources.push(...todo.requiredResouces);
    });
  }

  //And then implicit getter for _resources.

}

This way, if your TodoService emits a new array, ResourcesService will emit a new array of todos with status done, without needing any other operation.

Another approach

If your resources are coming from another endpoint (meaning that you need another request to fetch them after you updated your data) you might be better using a reloader pattern:

@Injectable()
export class DataReloaderService{

  public readonly reloader: BehaviorSubject<void> = new BehaviorSubject<void>(null);

  public reload():void{
    this.reloader.next(null);
  }
}

Then, whenever you create a data service, you just have to merge it with the reloader observable:

@Injectable()
export class ResourceService {

  private _resources: Observable<Resource[]>;

  constructor(private reloaderService: DataReloaderService, private http: HttpClient){
    this._resources = this.reloaderService.reloader.mergeMap(() => this.http.get(...));
  }

}

Finally, in your service doing modifications:

export class TodoService {

  private _todos: BehaviorSubject<Todo[]>;

  constructor(private reloaderService: DataReloaderService, private http: HttpClient) {
    this._todos = this.reloaderService.reloader.mergeMap(() => this.http.get(dataUrl));
  }

  public get todos(): Observable<Todo[]>{
    return this._todos;
  }

  addResourceRequirement(todo: ToDo, resource: Resource) {
    this.http.post(`${url}/${todo.id}/`, {addResource: resource.id})
      .subscribe(() => this.reloader.reload());
  }
}

NOTE: You should not subscribe in a service, the intent of Observables is to build a cold chain, where you just subscribe in the display part, this second pattern ensures that all your Observables are linked to the central reloader (you can also create multiple reloaders, one per model family for instance), subscription makes you loose that, resulting in strange ways to edit data only. if everything relies on your api, then your observables should only use this api, calling the reloader whenever you edit something to ensure all linked data are updated as well.