<commit_before>import { Observable, Subject, BehaviorSubject, Subscription } from 'rxjs';

import { Property } from './Interfaces';
import { isSubject, handleError } from './Utils';

export class ObservableProperty<T> extends Subscription implements Property<T> {
  protected changedSubject: BehaviorSubject<T>;
  protected thrownErrorsSubject: Subject<Error>;

  constructor(
    initialValue?: T,
    protected source: Observable<T> = new Subject<T>(),
  ) {
    super();

    this.changedSubject = this.addSubscription(new BehaviorSubject<T>(initialValue!));
    this.thrownErrorsSubject = this.addSubscription(new Subject<Error>());

    this.add(
      this.source
        // seed the observable subscription with the initial value so that
        // distinctUntilChanged knows what the initial value is
        .startWith(initialValue!)
        .distinctUntilChanged()
        .subscribe(
          x => {
            if (this.isNewValue(x)) {
              this.changedSubject.next(x);
            }
          },
          e => {
            handleError(e, this.thrownErrorsSubject);
          },
        ),
    );
  }

  protected isNewValue(newValue: T) {
    return newValue !== this.value;
  }

  get isReadOnly() {
    return isSubject(this.source) === false;
  }

  get value() {
    return this.changedSubject.getValue();
  }

  set value(newValue: T) {
    if (isSubject(this.source)) {
      this.source.next(newValue);
    }
    else {
      throw new Error('attempt to write to a read-only observable property');
    }
  }

  get changed() {
    return this.changedSubject
      // BehaviorSubject fires immediately, so skip the first event
      .skip(1);
  }

  get thrownErrors() {
    return this.thrownErrorsSubject
      .asObservable();
  }

  isProperty() {
    return true;
  }
}

export function property<T>(
  initialValue?: T,
  source?: Observable<T>,
): Property<T> {
  return new ObservableProperty(initialValue, source);
}<commit_msg>updating observable property to support custom comparator and key selector functions resolves #138<commit_after>import { Observable, Subject, BehaviorSubject, Subscription } from 'rxjs';

import { Property } from './Interfaces';
import { isObservable, isSubject, handleError } from './Utils';

export class ObservableProperty<T> extends Subscription implements Property<T> {
  protected changedSubject: BehaviorSubject<T>;
  protected thrownErrorsSubject: Subject<Error>;

  constructor(
    initialValue?: T,
    compare?: (x: T, y: T) => boolean,
    keySelector?: (x: T) => any,
    protected source: Observable<T> = new Subject<T>(),
  ) {
    super();

    this.changedSubject = this.addSubscription(new BehaviorSubject<T>(initialValue!));
    this.thrownErrorsSubject = this.addSubscription(new Subject<Error>());

    this.add(
      this.source
        // seed the observable subscription with the initial value so that
        // distinctUntilChanged knows what the initial value is
        .startWith(initialValue!)
        .distinctUntilChanged(compare!, keySelector!)
        .subscribe(
          x => {
            this.changedSubject.next(x);
          },
          e => {
            handleError(e, this.thrownErrorsSubject);
          },
        ),
    );
  }

  get isReadOnly() {
    return isSubject(this.source) === false;
  }

  get value() {
    return this.changedSubject.getValue();
  }

  set value(newValue: T) {
    if (isSubject(this.source)) {
      this.source.next(newValue);
    }
    else {
      throw new Error('attempt to write to a read-only observable property');
    }
  }

  get changed() {
    return this.changedSubject
      // BehaviorSubject fires immediately, so skip the first event
      .skip(1);
  }

  get thrownErrors() {
    return this.thrownErrorsSubject
      .asObservable();
  }

  isProperty() {
    return true;
  }
}

export function property<T>(
  initialValue?: T,
  source?: Observable<T>,
): Property<T>;

export function property<T>(
  initialValue?: T,
  compare?: (x: T, y: T) => boolean,
  source?: Observable<T>,
): Property<T>;

export function property<T>(
  initialValue?: T,
  compare?: (x: T, y: T) => boolean,
  keySelector?: (x: T) => any,
  source?: Observable<T>,
): Property<T>;

export function property<T>(...args: any[]): Property<T> {
  const initialValue: T = args.shift();

  let compare: undefined | ((x: T, y: T) => boolean);
  let keySelector: undefined | ((x: T) => any);
  let source: undefined | Observable<T>;

  let arg = args.shift();

  if (isObservable<T>(arg)) {
    source = arg;
  }
  else {
    compare = arg;
  }

  arg = args.shift();

  if (isObservable<T>(arg)) {
    source = arg;
  }
  else {
    keySelector = arg;
  }

  arg = args.shift();

  if (isObservable<T>(arg)) {
    source = arg;
  }

  return new ObservableProperty(initialValue, compare, keySelector, source);
}
{'fraction_non_alphanumeric': 0.1290455576169862, 'fraction_numerical': 0.0010307153164296021, 'mean_word_length': 2.95114006514658, 'pattern_counts': {'":': 0, '<': 37, '<?xml version=': 0, '>': 48, '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': '3061063', 'n_tokens_mistral': 1448, 'n_tokens_neox': 1406, 'n_words': 378}