<commit_before>/**
 * Created by tushar.mathur on 27/09/16.
 */

import {IObservable} from './types/core/IObservable';
import {ISubscription} from './types/core/ISubscription';
import {ISubscriberFunction} from './types/core/ISubscriberFunction';
import {IObserver} from './types/core/IObserver';
import {IScheduler} from './types/IScheduler';
import {DefaultScheduler} from './scheduling/DefaultScheduler';
import {Subscription, CompositeSubscription} from './Subscription';
import {SubscriptionObserver} from './SubscriptionObserver';
import {SafeExecutor} from './lib/SafeExecutor';
import {Safety} from './types/ISafeValue';


export class Observable<T> implements IObservable<T> {
  constructor (private func: ISubscriberFunction<T>) {
  }

  static of<T> (...t: T[]): IObservable<T> {
    return new Observable(observer => {
      for (var i = 0; i < t.length; ++i) {
        observer.next(t[i])
      }
      observer.complete()
    })
  }

  safelyExecuteFunc (observer: IObserver<T>, cSub: CompositeSubscription) {
    const r = SafeExecutor(() => {
      cSub.add(Subscription.from(this.func(SubscriptionObserver.of(observer))))
    })
    if (r.type === Safety.error && observer.error) {
      observer.error(r.value as Error)
    }
  }

  subscribe (observer: IObserver<T>, scheduler: IScheduler = new DefaultScheduler()): ISubscription {
    const subscription = new CompositeSubscription()
    const task = () => this.safelyExecuteFunc(observer, subscription);
    subscription.add(scheduler.scheduleNow(task))
    if (observer.start) observer.start(subscription)
    return subscription
  }
}
<commit_msg>throw type error for non-objects<commit_after>/**
 * Created by tushar.mathur on 27/09/16.
 */

import {IObservable} from './types/core/IObservable';
import {ISubscription} from './types/core/ISubscription';
import {ISubscriberFunction} from './types/core/ISubscriberFunction';
import {IObserver} from './types/core/IObserver';
import {IScheduler} from './types/IScheduler';
import {DefaultScheduler} from './scheduling/DefaultScheduler';
import {Subscription, CompositeSubscription} from './Subscription';
import {SubscriptionObserver} from './SubscriptionObserver';
import {SafeExecutor} from './lib/SafeExecutor';
import {Safety} from './types/ISafeValue';


export class Observable<T> implements IObservable<T> {
  constructor (private func: ISubscriberFunction<T>) {
  }

  static of<T> (...t: T[]): IObservable<T> {
    return new Observable(observer => {
      for (var i = 0; i < t.length; ++i) {
        observer.next(t[i])
      }
      observer.complete()
    })
  }

  safelyExecuteFunc (observer: IObserver<T>, cSub: CompositeSubscription) {
    const r = SafeExecutor(() => {
      cSub.add(Subscription.from(this.func(SubscriptionObserver.of(observer))))
    })
    if (r.type === Safety.error && observer.error) {
      observer.error(r.value as Error)
    }
  }

  subscribe (observer: IObserver<T>, scheduler: IScheduler = new DefaultScheduler()): ISubscription {
    if (typeof observer !== 'object') throw new TypeError('Observer should be of object type')
    const subscription = new CompositeSubscription()
    const task = () => this.safelyExecuteFunc(observer, subscription);
    subscription.add(scheduler.scheduleNow(task))
    if (observer.start) observer.start(subscription)
    return subscription
  }
}

{'fraction_non_alphanumeric': 0.14251639833035182, 'fraction_numerical': 0.0041741204531902205, 'mean_word_length': 4.266875981161696, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 23, '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': '5350160', 'n_tokens_mistral': 1007, 'n_tokens_neox': 1004, 'n_words': 238}