What is the purpose of rxjs observable if it does not react to changes?

Question

I have an array of items and I wrap that in an observable using of.
The observable is created before the array is populated.
When the array is finally populated, the observable callback passed to subscribe does not get called.

From what I understand, observable only calls this callback for the items already in the list, which in my opinion makes it redundant.

I have a case where I use this observable inside an *ngFor with async pipe, and this one reacts correctly, but when I pas the observable as data source to a mat-table or I pass my callback to the subscribe function, then I don't get anything when the list is eventually populated.

What's the thing that async pipe does behind the scenes, and I am missing?

export class DiscoveryService {
  private deviceList: DeviceModel[] = [];

  constructor() {
  }

  getDeviceList(): void {
    // Get devices from server and push them in the deviceList
  }

  observeDeviceList(): Observable<DeviceModel[]> {
    return of(this.deviceList);
  }
}

export class DeviceListComponent implements OnInit {
  deviceList$: Observable<DeviceModel[]>;

  constructor(private discoveryService: DiscoveryService) { }

  ngOnInit() {
    this.deviceList$ = this.discoveryService.observeDeviceList();

    // This callback get's called only once at the beginning, with an empty list
    this.deviceList$.subscribe(devices => console.log('got devices: ' , devices));

    // When the devices are retrieved from the server, the callback 
    //from the above subscription is not triggered again
    this.discoveryService.getDeviceListx();
  }
}

The async pipe gets updated correctly, but I guess this might be because the ngOnInit is called before the *ngFor runs. I'm not sure.

<mat-nav-list *ngFor="let device of deviceList$ | async">

It's kind of unclear what you want to do or where's the part that doesn't work. — martin, Nov 18 '19 at 14:26
The part that does not work is that when the list get's populated, my callback is not called. `got devices: ` is printed only once at the beginning, with an empty array. — codentary, Nov 18 '19 at 14:50
@MaximilianBoth, if I initialize it before, say in the constructor, it does not make a difference. The list is empty at that point also and the observable does not react to the future changes on that list. It's like the observable is a copy at the list at the point of subscription. — codentary, Nov 18 '19 at 14:51

score 3 · Answer 1 · answered Nov 18 '19 at 14:32

3

Your observable doesn't react to changes because its created from a static array using of which only emits once. Here is something you could do instead.

DiscoveryService

export class DiscoveryService {
  private _deviceList$ = new BehaviorSubject<DeviceModel[]>([]);

  construct() {
    this.fetchDeviceList();
  }

  get deviceList$() {
    return this._deviceList$.asObservable();
  }

  fetchDeviceList() {
     this.http.get<DeviceModel[]>('yourUrl').pipe(
       tap((list: DeviceModel[]) => this._deviceList$.next(list))
     ).subscribe();
  }
}

DeviceListComponent

export class DeviceListComponent implements OnInit {
  private _deviceList$: Observable<DeviceModel[]>;

  constructor(private discoveryService: DiscoveryService) { }

  ngOnInit() {
    this._deviceList$ = this.discoveryService.deviceList$;
  }
}

Then this should work just fine in your template

<mat-nav-list *ngFor="let device of _deviceList$ | async">

answered Nov 18 '19 at 14:32

Reqven

1,688
1
8
13

Thanks for the answer. I came across `BehaviorSubject` while researching this, but I got the impression that is has more functionality that I actually need. What is the purpose of `observable` then? Can I create it in some way other that with `of` in order for it to monitor my array? If not, when would you use the observable instead of accessing the array directly? This is what I'm trying to understand. – codentary Nov 18 '19 at 14:48
1

What is the purpose of a BehaviorSubject, when you retrieve the value from an Observable anyway? You basically map the Observable to a BehaviorSubject just to map it again to an Observable – MoxxiManagarm Nov 18 '19 at 14:54
1

BehaviorSubject makes it easy to update data over time. It is an observer in addition to being an observable. If your `deviceList` is likely to change over time (device going on/off, being added/deleted/edited etc..), you might want to use it to update its value. This way, any component subscribing to it will get notified when a new value gets emitted over time, meaning that your UI will be automatically refreshed based on that new value, unlike accessing the array directly as it will correctly render the page when it gets initialized, but won't be updated if any change is made to the array. – Reqven Nov 18 '19 at 16:00
So then... why would you ever use "plain" `observable` if it behaves just like a `copy` and does not react to any add/delete/edit that might happen after the point at which you `subscribe` to it? – codentary Nov 18 '19 at 16:10
A plain observable as you call it does react to changes, it's just not noticeable in your case because you created it using `of`, which means it only emits a single value being the one you pass over, and then completes. That's why I use BehaviorSubject, to update the data over time, whenever I need to. Feel free to ask if you have any further questions. – Reqven Nov 18 '19 at 16:26
Ok, so how else can I create the observable without `of` and without `subject.asObservable()` + `next`. Just an `observable` that `observes` changes to my list and calls the `callback` passed to `subscribe`. – codentary Nov 18 '19 at 16:37
I guess what I'm looking for is a `next` method for `observable`. Some way to tell the `observer` that there is new data. I found an answer to a similar question that uses this: `myObservable.pipe(map(items => {items.push(item); return items;}))`, but this does not seam to work for me wither. [details here](https://stackoverflow.com/questions/53260269/angular-6-add-items-into-observable) – codentary Nov 18 '19 at 17:25
This code clones the initial `Observable` into a new one, alter the array by adding a new item to it and then emits the final value. It doesn't work because you need to subscribe to this new `Observable`, which will still emit only once. Anyway, it doesn't make much sense because you're creating a copy of the initial `Observable`, and the original one is never getting updated. In your case, you need to use a `Subject` and its `next` method to update its value over time. To see an example in action, check this https://stackblitz.com/edit/angular-hqzdcz – Reqven Nov 19 '19 at 14:46

score 0 · Answer 2 · answered Nov 18 '19 at 15:02

0

export class DiscoveryService {
  construct(private http: HttpClient) {
  }

  getDeviceList(): Observable<DeviceModel[]> {
     return this.http.get<DeviceModel[]>('yourUrl');
  }
}

Or in case you want to cache it:

export class DiscoveryService {
  private deviceList$: Observable<DeviceList[]>;

  construct(private http: HttpClient) {
  }

  getDeviceList(): Observable<DeviceModel[]> {
     if (!this.deviceList$) {
       this.deviceList$ = this.http.get<DeviceModel[]>('yourUrl').pipe(
         shareReplay(1),
       );
     }

    return this.deviceList$;
  }
}

answered Nov 18 '19 at 15:02

MoxxiManagarm

8,735
3
14
43

I am using gRPC transport and I get a stream to which I `subscribe` like: `stream.on('data', (data) =>{console.log('on stream data: ', data})` . Not sure how to tie this to an observable, other that just inserting in the list that is observed, every time there is something in the stream. – codentary Nov 18 '19 at 15:51
1

ok. check out the GreeterService here: https://www.trion.de/news/2018/12/19/angular-grpc-web.html – MoxxiManagarm Nov 18 '19 at 19:56
Yes, this is how I do it, by setting a callback from which I call next in the case of the behavior subject. Thanks – codentary Nov 18 '19 at 21:05

What is the purpose of rxjs observable if it does not react to changes?

2 Answers2