<commit_before>import { Observable, Subject, IObserver, IDisposable } from 'rx';

import { Command } from './Interfaces';
import { asObservable, isObserver } from './Utils';

export class ObservableCommand<T> implements Command<T>, IDisposable {
  protected isExecutingSubject: Subject<boolean>;
  protected resultsSubject: Subject<T>;
  protected thrownErrorsSubject: Subject<Error>;

  public readonly isExecutingObservable: Observable<boolean>;
  public readonly canExecuteObservable: Observable<boolean>;
  public readonly results: Observable<T>;
  public readonly thrownErrors: Observable<Error>;

  constructor(
    protected executeAction: (parameter: any) => Observable<T>,
    canExecute: Observable<boolean> = asObservable(true),
  ) {
    this.isExecutingSubject = new Subject<boolean>();
    this.resultsSubject = new Subject<T>();
    this.thrownErrorsSubject = new Subject<Error>();

    this.isExecutingObservable = this.isExecutingSubject
      .distinctUntilChanged()
      .startWith(false)
      .shareReplay(1);

    this.canExecuteObservable = canExecute
      .combineLatest(this.isExecutingObservable, (ce, ie) => ce === true && ie === false)
      .catch(e => {
        const err = e instanceof Error ? e : new Error(e);
        this.thrownErrorsSubject.onNext(err);

        return asObservable(false);
      })
      .startWith(false)
      .distinctUntilChanged()
      .shareReplay(1);

    this.results = this.resultsSubject
      .share();

    this.thrownErrors = this.thrownErrorsSubject
      .share();
  }

  get isExecuting() {
    // COMPAT -- we need to return the observable instead of the current value
    // let result: boolean | undefined;

    // this.isExecutingObservable
    //   .take(1)
    //   .subscribe(x => {
    //     result = x;
    //   });

    // if (result == null) {
    //   this.thrownErrorsSubject
    //     .onNext(new Error('isExecuting Observable has terminated abnormally'));

    //   result = false;
    // }

    // return result;
    return this.isExecutingObservable;
  }

  // COMPAT -- we need to define this as a function instead of a getter
  // get canExecute() {
  canExecute(parameter?: any) {
    let result: boolean | undefined;

    this.canExecuteObservable
      .take(1)
      .subscribe(x => {
        result = x;
      });

    if (result == null) {
      this.thrownErrorsSubject
        .onNext(new Error('canExecute Observable has terminated abnormally'));

      result = false;
    }

    return result;
  }

  observeExecution(parameter?: any) {
    if (this.canExecute() === false) {
      return Observable.throw<T>(new Error('canExecute currently forbids execution'));
    }

    return Observable
      .defer(() => {
        this.isExecutingSubject.onNext(true);

        return Observable.empty<T>();
      })
      .concat(
        Observable.defer(() => {
          return this.executeAction(parameter);
        }),
      )
      .do(
        x => {
          this.resultsSubject.onNext(x);
        },
        undefined,
        () => {
          this.isExecutingSubject.onNext(false);
        },
      )
      .catch(e => {
        const err = e instanceof Error ? e : new Error(e);
        this.thrownErrorsSubject.onNext(err);

        return Observable.empty<T>();
      })
      .share();
  }

  execute(
    parameter?: any,
    observerOrNext?: IObserver<T> | ((value: T) => void),
    onError?: (exception: any) => void,
    onCompleted?: () => void,
  ) {
    return this
      .observeExecution(parameter)
      .subscribeWith(observerOrNext, onError, onCompleted);
  }

  dispose() {
    this.isExecutingSubject = Object.dispose(this.isExecutingSubject);
    this.resultsSubject = Object.dispose(this.resultsSubject);
    this.thrownErrorsSubject = Object.dispose(this.thrownErrorsSubject);
  }
}

export function command<T>(
  execute: (parameter: any) => T | Observable<T> = (x: T) => x,
  canExecute?: Observable<boolean>,
): Command<T> {
  return new ObservableCommand((parameter: any) => asObservable(execute(parameter)), canExecute);
}<commit_msg>fixing observable commands now subscribing to the canExecute composition and forwarding events onto the canExecute subject<commit_after>import { Observable, Subject, BehaviorSubject, IObserver, IDisposable, Scheduler } from 'rx';

