ladylilia.com/apps/twitch-logs/src/app/services/search.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 { SearchResult } from '../models/searchresult';


export class SearchDataSource implements DataSource<SearchResult> {
    private resultsSubject = new BehaviorSubject<SearchResult[]>([]);
    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) { }

    loadComments(commenter: string,
                 term: string,
                 sortColumn: string,
                 sortDirection: string,
                 pageIndex: number,
                 pageSize: number) {

        this.loadingSubject.next(true);

        this.commentsService.searchComments(commenter, term, sortColumn, sortDirection,
                                            pageIndex, pageSize)
            .pipe(
                catchError(() => of([])),
                finalize(() => this.loadingSubject.next(false))
            )
            .subscribe((data: any) => {
                this.resultsSubject.next(data.results);
                this.countSubject.next(data.totalCount);
            });
    }

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

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