How to use a declarative pattern on Observables that watch the network?

I have been trying to learn about the declarative pattern/approach in rxjs, which from my understanding it to not use .subscribe() directly in the typescript itself and use the async pipe instead.

I have a class that holds data about an image, and some of that data is a tags property which is an array of strings. Through the UI you can add/remove tags to this property. When you save a new item the value goes to the server via a websocket, and the socket will send a list of all the items tags back (including the newly added one).

So, I created a listener that listens for when the websocket responds with data related to the image tags. When it sees this message it should update all the tags in the object for this image.

Something like this pseudo code below:

interface FileInfo {
  tags: string[];
}

@Component({
  template: `
    <div *ngIf="image$ | async as image">
      <div *ngFor="let tag of image.tags">{{tag}}</div>
    </div>
  `
})
export class MyComponent {
  image = new BehaviorSubject<FileInfo | null>(null);
  image$ = this.image.asObservable();

  ngOnInit() {
    this.websocket.on('tags:image').pipe(
      map(i => i.data) // i.data is an array of strings
      tap(strs => this.image.value = strs)
    );
  }
}

Now when a message comes back it should update the image object and re-render the component.
Without using a subscribe on the this.websocket.on() how can this pattern/approach be possible? All this does is watch the network and modify a property on an object.

I modified the above to to this pattern, but I don’t really like that I have to manage more than one property now when doing .next(), this approach doesn’t seem right and seems like it is easy to get image.tags out of sync the value of tags$.

@Component({
  template: `
    <div *ngIf="image$ | async as image">
      <div *ngFor="let tag of tags$ | async">{{tag}}</div>
    </div>
  `
})
export class MyComponent {
  image = new BehaviorSubject<FileInfo | null>(null);
  image$ = this.image.asObservable();

  tags = new BehaviorSubject<string[]>([]);
  tags$ = merge(this.tags, this.websocket.on('tags:image')).pipe(
    map(i => i.data) // i.data is an array of strings
  );

  ngOnInit() {
   // I want to get rid of this subscribe too
   // However this is the next
   this.manageService.$collectionEvents.subscribe(event => {
     this.image.next(event.image);
     this.tags.next(event.image.tags);
    });
  }
}

>Solution :

It would be like this (I’ve just made an observable data$ to simulate your web socket response):

  image$ = new Observable<FileInfo>();

  ngOnInit() {
    const data$ = of({ data: ['string1', 'string2', 'string3'] });
    this.image$ = data$.pipe(
      map((i) => {
        return { tags: i.data };
      })
    );
  }

In your context

  image$ = new Observable<FileInfo>();

  ngOnInit() {
    this.image$ = this.websocket.on('tags:image').pipe(
      map((i) => {
        return { tags: i.data };
      })
    );
  }

The async pipe will automatically subscribe to image$ and also unsubscribe when the component is destroyed. It also handles updating html whenever the observable receives a new value.

Here’s a stackblitz where I simulate the observable value changing every second: https://stackblitz.com/edit/angular-ivy-m2digv?file=src/app/app.component.ts