import { Command } from './Interfaces';
import { asObservable, isObserver, handleError } from './Utils';

export class ObservableCommand<T> implements Command<T>, IDisposable {
  private canExecuteSubscription: IDisposable;

  protected isExecutingSubject: BehaviorSubject<boolean>;
  protected canExecuteSubject: BehaviorSubject<boolean>;
  protected resultsSubject: Subject<T>;
  protected thrownErrorsSubject: Subject<Error>;

  constructor(
    protected executeAction: (parameter: any) => Observable<T>,
    canExecute?: Observable<boolean>,
  ) {
    this.isExecutingSubject = new BehaviorSubject<boolean>(false);
    this.canExecuteSubject = new BehaviorSubject<boolean>(canExecute == null);
    this.resultsSubject = new Subject<T>();
    this.thrownErrorsSubject = new Subject<Error>();

    this.canExecuteSubscription = (canExecute || asObservable(true))
      .combineLatest(this.isExecutingSubject, (ce, ie) => ce === true && ie === false)
      .catch(e => {
        handleError(e, this.thrownErrorsSubject);

        return asObservable(false);
      })
      .distinctUntilChanged()
      .subscribe(
        x => {
          this.canExecuteSubject.onNext(x);
        },
      );
  }

  get isExecutingObservable() {
    return this.isExecutingSubject
      .distinctUntilChanged();
  }

  get isExecuting() {
    // COMPAT -- we need to return the observable instead of the current value
    // return this.isExecutingSubject.getValue();

    return this.isExecutingObservable;
  }

  get canExecuteObservable() {
    return this.canExecuteSubject
      .distinctUntilChanged();
  }

  // COMPAT -- we need to define this as a function instead of a getter
  // get canExecute() {
  canExecute(parameter?: any) {
    return this.canExecuteSubject.getValue();
  }

  observeExecution(parameter?: any) {
    if (this.canExecute() === false) {
      return Observable.throw<T>(new Error('canExecute currently forbids execution'));
    }

    return Observable
      .of(parameter)
      .flatMap(x => {
        this.isExecutingSubject.onNext(true);

        return this.executeAction(x);
      })
      .do(
        x => {
          this.resultsSubject.onNext(x);

          this.isExecutingSubject.onNext(false);
        },
        e => {
          this.isExecutingSubject.onNext(false);
        },
        () => {
          this.isExecutingSubject.onNext(false);
        },
      )
      .catch(e => {
        handleError(e, this.thrownErrorsSubject);

        return Observable.empty<T>();
      })
      // this will prevent execution if nobody is subscribing to the result
      .share();
  }

  execute(
    parameter?: any,
    observerOrNext?: IObserver<T> | ((value: T) => void),
    onError: (exception: any) => void = e => handleError(e, this.thrownErrorsSubject),
    onCompleted?: () => void,
  ) {
    return this
      .observeExecution(parameter)
      .subscribeWith(observerOrNext, onError, onCompleted);
  }

  get results() {
    return this.resultsSubject
      .asObservable();
  }

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

  dispose() {
    this.canExecuteSubscription = Object.dispose(this.canExecuteSubscription);
    this.isExecutingSubject = Object.dispose(this.isExecutingSubject);
    this.resultsSubject = Object.dispose(this.resultsSubject);
    this.thrownErrorsSubject = Object.dispose(this.thrownErrorsSubject);
  }
}

export function command<T>(
  execute: (parameter: any) => T | Observable<T> = (x: T) => x,
  canExecute?: Observable<boolean>,
): Command<T> {
  return new ObservableCommand((parameter: any) => asObservable(execute(parameter)), canExecute);
}
{'fraction_non_alphanumeric': 0.12511002137558155, 'fraction_numerical': 0.0005029548598013328, 'mean_word_length': 3.0232675771370765, 'pattern_counts': {'":': 0, '<': 44, '<?xml version=': 0, '>': 76, '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': '3473473', 'n_tokens_mistral': 2301, 'n_tokens_neox': 2241, 'n_words': 511}