<commit_before>import * as DebugFlags from "./debug-flags";
import { Feedback } from "./coq/feedback";
import * as FeedbackContent from "./coq/feedback-content";
import { Message } from "./coq/message";
import { ValueFail } from "./coq/value-fail";
import * as CoqtopInput from "./coqtop-input";
import { processSequentiallyForever } from "./rx";

let statusPeriod = 10000; // milliseconds

export function setupCoqtopCommunication(
  input$s: Rx.Observable<ICoqtopInput>[]
): CoqtopOutputStreams {

  let inputStatus$: Rx.Observable<ICoqtopInput> =
    Rx.Observable
      .interval(statusPeriod)
      .map(() => new CoqtopInput.Status(false));

  let inputSubject: Rx.Subject<ICoqtopInput> =
    new Rx.Subject<ICoqtopInput>();

  let input$: Rx.ConnectableObservable<ICoqtopInput> =
    Rx.Observable
      .merge(
      inputStatus$,
      inputSubject.asObservable(),
      ...input$s
      )
      .publish();

  // subscribeAndLog(coqtopInputStream);

  let outputAndError$s = processSequentiallyForever(input$, processCommands);

  const io$ = outputAndError$s.output$;

  // const error$ = outputAndError$s.error$
  //   .map(r => {
  //     return new ValueFail(r.response.contents);
  //   })
  //   .share();

  // error$.subscribe(vf => inputSubject.onNext(new CoqtopInput.EditAt(vf.stateId)));

  if (DebugFlags.requestToCoqtop) {
    input$
      .filter(i => !(i instanceof CoqtopInput.Status))
      .subscribe(input => { console.log("⟸", input); });
  }
  if (DebugFlags.responseFromCoqtop) {
    io$
      .filter(io => !(io.input instanceof CoqtopInput.Status))
      .subscribe(io => { console.log("   ⟹", io.input, io.output); });
  }
  // if (DebugFlags.errorFromCoqtop) {
  //   error$
  //     .subscribe(vf => { console.log("   ⟹ ERROR", vf); });
  // }

  let message$: Rx.Observable<IMessage> =
    io$
      .concatMap(io => _(io.output.messages).map(m => new Message(m)).value())
      .share();
  if (DebugFlags.messageFromCoqtop) {
    message$.subscribe(m => console.log("Message", m));
  }

  /* IMPORTANT: It is important to process Feedback messages before
  processing the output value, because a ValueFail may refer to state
  IDs that are only present in the current output's feedback. We
  artificially delay the output value stream to achieve this easily. */

  const feedback$: Rx.Subject<IFeedback<IFeedbackContent>> = new Rx.Subject<IFeedback<IFeedbackContent>>();
  const value$: Rx.Subject<ICoqtopResponse<any>> = new Rx.Subject<any>();

  io$.subscribe(io => {
    Rx.Observable.fromArray(_(io.output.feedback).map(f => new Feedback(f)).value())
      .subscribe(f => feedback$.onNext(f));
    console.log("All feedbacks should have been processed, now pushing return value");
    value$.onNext(io.output.response);
  });
  if (DebugFlags.feedbackFromCoqtop) {
    feedback$
      .distinctUntilChanged() // so redundant when calling status
      .subscribe(f => console.log("Feedback", f));
  }
<commit_msg>comment logging<commit_after>import * as DebugFlags from "./debug-flags";
import { Feedback } from "./coq/feedback";
import * as FeedbackContent from "./coq/feedback-content";
import { Message } from "./coq/message";
import { ValueFail } from "./coq/value-fail";
import * as CoqtopInput from "./coqtop-input";
import { processSequentiallyForever } from "./rx";

let statusPeriod = 10000; // milliseconds

export function setupCoqtopCommunication(
  input$s: Rx.Observable<ICoqtopInput>[]
): CoqtopOutputStreams {

  let inputStatus$: Rx.Observable<ICoqtopInput> =
    Rx.Observable
      .interval(statusPeriod)
      .map(() => new CoqtopInput.Status(false));

  let inputSubject: Rx.Subject<ICoqtopInput> =
    new Rx.Subject<ICoqtopInput>();

  let input$: Rx.ConnectableObservable<ICoqtopInput> =
    Rx.Observable
      .merge(
      inputStatus$,
      inputSubject.asObservable(),
      ...input$s
      )
      .publish();

  // subscribeAndLog(coqtopInputStream);

  let outputAndError$s = processSequentiallyForever(input$, processCommands);

  const io$ = outputAndError$s.output$;

  // const error$ = outputAndError$s.error$
  //   .map(r => {
  //     return new ValueFail(r.response.contents);
  //   })
  //   .share();

  // error$.subscribe(vf => inputSubject.onNext(new CoqtopInput.EditAt(vf.stateId)));

  if (DebugFlags.requestToCoqtop) {
    input$
      .filter(i => !(i instanceof CoqtopInput.Status))
      .subscribe(input => { console.log("⟸", input); });
  }
  if (DebugFlags.responseFromCoqtop) {
    io$
      .filter(io => !(io.input instanceof CoqtopInput.Status))
      .subscribe(io => { console.log("   ⟹", io.input, io.output); });
  }
  // if (DebugFlags.errorFromCoqtop) {
  //   error$
  //     .subscribe(vf => { console.log("   ⟹ ERROR", vf); });
  // }

  let message$: Rx.Observable<IMessage> =
    io$
      .concatMap(io => _(io.output.messages).map(m => new Message(m)).value())
      .share();
  if (DebugFlags.messageFromCoqtop) {
    message$.subscribe(m => console.log("Message", m));
  }

  /* IMPORTANT: It is important to process Feedback messages before
  processing the output value, because a ValueFail may refer to state
  IDs that are only present in the current output's feedback. We
  artificially delay the output value stream to achieve this easily. */

  const feedback$: Rx.Subject<IFeedback<IFeedbackContent>> = new Rx.Subject<IFeedback<IFeedbackContent>>();
  const value$: Rx.Subject<ICoqtopResponse<any>> = new Rx.Subject<any>();

  io$.subscribe(io => {
    Rx.Observable.fromArray(_(io.output.feedback).map(f => new Feedback(f)).value())
      .subscribe(f => feedback$.onNext(f));
    // console.log("All feedbacks should have been processed, now pushing return value");
    value$.onNext(io.output.response);
  });
  if (DebugFlags.feedbackFromCoqtop) {
    feedback$
      .distinctUntilChanged() // so redundant when calling status
      .subscribe(f => console.log("Feedback", f));
  }

{'fraction_non_alphanumeric': 0.1524421159371303, 'fraction_numerical': 0.0016900456312320432, 'mean_word_length': 3.507235338918507, 'pattern_counts': {'":': 0, '<': 29, '<?xml version=': 0, '>': 59, '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': '6693637', 'n_tokens_mistral': 1951, 'n_tokens_neox': 1866, 'n_words': 413}