<commit_before>// Copyright (C) 2003-2009 by Object Mentor, Inc. All rights reserved.
// Released under the terms of the CPL Common Public License version 1.0.
package fitnesse.testsystems.fit;

import java.io.IOException;
import java.io.OutputStream;
import java.net.Socket;

import util.StreamReader;
import fit.Counts;
import fit.FitProtocol;
import fitnesse.testsystems.TestSummary;
import fitnesse.testsystems.TestSystemListener;

public class FitClient {

  private TestSystemListener listener;
  private Socket fitSocket;
  private OutputStream fitInput;
  private StreamReader fitOutput;

  private volatile int sent = 0;
  private volatile int received = 0;
  private volatile boolean isDoneSending = false;
  protected volatile boolean killed = false;
  protected Thread fitListeningThread;

  public FitClient(TestSystemListener listener) {
    this.listener = listener;
  }

  public synchronized void acceptSocket(Socket socket) throws IOException, InterruptedException {
    checkForPulse();
    fitSocket = socket;
    fitInput = fitSocket.getOutputStream();
    FitProtocol.writeData("", fitInput);
    fitOutput = new StreamReader(fitSocket.getInputStream());

    fitListeningThread = new Thread(new FitListeningRunnable(), "FitClient fitOutput");
    fitListeningThread.start();
  }

  public void send(String data) throws IOException, InterruptedException {
    checkForPulse();
    FitProtocol.writeData(data, fitInput);
    sent++;
  }

  public void done() throws IOException, InterruptedException {
    checkForPulse();
    FitProtocol.writeSize(0, fitInput);
    isDoneSending = true;
  }

