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

/**
 * Groups the items emitted by an Observable according to a specified criterion,
 * and emits these grouped items as `GroupedObservables`, one `GroupedObservable` per group.
 *
 * <img src="./img/groupBy.png" width="100%">
 *
 * @param {Function} keySelector - a function that extracts the key for each item
 * @param {Function} elementSelector - a function that extracts the return element for each item
 * @returns {Observable} an Observable that emits GroupedObservables, each of which corresponds
 * to a unique key value and each of which emits those items from the source Observable that share
 * that key value.
 */
export function groupBy<T, K, R>(keySelector: (value: T) => K,
                                 elementSelector?: (value: T) => R,
                                 durationSelector?: (grouped: GroupedObservable<K, R>) => Observable<any>): Observable<GroupedObservable<K, R>> {
  return this.lift(new GroupByOperator(this, keySelector, elementSelector, durationSelector));
}

export interface RefCountSubscription {
  count: number;
  unsubscribe: () => void;
  isUnsubscribed: boolean;
  attemptedToUnsubscribe: boolean;
}

class GroupByOperator<T, K, R> extends Operator<T, R> {
  constructor(public source: Observable<T>,
              private keySelector: (value: T) => K,
              private elementSelector?: (value: T) => R,
              private durationSelector?: (grouped: GroupedObservable<K, R>) => Observable<any>) {
    super();
  }

  call(subscriber: Subscriber<GroupedObservable<K, R>>): Subscriber<T> {
    return new GroupBySubscriber(
      subscriber, this.keySelector, this.elementSelector, this.durationSelector
    );
  }
}

class GroupBySubscriber<T, K, R> extends Subscriber<T> implements RefCountSubscription {
  private groups: Map<K, Subject<T|R>> = null;
  public attemptedToUnsubscribe: boolean = false;
  public count: number = 0;

  constructor(destination: Subscriber<GroupedObservable<K, R>>,
              private keySelector: (value: T) => K,
              private elementSelector?: (value: T) => R,
              private durationSelector?: (grouped: GroupedObservable<K, R>) => Observable<any>) {
    super();
    this.destination = destination;
    this.add(destination);
  }

  protected _next(value: T): void {
    let key: K;
    try {
      key = this.keySelector(value);
    } catch (err) {
      this.error(err);
      return;
    }

    this._group(value, key);
  }

  private _group(value: T, key: K) {
    let groups = this.groups;

    if (!groups) {
      groups = this.groups = typeof key === 'string' ? new FastMap() : new Map();
    }

    let group = groups.get(key);

    if (!group) {
      groups.set(key, group = new Subject<R>());
      const groupedObservable = new GroupedObservable(key, group, this);

      if (this.durationSelector) {
        this._selectDuration(key, group);
      }

      this.destination.next(groupedObservable);
    }

    if (this.elementSelector) {
      this._selectElement(value, group);
    } else {
      this.tryGroupNext(value, group);
    }
  }

  private _selectElement(value: T, group: Subject<T | R>) {
    let result: R;
    try {
      result = this.elementSelector(value);
    } catch (err) {
      this.error(err);
      return;
    }
    this.tryGroupNext(result, group);
  }

  private _selectDuration(key: K, group: any) {
    let duration: any;
    try {
      duration = this.durationSelector(new GroupedObservable<K, R>(key, group));
    } catch (err) {
      this.error(err);
      return;
    }
    this.add(duration.subscribe(new GroupDurationSubscriber(key, group, this)));
  }

  private tryGroupNext(value: T|R, group: Subject<T | R>): void {
    if (!group.isUnsubscribed) {
      group.next(value);
    }
  }

  protected _error(err: any): void {
    const groups = this.groups;
    if (groups) {
      groups.forEach((group, key) => {
        group.error(err);
        this.removeGroup(key);
      });
    }
    this.destination.error(err);
  }

  protected _complete(): void {
    const groups = this.groups;
    if (groups) {
      groups.forEach((group, key) => {
        group.complete();
        this.removeGroup(key);
      });
    }
    this.destination.complete();
  }

