<commit_before>import { Injectable } from '@angular/core';

import { Observable } from 'rxjs/Observable';
import { Observer } from 'rxjs/Observer';
import { Subject } from 'rxjs/Subject';

export enum ReadyState {
    Connecting  = 0,
    Open        = 1,
    Closing     = 2,
    Closed      = 3,
}

export class WebSocketError extends Error {
  code: number;
  reason: string;
}

@Injectable()
export class WebSocketService {
  public connect<T>(url: string): Subject<T> {
    if (url[0] == '/') {
      url = (location.protocol == 'http:' ? 'ws:' : 'wss:') + '//' + location.host + url;
    }
    let ws = new WebSocket(url);
    let subject = new Subject<T>();

    ws.onopen = (event: Event) => {
      // wait for first message before advancing subject
    };
    ws.onerror = (event: Event) => {
      // the websocket error event is intentionally useless, it does not contain any other information
      // just wait for the close event...
      console.log("WebSocket onerror", event);
    };
    ws.onmessage = (event: MessageEvent) => {
      subject.next(JSON.parse(event.data));
    };
    ws.onclose = (closeEvent: CloseEvent) => {
      let error = new WebSocketError("Websocket closed with code=" + closeEvent.code + ": " + closeEvent.reason);
      error.code = closeEvent.code;
      error.reason = closeEvent.reason;

      if (closeEvent.wasClean) {
        subject.complete();
      } else {
        subject.error(error);
      }
    };

    return subject;
  }
}<commit_msg>web/lib/websocket: convert from subject to pure Observable<commit_after>import { Injectable } from '@angular/core';

import { Observable } from 'rxjs/Observable';
import { Observer } from 'rxjs/Observer';

export enum ReadyState {
    Connecting  = 0,
    Open        = 1,
    Closing     = 2,
    Closed      = 3,
}

export class WebSocketError extends Error {
  code: number;
  reason: string;
}

@Injectable()
export class WebSocketService {
  public connect<T>(url: string): Observable<T> {
    if (url[0] == '/') {
      url = (location.protocol == 'http:' ? 'ws:' : 'wss:') + '//' + location.host + url;
    }

    return Observable.create(function (observer: Observer<T>) {
      console.log("WebSocket connect");

      let ws = new WebSocket(url);

      ws.onopen = (event: Event) => {
        // wait for first message before advancing subject
        console.log("WebSocket open");
      };
      ws.onerror = (event: Event) => {
        // the websocket error event is intentionally useless, it does not contain any other information
        // just wait for the close event...
        console.log("WebSocket error", event);
      };
      ws.onmessage = (event: MessageEvent) => {
        observer.next(JSON.parse(event.data));
      };
      ws.onclose = (closeEvent: CloseEvent) => {
        console.log("WebSocket close", closeEvent);

        let error = new WebSocketError("Websocket closed with code=" + closeEvent.code + ": " + closeEvent.reason);
        error.code = closeEvent.code;
        error.reason = closeEvent.reason;

        if (closeEvent.wasClean) {
          observer.complete();
        } else {
          observer.error(error);
        }
      };

      return function() {
        console.log("WebSocket disconnect");

        ws.close();
      }
    });
  }
}
{'fraction_non_alphanumeric': 0.1377845220030349, 'fraction_numerical': 0.0030349013657056147, 'mean_word_length': 2.5026567481402764, 'pattern_counts': {'":': 2, '<': 9, '<?xml version=': 0, '>': 17, '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': '1339555', 'n_tokens_mistral': 1007, 'n_tokens_neox': 966, 'n_words': 260}