RxJS observable of array not working as expected

Question

I have an angular service that has an array of custom objects and an observable of this array. An outside component subscribes to the observable, but does not react when the list get's changed.
Also, the observable is set as data source for a mat-table which does not react eather.

The service (uses a gRPC stream to get data):

export class ScanService {
    private scanCient: ScanClient = new ScanClient('http://' + window.location.hostname + ':8080', null, null);
    private results: ItemModel[] = [];
    readonly results$ = from(this.results);

    constructor() { }

    start(ip: string): boolean {
    let stream = this.scanCient.listen(ip, {});

    stream.on('data', (response) => {
        this.results.push({
            name: response.getName(),
            type: response.getType(),
        });

        console.log('stream got item: ' + response.getName());
        }
    }
}

The consumer:

export class ScanComponent implements OnInit {
    //...
    results$: Observable<ItemModel[]>;

    constructor(private scanService: ScanService) { }

    ngOnInit() {
      this.results$ = this.scanService.results$;
      this.results$.subscribe({next: function(results){console.log('subscriber got item: ', results);}});
    }

    onScan() {
      this.scanService.start(this.ip);
    }
  }

<table mat-table [dataSource]="results$" class="mat-elevation-z8">

In the console output I see the stream got item for each item, but not the subscriber got item and also the mat-table remains empty.

EDIT:

I have a similar example using of that appears to work as expected.
The difference is that the gRPC call return an object directly, not a stream.

Service:

export class DiscoveryService {
    private deviceList: DeviceModel[] = [];
    private discoveryCient: DiscoveryClient = new DiscoveryClient('http://' + window.location.hostname + ':8080', null, null);

    constructor() {
    }

    getDeviceList(): void {
       this.discoveryCient.devices(new EmptyProto(), {}, (err, reply) => {
        const pbDeviceList = reply.getDevicesList();
        for (const pbDevice of pbDeviceList) {
          this.deviceList.push({ ip: pbDevice.getIp()});
        }
      });
    }

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

Consumer:

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

  constructor(private discoveryService: DiscoveryService) { }

  ngOnInit() {
    // First get the observable to the list(which is still empty)
    this.deviceList$ = this.discoveryService.observeDeviceList();

    // Then tell the service to populate the list
    this.discoveryService.getDeviceList();
  }
}

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

You're creating an Observable using from, using an empty array. So that creates an empty observable, which never emits anything. As the from documentation explains: https://rxjs-dev.firebaseapp.com/api/index/function/from. If you want an Observable which emits every time you push to an assay, then you need a Subject, and you need to call next() on that Subject every time you change the array. — JB Nizet, Nov 18 '19 at 12:48
if instead of `from`, I use `of`, is there any difference? I am sacking because I have a case like this that behaves as expected, I call `of` to wrap the array when the it's still empty, and when it get's populated, the subscriber get's the data. — codentary, Nov 18 '19 at 12:59
There is a different. That would create an observable that amits an empty array, once, and then completes. Still not what you want. — JB Nizet, Nov 18 '19 at 13:00
How come `async` pipe that binds inside an `*ngFor` works with an empty wrapped array. — codentary, Nov 18 '19 at 13:01
I edited the question to show also the case that appears to be working as expected. — codentary, Nov 18 '19 at 13:14
Because you push to that array, and thus modify its state. So Angular detects the change. That doesn't mean that the observable emits a new value. — JB Nizet, Nov 18 '19 at 13:51

Chund · Accepted Answer · 2019-11-18T13:49:26.740

0

To accumulate what the other helpful voices already told you in the comments:

You are facing two issues here.

of() vs from()

I refer to another question here. Basically of([1,2,3]) creates an observable that will emit "[1,2,3]" as on array, so you will get one emission, while from([1,2,3]) will emit "1" then "2" and "3" at last. And from the usage of your result$ I would guess you want to to use of().

Subject vs Observable

You want updates to you result-Array to be published to the subscribers of the result$-Observable, but your Observable completes, once it has emitted all its values, therefore it will complete and automatically unsubscribe its subscribers once it emitted the static data present at the moment of subscription. A simple fix would be to use Subjects especially BehaviourSubjects instead of Observables. These have the property that they behave like a publisher of sorts.

const result$ = new BehaviourSubject<ItemModel[]>(this.result);

These emit the last emited value to any new subscribers and additionally allow you to release new data to their subscribers by calling their next()-method. So you would need to call

result$.next(newValue);

instead of

result = newValue;

edited Nov 18 '19 at 13:49

answered Nov 18 '19 at 13:27

Chund

355
2
15

The weird thing is that I have an example of a observed list that get's populated after the observable is created with `of` and the subscriber get the data without doing anything extra. I will clarify this example in the question. – codentary Nov 18 '19 at 13:31
Please check now the `EDIT` part of the question which is a working example (modified now to be a little more clear). – codentary Nov 18 '19 at 13:36
So how do I do that in my code? Because otherwise... from what you said, I understood that an `observable` would not react to changes of the `observed` array and will only report what it already had when the `observable` was created with `of`, lets say. If that is indeed the case, what is the purpose of `observable`? It's just a static array, it adds no extra functionality on top of it. Why is it called `observable` and why would I wanna get an observable to an array instead of the actual array? – codentary Nov 18 '19 at 13:44
since you would open a giant can of worms with this I would recommend either reading up on that yourself or asking this in a seperate question as I cannot claim to be all knowing in that matter :) – Chund Nov 18 '19 at 13:47
Seems like i missunderstood your question there. The fact that it completes is da feature of the of() function, normally observable don't just complete so you would have a stream of data once subscribed – Chund Nov 26 '19 at 08:37

RxJS observable of array not working as expected

1 Answers1