<commit_before>package io.cloudracer.mocktcpserver;

import static org.junit.Assert.assertThat;

import java.io.BufferedReader;
import java.io.Closeable;
import java.io.DataOutputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketException;
import java.util.Arrays;

import org.apache.commons.io.IOUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.log4j.Logger;

import io.cloudracer.datastream.DataStream;
import io.cloudracer.datastream.DataStreamRegexMatcher;

/**
 * A mock server used for testing purposes only.
 *
 * @author <NAME>
 */
public class MockTCPServer extends Thread implements Closeable {

    private Logger logger = Logger.getLogger(getRootLoggerName());

    public final static byte[] DEFAULT_TERMINATOR = { 13, 10, 10 };
    public final static byte[] DEFAULT_ACK = { 65 };
    public final static byte[] DEFAULT_NAK = { 78 };

    private byte[] terminator = null;
    private byte[] ack = null;
    private byte[] nak = null;

    private AssertionError assertionError;

    private ServerSocket socket;
    private BufferedReader inputStream;
    private DataOutputStream outputStream;
    private DataStreamRegexMatcher expectedMessage;

    private DataStream dataStream;
    private Socket connectionSocket;

    private int port;
    private boolean setIsAlwaysNAKResponse = false;
    private boolean setIsAlwaysNoResponse = false;
    private boolean isCloseAfterNextResponse = false;
    private int messagesReceivedCount = 0;

    private boolean isClosed = false;

    public MockTCPServer(final int port) {
        logger.info("Starting...");

        super.setName(String.format("%s-%d", getThreadName(), port));

        setPort(port);
        start();
    }

    @Override
    public void run() {
        try {
            while (!isClosed() && (getSocket() != null)) {
                try {
                    setDataStream(null);
                    while ((getDataStream().write(getInputStream().read())) != -1) {
                        if (Arrays.equals(getDataStream().getTail(), getTerminator())) {
                            incrementMessagesReceivedCount();

                            break;
                        }
                    }
                    // Ignore null in order allow a probing ping e.g. paping.exe
                    if (getDataStream().size() > 0) {
                        setAssertionError(null);
                        try {
                            if (getExpectedMessage() != null) {
                                assertThat("Unexpected message from the AM Host Client.", getDataStream(), getExpectedMessage());
                            }
                        } catch (AssertionError e) {
                            setAssertionError(e);
                        }
                        onMessage(getDataStream());
                        // If the stream has not ended and a response is required, send one.
                        if (getDataStream().getLastByte() != -1 && !getIsAlwaysNoResponse()) {
                            byte[] response = null;

                            if (getAssertionError() == null && !getIsAlwaysNAKResponse()) {
                                response = getACK();
                            } else {
                                response = getNAK();
                            }

                            getOutputStream().write(response);

                            afterResponse(response);
                        }
                    }
                } catch (SocketException e) {
                    handleException(e);
                } catch (IOException e) {
                    handleException(e);
                } catch (Exception e) {
                    handleException(e);
                }
            }
        } catch (SocketException e) {
            handleException(e);
        } catch (IOException e) {
            handleException(e);
        } catch (Exception e) {
            handleException(e);
        } finally {
            close();
        }
    }

    private void handleException(Exception e) {
        if (e.getMessage().toLowerCase().equals("socket closed")) {
            logger.warn(e.getMessage());
        } else if (e.getMessage().toLowerCase().equals("socket input is shutdown")) {
            logger.warn(e.getMessage());
        } else if (e.getMessage().toLowerCase().equals("socket is closed")) {
            logger.warn(e.getMessage());
        } else if (e.getMessage().toLowerCase().equals("stream closed")) {
            logger.warn(e.getMessage());
        } else {
            logger.error(e.getMessage(), e);
        }
    }

