ladylilia.com/apps/twitch-logs/src/app/services/videos.datasource.ts

import {
    CollectionViewer,
    DataSource
} from '@angular/cdk/collections';

import { Observable } from 'rxjs/Observable';
import { BehaviorSubject } from 'rxjs/BehaviorSubject';
import { of } from 'rxjs/observable/of';

import {
    catchError,
    finalize
} from 'rxjs/operators';

import { CommentsService } from './comments.service';

import { Video } from '../models/video';


export class VideosDataSource implements DataSource<Video> {
    private videosSubject = new BehaviorSubject<Video[]>([]);
    private countSubject = new BehaviorSubject<number>(0);
    private loadingSubject = new BehaviorSubject<boolean>(false);

    public count$ = this.countSubject.asObservable();
    public loading$ = this.loadingSubject.asObservable();

    constructor(private commentsService: CommentsService) { }

    loadVideos(filter: string,
               sortColumn: string,
               sortDirection: string,
               pageIndex: number,
               pageSize: number) {

        this.loadingSubject.next(true);

        this.commentsService.findVideos(filter, sortColumn, sortDirection,
                                        pageIndex, pageSize)
            .pipe(
                catchError(() => of([])),
                finalize(() => this.loadingSubject.next(false))
            )
            .subscribe((data: any) => {
                this.videosSubject.next(data.videos);
                this.countSubject.next(data.totalCount);
            });
    }

    connect(collectionViewer: CollectionViewer): Observable<Video[]> {
        console.log('Connecting data source');
        return this.videosSubject.asObservable();
    }

    disconnect(collectionViewer: CollectionViewer): void {
        this.videosSubject.complete();
        this.countSubject.complete();
        this.loadingSubject.complete();
    }
}