data-source-generic-reactive-class.ts

import { Subject } from 'rxjs/Subject';
import { takeUntil } from 'rxjs/operators';
import { Observable } from 'rxjs/Observable';
import { DataSource } from '@angular/cdk/collections';

export class ReactiveDataSource<T> extends DataSource<T> {

  private _disconnector: Subject<void> = new Subject();

  private _source: Observable<T[]>;

  constructor(source: Observable<T[]>) {
    super();

    this._source = source;
  }

  connect(): Observable<T[]> {
    return this._source.pipe(takeUntil(this._disconnector));
  }

  disconnect(): void {
    this._disconnector.complete();
  }
}