<commit_before>import {Observable} from "rxjs/Observable";
import {Subject} from "rxjs/Subject";
import {Subscription} from "rxjs/Subscription";

import "rxjs/add/operator/timeout";

import {EdgeDirection} from "../Edge";
import {
    Graph,
    GraphMode,
    GraphService,
    IEdgeStatus,
    Node,
    Sequence,
} from "../Graph";
import {
    IFrame,
    StateService,
} from "../State";

export class PlayService {
    private _graphService: GraphService;
    private _stateService: StateService;

    private _nodesAhead: number;
    private _playing: boolean;
    private _speed: number;

    private _direction$: Observable<EdgeDirection>;
    private _directionSubject$: Subject<EdgeDirection>;
    private _playing$: Observable<boolean>;
    private _playingSubject$: Subject<boolean>;
    private _speed$: Observable<number>;
    private _speedSubject$: Subject<number>;

    private _playingSubscription: Subscription;
    private _cacheSubscription: Subscription;
    private _clearSubscription: Subscription;
    private _graphModeSubscription: Subscription;

    constructor(graphService: GraphService, stateService: StateService) {
        this._graphService = graphService;
        this._stateService = stateService;

        this._directionSubject$ = new Subject<EdgeDirection>();
        this._direction$ = this._directionSubject$
            .startWith(EdgeDirection.Next)
            .publishReplay(1)
            .refCount();

        this._direction$.subscribe();

        this._playing = false;
        this._playingSubject$ = new Subject<boolean>();
        this._playing$ = this._playingSubject$
            .startWith(this._playing)
            .publishReplay(1)
            .refCount();

        this._speed = 0.5;
        this._speedSubject$ = new Subject<number>();
        this._speed$ = this._speedSubject$
            .startWith(this._speed)
            .publishReplay(1)
            .refCount();

        this._speed$.subscribe();

        this._nodesAhead = this._mapNodesAhead(this._mapSpeed(this._speed));

        this._playing$
            .switchMap(
                (playing: boolean): Observable<[EdgeDirection, IEdgeStatus]> => {
                    return !playing ?
                        Observable.empty() :
                        Observable
                            .combineLatest(
                                this._direction$,
                                this._stateService.currentNode$
                                    .switchMap(
                                        (node: Node): Observable<IEdgeStatus> => {
                                            return node.sequenceEdges$
                                                .first(
                                                    (status: IEdgeStatus): boolean => {
                                                        return status.cached;
                                                    })
                                                .timeout(15000)
                                                .catch(
                                                    (error: Error): Observable<IEdgeStatus> => {
                                                        console.error(error);
                                                        this.stop();

                                                        return Observable.empty();
                                                    });
                                        }));
                })
            .map(
                ([direction, edgeStatus]: [EdgeDirection, IEdgeStatus]): boolean => {
                    for (let edge of edgeStatus.edges) {
                        if (edge.data.direction === direction) {
                            return true;
                        }
                    }

                    return false;
                })<commit_msg>fix: handle different directions in stop subscription<commit_after>import {Observable} from "rxjs/Observable";
import {Subject} from "rxjs/Subject";
import {Subscription} from "rxjs/Subscription";

import "rxjs/add/operator/timeout";

import {EdgeDirection} from "../Edge";
import {
    Graph,
    GraphMode,
    GraphService,
    IEdgeStatus,
    Node,
    Sequence,
} from "../Graph";
import {
    IFrame,
    StateService,
} from "../State";

export class PlayService {
    private _graphService: GraphService;
    private _stateService: StateService;

    private _nodesAhead: number;
    private _playing: boolean;
    private _speed: number;

    private _direction$: Observable<EdgeDirection>;
    private _directionSubject$: Subject<EdgeDirection>;
    private _playing$: Observable<boolean>;
    private _playingSubject$: Subject<boolean>;
    private _speed$: Observable<number>;
    private _speedSubject$: Subject<number>;

    private _playingSubscription: Subscription;
    private _cacheSubscription: Subscription;
    private _clearSubscription: Subscription;
    private _graphModeSubscription: Subscription;

    constructor(graphService: GraphService, stateService: StateService) {
        this._graphService = graphService;
        this._stateService = stateService;

        this._directionSubject$ = new Subject<EdgeDirection>();
        this._direction$ = this._directionSubject$
            .startWith(EdgeDirection.Next)
            .publishReplay(1)
            .refCount();

        this._direction$.subscribe();

        this._playing = false;
        this._playingSubject$ = new Subject<boolean>();
        this._playing$ = this._playingSubject$
            .startWith(this._playing)
            .publishReplay(1)
            .refCount();

        this._speed = 0.5;
        this._speedSubject$ = new Subject<number>();
        this._speed$ = this._speedSubject$
            .startWith(this._speed)
            .publishReplay(1)
            .refCount();

        this._speed$.subscribe();

        this._nodesAhead = this._mapNodesAhead(this._mapSpeed(this._speed));

        this._playing$
            .switchMap(
                (playing: boolean): Observable<[EdgeDirection, IEdgeStatus]> => {
                    return !playing ?
                        Observable.empty() :
                        Observable
                            .combineLatest(
                                this._stateService.currentNode$,
                                this._direction$)
                            .switchMap(
                                    ([node, direction]: [Node, EdgeDirection]): Observable<[EdgeDirection, IEdgeStatus]> => {
                                        const edgeStatus$: Observable<IEdgeStatus> = (
                                            [EdgeDirection.Next, EdgeDirection.Prev].indexOf(direction) > -1 ?
                                                node.sequenceEdges$ :
                                                node.spatialEdges$)
                                            .first(
                                                (status: IEdgeStatus): boolean => {
                                                    return status.cached;
                                                })
                                            .timeout(15000)
                                            .catch(
                                                (error: Error): Observable<IEdgeStatus> => {
                                                    console.error(error);
                                                    this.stop();

                                                    return Observable.of<IEdgeStatus>({ cached: false, edges: [] });
                                                });

                                        return Observable
                                            .combineLatest(
                                                Observable.of(direction),
                                                edgeStatus$);
                                    });
                })
            .map(
                ([direction, edgeStatus]: [EdgeDirection, IEdgeStatus]): boolean => {
                    for (let edge of edgeStatus.edges) {
                        if (edge.data.direction === direction) {
                            return true;
                        }
                    }

                    return false;
                })
{'fraction_non_alphanumeric': 0.08861973857776712, 'fraction_numerical': 0.00251828756445617, 'mean_word_length': 1.219856268299175, 'pattern_counts': {'":': 0, '<': 29, '<?xml version=': 0, '>': 40, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 65, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2068860', 'n_tokens_mistral': 1754, 'n_tokens_neox': 1668, 'n_words': 367}