    /**
     * The server will read the stream until these characters are encountered.
     *
     * @return the terminator.
     */
    public byte[] getTerminator() {
        if (terminator == null) {<commit_msg>Improved error handling.<commit_after>package io.cloudracer.mocktcpserver;

import static org.junit.Assert.assertThat;

import java.io.BufferedReader;
import java.io.Closeable;
import java.io.DataOutputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketException;
import java.util.Arrays;

import org.apache.commons.io.IOUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.log4j.Logger;

import io.cloudracer.datastream.DataStream;
import io.cloudracer.datastream.DataStreamRegexMatcher;

/**
 * A mock server used for testing purposes only.
 *
 * @author <NAME>
 */
public class MockTCPServer extends Thread implements Closeable {

    private Logger logger = Logger.getLogger(getRootLoggerName());

    public final static byte[] DEFAULT_TERMINATOR = { 13, 10, 10 };
    public final static byte[] DEFAULT_ACK = { 65 };
    public final static byte[] DEFAULT_NAK = { 78 };

    private byte[] terminator = null;
    private byte[] ack = null;
    private byte[] nak = null;

    private AssertionError assertionError;

    private ServerSocket socket;
    private BufferedReader inputStream;
    private DataOutputStream outputStream;
    private DataStreamRegexMatcher expectedMessage;

    private DataStream dataStream;
    private Socket connectionSocket;

    private int port;
    private boolean setIsAlwaysNAKResponse = false;
    private boolean setIsAlwaysNoResponse = false;
    private boolean isCloseAfterNextResponse = false;
    private int messagesReceivedCount = 0;

    private boolean isClosed = false;

    public MockTCPServer(final int port) {
        logger.info("Starting...");

        super.setName(String.format("%s-%d", getThreadName(), port));

        setPort(port);
        start();
    }

    @Override
    public void run() {
        try {
            while (!isClosed() && (getSocket() != null)) {
                try {
                    setDataStream(null);
                    while ((getDataStream().write(getInputStream().read())) != -1) {
                        if (Arrays.equals(getDataStream().getTail(), getTerminator())) {
                            incrementMessagesReceivedCount();

                            break;
                        }
                    }
                    // Ignore null in order allow a probing ping e.g. paping.exe
                    if (getDataStream().size() > 0) {
                        setAssertionError(null);
                        try {
                            if (getExpectedMessage() != null) {
                                assertThat("Unexpected message from the AM Host Client.", getDataStream(), getExpectedMessage());
                            }
                        } catch (AssertionError e) {
                            setAssertionError(e);
                        }
                        onMessage(getDataStream());
                        // If the stream has not ended and a response is required, send one.
                        if (getDataStream().getLastByte() != -1 && !getIsAlwaysNoResponse()) {
                            byte[] response = null;

                            if (getAssertionError() == null && !getIsAlwaysNAKResponse()) {
                                response = getACK();
                            } else {
                                response = getNAK();
                            }

                            getOutputStream().write(response);

                            afterResponse(response);
                        }
                    }
                } catch (SocketException e) {
                    handleException(e);
                } catch (IOException e) {
                    handleException(e);
                } catch (Exception e) {
                    handleException(e);
                }
            }
        } catch (SocketException e) {
            handleException(e);
        } catch (IOException e) {
            handleException(e);
        } catch (Exception e) {
            handleException(e);
        } finally {
            close();
        }
    }

    private void handleException(Exception e) {
        if (e.getMessage().toLowerCase().equals("socket closed")) {
            logger.warn(e.getMessage());
        } else if (e.getMessage().toLowerCase().equals("socket input is shutdown")) {
            logger.warn(e.getMessage());
        } else if (e.getMessage().toLowerCase().equals("socket output is shutdown")) {
            logger.warn(e.getMessage());
        } else if (e.getMessage().toLowerCase().equals("socket is closed")) {
            logger.warn(e.getMessage());
        } else if (e.getMessage().toLowerCase().equals("stream closed")) {
            logger.warn(e.getMessage());
        } else {
            logger.error(e.getMessage(), e);
        }
    }

    /**
     * The server will read the stream until these characters are encountered.
     *
     * @return the terminator.
     */
    public byte[] getTerminator() {
        if (terminator == null) {
{'fraction_non_alphanumeric': 0.10353761833582462, 'fraction_numerical': 0.003188839063278525, 'mean_word_length': 1.7197831978319784, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 7, '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': '2499270', 'n_tokens_mistral': 2480, 'n_tokens_neox': 2405, 'n_words': 603}