<commit_before>import { SignalRConnectionBase } from './signalr.connection.base';
import { Observable } from 'rxjs/Observable';
import { ReplaySubject } from 'rxjs/ReplaySubject';
import { BroadcastEventListener } from '../eventing/broadcast.event.listener';
import { ConnectionStatus } from './connection.status';
import { NgZone } from '@angular/core';

export class SignalRConnection extends SignalRConnectionBase {
    private _status: Observable<ConnectionStatus>;
    private _errors: Observable<any>;
    private _jConnection: any;
    private _jProxy: any;
    private _zone: NgZone;

    constructor(jConnection: any, jProxy: any, zone: NgZone) {
        super();
        this._jProxy = jProxy;
        this._jConnection = jConnection;
        this._zone = zone;
        this._errors = this.wireUpErrorsAsObservable();
        this._status = this.wireUpStatusEventsAsObservable();
    }

    public get errors(): Observable<any> {
        return this._errors;
    }

    public get status(): Observable<ConnectionStatus> {
        return this._status;
    }

    public start(): Promise<any> {
        let $promise = new Promise<any>((resolve, reject) => {
            this._jConnection.start().done(function (...results: any[]) {
                resolve(results);
            })
                .fail(function (err: any) {
                    reject(err);
                });
        });
        return $promise;
    }

    public stop(): void {
        this._jConnection.stop();
    }

    public invoke(method: string, ...parameters: any[]): Promise<any> {
        if (method == null) throw new Error('SignalRConnection: Failed to invoke. Argument \'method\' can not be null');

        console.debug(`SignalRConnection. Start invoking \'${method}\'...`);
        let $promise = new Promise<any>((resolve, reject) => {
            this._jProxy.invoke(method, ...parameters)
                .done(function (result: any) {
                    console.debug(`\'${method}\' invoked succesfully. Resolving promise...`);
                    resolve(result);
                    console.debug(`Promise resolved.`);
                })
                .fail(function (err: any) {
                    console.debug(`Invoking \'${method}\' failed. Rejecting promise...`);
                    reject(err);
                    console.debug(`Promise rejected.`);
                });
        });
        return $promise;
    }

    private wireUpErrorsAsObservable(): Observable<any> {
        let sError = new ReplaySubject<any>();

        this._jConnection.error((error: any) => {
            this._zone.run(() => {
                sError.next(error);
            });
        });
        return sError;
    }

    private wireUpStatusEventsAsObservable(): Observable<ConnectionStatus> {
        let sStatus = new ReplaySubject<ConnectionStatus>();
        let connStatusNames = ['starting', 'received', 'connectionSlow', 'reconnecting', 'reconnected', 'stateChanged', 'disconnected'];
        // aggregate all signalr connection status handlers into 1 observable. 
        connStatusNames.forEach((statusName) => {
            // handler wire up, for signalr connection status callback.
            this._jConnection[statusName]((...args: any[]) => {
                this._zone.run(() => {
                    sStatus.next(new ConnectionStatus(statusName));
                });
            });
        });
        return sStatus;
    }

    public listen<T>(listener: BroadcastEventListener<T>): void {
        if (listener == null) throw new Error('Failed to listen. Argument \'listener\' can not be null');

        console.log(`SignalRConnection: Starting to listen to server event with name ${listener.event}`);

        this._jProxy.on(listener.event, (...args: any[]) => {
            console.log('SignalRConnection.proxy.on invoked. Calling listener next() ...');
            let casted: T = null;
            if (args.length > 0) casted = <T>args[0];
            
            this._zone.run(() => {
                listener.next(casted);
            });
            console.log('listener next() called.');
        });
    }

    private onBroadcastEventReceived<T>(listener: BroadcastEventListener<T>, ...args: any[]) {
        console.log('SignalRConnection.proxy.on invoked. Calling listener next() ...');

        let casted: T = null;
        if (args.length > 0) casted = <T>args[0];

        this._zone.run(() => {
            listener.next(casted);
        });

        console.log('listener next() called.');
    }

}<commit_msg>fix: connection.status should not replay events prior to subscription<commit_after>import { SignalRConnectionBase } from './signalr.connection.base';
import { Observable } from 'rxjs/Observable';
import { BroadcastEventListener } from '../eventing/broadcast.event.listener';
import { ConnectionStatus } from './connection.status';
import { NgZone } from '@angular/core';
import { Subject } from 'rxjs/Subject';

