<commit_before>import {Operator} from '../Operator';
import {Subscriber} from '../Subscriber';
import {Observable} from '../Observable';
import {Subject} from '../Subject';
import {Subscription} from '../Subscription';

import {tryCatch} from '../util/tryCatch';
import {errorObject} from '../util/errorObject';

export function windowWhen<T>(closingSelector: () => Observable<any>): Observable<Observable<T>> {
  return this.lift(new WindowOperator(closingSelector));
}

class WindowOperator<T, R> implements Operator<T, R> {

  constructor(private closingSelector: () => Observable<any>) {
  }

  call(subscriber: Subscriber<T>): Subscriber<T> {
    return new WindowSubscriber(subscriber, this.closingSelector);
  }
}

class WindowSubscriber<T> extends Subscriber<T> {
  private window: Subject<T> = new Subject<T>();
  private closingNotification: Subscription<any>;

  constructor(destination: Subscriber<T>, private closingSelector: () => Observable<any>) {
    super(destination);
    this.openWindow();
  }

  _next(value: T) {
    this.window.next(value);
  }

  _error(err: any) {
    this.window.error(err);
    this.destination.error(err);
    this._unsubscribeClosingNotification();
  }

  _complete() {
    this.window.complete();
    this.destination.complete();
    this._unsubscribeClosingNotification();
  }

  unsubscribe() {
    super.unsubscribe();
    this._unsubscribeClosingNotification();
  }

  _unsubscribeClosingNotification() {
    let closingNotification = this.closingNotification;
    if (closingNotification) {
      closingNotification.unsubscribe();
    }
  }

  openWindow() {
    const prevClosingNotification = this.closingNotification;
    if (prevClosingNotification) {
      this.remove(prevClosingNotification);
      prevClosingNotification.unsubscribe();
    }

    const prevWindow = this.window;
    if (prevWindow) {
      prevWindow.complete();
    }

    this.destination.next(this.window = new Subject<T>());

    let closingNotifier = tryCatch(this.closingSelector)();
    if (closingNotifier === errorObject) {
      const err = closingNotifier.e;
      this.destination.error(err);
      this.window.error(err);
    } else {
      let closingNotification = this.closingNotification = new Subscription();
      this.add(closingNotification.add(closingNotifier._subscribe(new WindowClosingNotifierSubscriber(this))));
    }
  }
}

class WindowClosingNotifierSubscriber<T> extends Subscriber<T> {
  constructor(private parent: WindowSubscriber<any>) {
    super(null);
  }

  _next() {
    this.parent.openWindow();
  }

  _error(err) {
    this.parent.error(err);
  }

  _complete() {
    this.parent.openWindow();
  }
}
<commit_msg>fix(windowWhen): fix windowWhen to dispose window Subjects<commit_after>import {Operator} from '../Operator';
import {Subscriber} from '../Subscriber';
import {Observable} from '../Observable';
import {Subject} from '../Subject';
import {Subscription} from '../Subscription';

import {tryCatch} from '../util/tryCatch';
import {errorObject} from '../util/errorObject';

export function windowWhen<T>(closingSelector: () => Observable<any>): Observable<Observable<T>> {
  return this.lift(new WindowOperator(closingSelector));
}

class WindowOperator<T, R> implements Operator<T, R> {

  constructor(private closingSelector: () => Observable<any>) {
  }

  call(subscriber: Subscriber<Observable<T>>): Subscriber<T> {
    return new WindowSubscriber(subscriber, this.closingSelector);
  }
}

class WindowSubscriber<T> extends Subscriber<T> {
  private window: Subject<T>;
  private closingNotification: Subscription<any>;

  constructor(protected destination: Subscriber<Observable<T>>,
              private closingSelector: () => Observable<any>) {
    super(destination);
    this.openWindow();
  }

  _next(value: T) {
    this.window.next(value);
  }

  _error(err: any) {
    this.window.error(err);
    this.destination.error(err);
    this._unsubscribeClosingNotification();
  }

  _complete() {
    this.window.complete();
    this.destination.complete();
    this._unsubscribeClosingNotification();
  }

  unsubscribe() {
    super.unsubscribe();
    this._unsubscribeClosingNotification();
  }

  _unsubscribeClosingNotification() {
    let closingNotification = this.closingNotification;
    if (closingNotification) {
      closingNotification.unsubscribe();
    }
  }

  openWindow() {
    const prevClosingNotification = this.closingNotification;
    if (prevClosingNotification) {
      this.remove(prevClosingNotification);
      prevClosingNotification.unsubscribe();
    }

    const prevWindow = this.window;
    if (prevWindow) {
      prevWindow.complete();
    }

    const window = this.window = new Subject<T>();
    this.destination.next(window);

    const closingNotifier = tryCatch(this.closingSelector)();
    if (closingNotifier === errorObject) {
      const err = closingNotifier.e;
      this.destination.error(err);
      this.window.error(err);
    } else {
      const closingNotification = this.closingNotification = new Subscription();
      closingNotification.add(closingNotifier._subscribe(new WindowClosingNotifierSubscriber(this)));
      this.add(closingNotification);
      this.add(window);
    }
  }
}

class WindowClosingNotifierSubscriber<T> extends Subscriber<T> {
  constructor(private parent: WindowSubscriber<any>) {
    super(null);
  }

  _next() {
    this.parent.openWindow();
  }

  _error(err) {
    this.parent.error(err);
  }

  _complete() {
    this.parent.openWindow();
  }
}

{'fraction_non_alphanumeric': 0.1337798696596669, 'fraction_numerical': 0.0, 'mean_word_length': 4.0968634686346865, 'pattern_counts': {'":': 0, '<': 44, '<?xml version=': 0, '>': 50, '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': '7218540', 'n_tokens_mistral': 1638, 'n_tokens_neox': 1583, 'n_words': 296}