<commit_before>/// <reference path="../../typings/_custom.d.ts" />
import {Pipe, PipeFactory} from 'angular2/angular2';
import {ObservablePipe} from 'angular2/change_detection';
import {async} from 'angular2/src/change_detection/change_detection';
import * as Rx from 'rx';

export function isObservable(obs) {
  return obs && typeof obs.subscribe === 'function';
}

export class RxPipe extends ObservablePipe {
  _subscription: any;
  _observable: any;
  constructor(ref) { super(ref); }
  supports(obs) { return isObservable(obs); }
  _subscribe(obs) {
    this._observable = obs;
    this._subscription = obs.subscribe(
      value => this._updateLatestValue(value),
      e => { throw e; }
    );
  }
  transform(value: any, args?: List<any>): any {
    return super.transform(value, args);
  }
  onDestroy(): void {
    return super.onDestroy();
  }
}

export class RxPipeFactory implements PipeFactory {
  supports(obs) { return isObservable(obs); }
  create(cdRef): Pipe { return new RxPipe(cdRef); }
}

export var rxAsync = [ new RxPipeFactory() ].concat(async);
<commit_msg>fix(RxPipe): correct typings<commit_after>/// <reference path="../../typings/_custom.d.ts" />
import {Pipe, PipeFactory} from 'angular2/angular2';
import {ObservablePipe} from 'angular2/pipes';
import {async} from 'angular2/src/change_detection/change_detection';
import * as Rx from 'rx';

export function isObservable(obs) {
  return obs && typeof obs.subscribe === 'function';
}

export class RxPipe extends ObservablePipe {
  _subscription: any;
  _observable: any;
  constructor(ref) { super(ref); }
  supports(obs) { return isObservable(obs); }
  _subscribe(obs) {
    this._observable = obs;
    this._subscription = obs.subscribe(
      value => this._updateLatestValue(value),
      e => { throw e; }
    );
  }
  transform(value: any, args?: List<any>): any {
    return super.transform(value, args);
  }
  onDestroy(): void {
    return super.onDestroy();
  }
}

export class RxPipeFactory implements PipeFactory {
  supports(obs) { return isObservable(obs); }
  create(cdRef): Pipe { return new RxPipe(cdRef); }
}

export var rxAsync = [ new RxPipeFactory() ].concat(async);

{'fraction_non_alphanumeric': 0.14331797235023042, 'fraction_numerical': 0.003686635944700461, 'mean_word_length': 4.256658595641646, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 11, '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': '2608091', 'n_tokens_mistral': 711, 'n_tokens_neox': 703, 'n_words': 170}