  public void join() {
    if (fitListeningThread != null)
      try {
        fitListeningThread.join();
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
  }

  public void kill() {
    killed = true;
    if (fitListeningThread != null)
      fitListeningThread.interrupt();
  }

  public synchronized boolean isSuccessfullyStarted() {
    return fitSocket != null;
  }

  public void exceptionOccurred(Exception e) {
    listener.exceptionOccurred(e);
  }

  protected void checkForPulse() throws InterruptedException {
    if (killed)
      throw new InterruptedException("FitClient was killed");
  }

  private void listenToFit() {
    try {
      attemptToListenToFit();
    } catch (Exception e) {
      exceptionOccurred(e);
    }
  }

  private void attemptToListenToFit() throws IOException {
    while (!finishedReading()) {
      int size;
      size = FitProtocol.readSize(fitOutput);
      if (size != 0) {
        String readValue = fitOutput.read(size);
        if (fitOutput.byteCount() < size)
          throw new IOException("I was expecting " + size + " bytes but I only got " + fitOutput.byteCount());
        listener.testOutputChunk(readValue);
      } else {
        Counts counts = FitProtocol.readCounts(fitOutput);
        TestSummary summary = new TestSummary();
        summary.right = counts.right;
        summary.wrong = counts.wrong;
        summary.ignores = counts.ignores;
        summary.exceptions = counts.exceptions;
        listener.testComplete(summary);
        received++;
      }
    }
  }

  private boolean finishedReading() {
    while (stateIndeterminate())
      shortSleep();
    return isDoneSending && received == sent;
  }

  /**
   * @return true if the current state of the transission is indeterminate.
   *         <p/>
   *         When the number of pages sent and recieved is the same, we may be done with the whole job, or we may just
   *         be waiting for FitNesse to send the next page. There's no way to know until FitNesse either calls send, or
   *         done.
   */
  private boolean stateIndeterminate() {
    return (received == sent) && !isDoneSending;
  }

  private void shortSleep() {
    try {
      Thread.sleep(10);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
  }

  private class FitListeningRunnable implements Runnable {
    public void run() {
      listenToFit();
    }
  }

}<commit_msg>Move all Fit listener code in the responsible thread.<commit_after>// Copyright (C) 2003-2009 by Object Mentor, Inc. All rights reserved.
// Released under the terms of the CPL Common Public License version 1.0.
package fitnesse.testsystems.fit;

import java.io.IOException;
import java.io.OutputStream;
import java.net.Socket;

import util.StreamReader;
import fit.Counts;
import fit.FitProtocol;
import fitnesse.testsystems.TestSummary;
import fitnesse.testsystems.TestSystemListener;

public class FitClient {

  private TestSystemListener listener;
  private Socket fitSocket;
  private OutputStream fitInput;
  private StreamReader fitOutput;

  private volatile int sent = 0;
  private volatile int received = 0;
  private volatile boolean isDoneSending = false;
  protected volatile boolean killed = false;
  protected Thread fitListeningThread;

  public FitClient(TestSystemListener listener) {
    this.listener = listener;
  }

  public synchronized void acceptSocket(Socket socket) throws IOException, InterruptedException {
    checkForPulse();
    fitSocket = socket;
    fitInput = fitSocket.getOutputStream();
    FitProtocol.writeData("", fitInput);
    fitOutput = new StreamReader(fitSocket.getInputStream());

    fitListeningThread = new Thread(new FitListeningRunnable(), "FitClient fitOutput");
    fitListeningThread.start();
  }

  public void send(String data) throws IOException, InterruptedException {
    checkForPulse();
    FitProtocol.writeData(data, fitInput);
    sent++;
  }

  public void done() throws IOException, InterruptedException {
    checkForPulse();
    FitProtocol.writeSize(0, fitInput);
    isDoneSending = true;
  }

  public void join() {
    if (fitListeningThread != null)
      try {
        fitListeningThread.join();
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
  }

  public void kill() {
    killed = true;
    if (fitListeningThread != null)
      fitListeningThread.interrupt();
  }

  public synchronized boolean isSuccessfullyStarted() {
    return fitSocket != null;
  }

  public void exceptionOccurred(Exception e) {
    listener.exceptionOccurred(e);
  }

  protected void checkForPulse() throws InterruptedException {
    if (killed)
      throw new InterruptedException("FitClient was killed");
  }

  private boolean finishedReading() {
    while (stateIndeterminate())
      shortSleep();
    return isDoneSending && received == sent;
  }

  /**
   * @return true if the current state of the transission is indeterminate.
   *         <p/>
   *         When the number of pages sent and recieved is the same, we may be done with the whole job, or we may just
   *         be waiting for FitNesse to send the next page. There's no way to know until FitNesse either calls send, or
   *         done.
   */
  private boolean stateIndeterminate() {
    return (received == sent) && !isDoneSending;
  }

  private void shortSleep() {
    try {
      Thread.sleep(10);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
  }

  private class FitListeningRunnable implements Runnable {
    public void run() {
      listenToFit();
    }


    private void listenToFit() {
      try {
        attemptToListenToFit();
      } catch (Exception e) {
        exceptionOccurred(e);
      }
    }

    private void attemptToListenToFit() throws IOException {
      while (!finishedReading()) {
        int size;
        size = FitProtocol.readSize(fitOutput);
        if (size != 0) {
          String readValue = fitOutput.read(size);
          if (fitOutput.byteCount() < size)
            throw new IOException("I was expecting " + size + " bytes but I only got " + fitOutput.byteCount());
          listener.testOutputChunk(readValue);
        } else {
          Counts counts = FitProtocol.readCounts(fitOutput);
          TestSummary summary = new TestSummary();
          summary.right = counts.right;
          summary.wrong = counts.wrong;
          summary.ignores = counts.ignores;
          summary.exceptions = counts.exceptions;
          listener.testComplete(summary);
          received++;
        }
      }
    }
  }

}
{'fraction_non_alphanumeric': 0.09115314215985357, 'fraction_numerical': 0.0039048200122025626, 'mean_word_length': 3.0755842864246645, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 5, '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': '7649230', 'n_tokens_mistral': 2355, 'n_tokens_neox': 2217, 'n_words': 662}