  removeGroup(key: K): void {
    this.groups.delete(key);
  }<commit_msg>refactor(groupBy): clear groups via map.clear interface<commit_after>import {Subscriber} from '../Subscriber';
import {Subscription} from '../Subscription';
import {Observable} from '../Observable';
import {Operator} from '../Operator';
import {Subject} from '../Subject';
import {Map} from '../util/Map';
import {FastMap} from '../util/FastMap';

/**
 * Groups the items emitted by an Observable according to a specified criterion,
 * and emits these grouped items as `GroupedObservables`, one `GroupedObservable` per group.
 *
 * <img src="./img/groupBy.png" width="100%">
 *
 * @param {Function} keySelector - a function that extracts the key for each item
 * @param {Function} elementSelector - a function that extracts the return element for each item
 * @returns {Observable} an Observable that emits GroupedObservables, each of which corresponds
 * to a unique key value and each of which emits those items from the source Observable that share
 * that key value.
 */
export function groupBy<T, K, R>(keySelector: (value: T) => K,
                                 elementSelector?: (value: T) => R,
                                 durationSelector?: (grouped: GroupedObservable<K, R>) => Observable<any>): Observable<GroupedObservable<K, R>> {
  return this.lift(new GroupByOperator(this, keySelector, elementSelector, durationSelector));
}

export interface RefCountSubscription {
  count: number;
  unsubscribe: () => void;
  isUnsubscribed: boolean;
  attemptedToUnsubscribe: boolean;
}

class GroupByOperator<T, K, R> extends Operator<T, R> {
  constructor(public source: Observable<T>,
              private keySelector: (value: T) => K,
              private elementSelector?: (value: T) => R,
              private durationSelector?: (grouped: GroupedObservable<K, R>) => Observable<any>) {
    super();
  }

  call(subscriber: Subscriber<GroupedObservable<K, R>>): Subscriber<T> {
    return new GroupBySubscriber(
      subscriber, this.keySelector, this.elementSelector, this.durationSelector
    );
  }
}

class GroupBySubscriber<T, K, R> extends Subscriber<T> implements RefCountSubscription {
  private groups: Map<K, Subject<T|R>> = null;
  public attemptedToUnsubscribe: boolean = false;
  public count: number = 0;

  constructor(destination: Subscriber<GroupedObservable<K, R>>,
              private keySelector: (value: T) => K,
              private elementSelector?: (value: T) => R,
              private durationSelector?: (grouped: GroupedObservable<K, R>) => Observable<any>) {
    super();
    this.destination = destination;
    this.add(destination);
  }

  protected _next(value: T): void {
    let key: K;
    try {
      key = this.keySelector(value);
    } catch (err) {
      this.error(err);
      return;
    }

    this._group(value, key);
  }

  private _group(value: T, key: K) {
    let groups = this.groups;

    if (!groups) {
      groups = this.groups = typeof key === 'string' ? new FastMap() : new Map();
    }

    let group = groups.get(key);

    if (!group) {
      groups.set(key, group = new Subject<R>());
      const groupedObservable = new GroupedObservable(key, group, this);

      if (this.durationSelector) {
        this._selectDuration(key, group);
      }

      this.destination.next(groupedObservable);
    }

    if (this.elementSelector) {
      this._selectElement(value, group);
    } else {
      this.tryGroupNext(value, group);
    }
  }

  private _selectElement(value: T, group: Subject<T | R>) {
    let result: R;
    try {
      result = this.elementSelector(value);
    } catch (err) {
      this.error(err);
      return;
    }
    this.tryGroupNext(result, group);
  }

  private _selectDuration(key: K, group: any) {
    let duration: any;
    try {
      duration = this.durationSelector(new GroupedObservable<K, R>(key, group));
    } catch (err) {
      this.error(err);
      return;
    }
    this.add(duration.subscribe(new GroupDurationSubscriber(key, group, this)));
  }

  private tryGroupNext(value: T|R, group: Subject<T | R>): void {
    if (!group.isUnsubscribed) {
      group.next(value);
    }
  }

  protected _error(err: any): void {
    const groups = this.groups;
    if (groups) {
      groups.forEach((group, key) => {
        group.error(err);
      });

      groups.clear();
    }
    this.destination.error(err);
  }

  protected _complete(): void {
    const groups = this.groups;
    if (groups) {
      groups.forEach((group, key) => {
        group.complete();
      });

      groups.clear();
    }
    this.destination.complete();
  }

  removeGroup(key: K): void {
    this.groups.delete(key);
  }
{'fraction_non_alphanumeric': 0.13292788879235448, 'fraction_numerical': 0.0008688097306689834, 'mean_word_length': 3.0301969365426698, 'pattern_counts': {'":': 0, '<': 55, '<?xml version=': 0, '>': 79, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5075045', 'n_tokens_mistral': 2771, 'n_tokens_neox': 2721, 'n_words': 749}