export class SignalRConnection extends SignalRConnectionBase {
    private _status: Observable<ConnectionStatus>;
    private _errors: Observable<any>;
    private _jConnection: any;
    private _jProxy: any;
    private _zone: NgZone;

    constructor(jConnection: any, jProxy: any, zone: NgZone) {
        super();
        this._jProxy = jProxy;
        this._jConnection = jConnection;
        this._zone = zone;
        this._errors = this.wireUpErrorsAsObservable();
        this._status = this.wireUpStatusEventsAsObservable();
    }

    public get errors(): Observable<any> {
        return this._errors;
    }

    public get status(): Observable<ConnectionStatus> {
        return this._status;
    }

    public start(): Promise<any> {
        let $promise = new Promise<any>((resolve, reject) => {
            this._jConnection.start().done(function (...results: any[]) {
                resolve(results);
            })
                .fail(function (err: any) {
                    reject(err);
                });
        });
        return $promise;
    }

    public stop(): void {
        this._jConnection.stop();
    }

    public invoke(method: string, ...parameters: any[]): Promise<any> {
        if (method == null) throw new Error('SignalRConnection: Failed to invoke. Argument \'method\' can not be null');

        console.debug(`SignalRConnection. Start invoking \'${method}\'...`);
        let $promise = new Promise<any>((resolve, reject) => {
            this._jProxy.invoke(method, ...parameters)
                .done(function (result: any) {
                    console.debug(`\'${method}\' invoked succesfully. Resolving promise...`);
                    resolve(result);
                    console.debug(`Promise resolved.`);
                })
                .fail(function (err: any) {
                    console.debug(`Invoking \'${method}\' failed. Rejecting promise...`);
                    reject(err);
                    console.debug(`Promise rejected.`);
                });
        });
        return $promise;
    }

    private wireUpErrorsAsObservable(): Observable<any> {
        let sError = new Subject<any>();

        this._jConnection.error((error: any) => {
            this._zone.run(() => {
                sError.next(error);
            });
        });
        return sError;
    }

    private wireUpStatusEventsAsObservable(): Observable<ConnectionStatus> {
        let sStatus = new Subject<ConnectionStatus>();
        let connStatusNames = ['starting', 'received', 'connectionSlow', 'reconnecting', 'reconnected', 'stateChanged', 'disconnected'];
        // aggregate all signalr connection status handlers into 1 observable. 
        connStatusNames.forEach((statusName) => {
            // handler wire up, for signalr connection status callback.
            this._jConnection[statusName]((...args: any[]) => {
                this._zone.run(() => {
                    sStatus.next(new ConnectionStatus(statusName));
                });
            });
        });
        return sStatus;
    }

    public listen<T>(listener: BroadcastEventListener<T>): void {
        if (listener == null) throw new Error('Failed to listen. Argument \'listener\' can not be null');

        console.log(`SignalRConnection: Starting to listen to server event with name ${listener.event}`);

        this._jProxy.on(listener.event, (...args: any[]) => {
            console.log('SignalRConnection.proxy.on invoked. Calling listener next() ...');
            let casted: T = null;
            if (args.length > 0) casted = <T>args[0];
            
            this._zone.run(() => {
                listener.next(casted);
            });
            console.log('listener next() called.');
        });
    }

    private onBroadcastEventReceived<T>(listener: BroadcastEventListener<T>, ...args: any[]) {
        console.log('SignalRConnection.proxy.on invoked. Calling listener next() ...');

        let casted: T = null;
        if (args.length > 0) casted = <T>args[0];

        this._zone.run(() => {
            listener.next(casted);
        });

        console.log('listener next() called.');
    }

}
{'fraction_non_alphanumeric': 0.13765892152564665, 'fraction_numerical': 0.0010960105217010083, 'mean_word_length': 2.140055058499656, 'pattern_counts': {'":': 0, '<': 39, '<?xml version=': 0, '>': 63, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1671287', 'n_tokens_mistral': 2421, 'n_tokens_neox': 2326, 'n_words': 566}