package com.timgroup.io;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.ConnectException;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.Random;
import java.util.concurrent.Callable;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TestRule;
import org.junit.rules.Timeout;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.fail;

public class ReconnectingSocketOutputStreamTest {

    private static final Random RANDOM = new Random();

    @Rule
    public final TestRule timeout = new Timeout(5000);

    private final int port = 1024 + RANDOM.nextInt(65536 - 1024);
    private ServerSocket serverSocket;

    @Before
    public void closeServerSocket() throws IOException {
        if (serverSocket != null) {
            serverSocket.close();
        }
    }

    @Test
    public void reportsFailureToConnect() throws Exception {
        try {
            new ReconnectingSocketOutputStream("localhost", port);
            fail("should have thrown ConnectException");
        } catch (ConnectException e) {
            // expected
        }
    }

    @Test
    public void writesBytesToASocket() throws Exception {
        serverSocket = new ServerSocket(port);

        Future<String> line = readLine(serverSocket);

        ReconnectingSocketOutputStream out = new ReconnectingSocketOutputStream("localhost", port);
        out.write("hello\n".getBytes());
        out.close();

        assertEquals("hello", line.get(100, TimeUnit.MILLISECONDS));
    }

    @Test
    public void reconnectsIfNecessaryOnWrite() throws Exception {
        serverSocket = new ServerSocket(port);

        Future<Socket> socket = connect(serverSocket);

        ReconnectingSocketOutputStream out = new ReconnectingSocketOutputStream("localhost", port);

        socket.get(100, TimeUnit.MILLISECONDS).close();
        Future<String> line = readLine(serverSocket);

        out.write("hello\n".getBytes());
        out.close();

        assertEquals("hello", line.get(100, TimeUnit.MILLISECONDS));
    }

    @Test
    public void blocksIfNecessaryOnWrite() throws Exception {
        serverSocket = new ServerSocket(port);
        byte[] stuffing = ByteArrayUtils.fill(1024, (byte) '\n');

        Future<String> line = eventuallyReadFirstNonBlankLine(serverSocket, 500);

        ReconnectingSocketOutputStream out = new ReconnectingSocketOutputStream("localhost", port);

        for (int i = 0; i < 320; ++i) {
            out.write(stuffing);
        }
        out.write("hello\n".getBytes());
        out.close();

        assertEquals("hello", line.get(700, TimeUnit.MILLISECONDS));
    }

    @Test
    public void blocksIfNecessaryOnReconnect() throws Exception {
        serverSocket = new ServerSocket(port);

        Future<Socket> socket = connect(serverSocket);

        ReconnectingSocketOutputStream out = new ReconnectingSocketOutputStream("localhost", port);

        socket.get(100, TimeUnit.MILLISECONDS).close();
        serverSocket.close();
        Future<String> line = eventuallyBindAcceptAndReadLine(port, 500);

        out.write("hello\n".getBytes());
        out.close();

        assertEquals("hello", line.get(600, TimeUnit.MILLISECONDS));
    }

    private Future<String> readLine(final ServerSocket serverSocket) {
        return Executors.newSingleThreadExecutor().submit(new Callable<String>() {
            @Override
            public String call() throws Exception {
                Socket socket = serverSocket.accept();
                try {
                    return new BufferedReader(new InputStreamReader(socket.getInputStream())).readLine();
                } finally {
                    socket.close();
                }
            }
        });
    }

    private Future<Socket> connect(final ServerSocket serverSocket) {
        return Executors.newSingleThreadExecutor().submit(new Callable<Socket>() {
            @Override
            public Socket call() throws Exception {
                return serverSocket.accept();
            }
        });
    }

    private Future<String> eventuallyReadFirstNonBlankLine(final ServerSocket serverSocket, final int delay) {
        return Executors.newSingleThreadExecutor().submit(new Callable<String>() {
            @Override
            public String call() throws Exception {
                Socket socket = serverSocket.accept();
                Thread.sleep(delay);
                try {
                    BufferedReader in = new BufferedReader(new InputStreamReader(socket.getInputStream()));
                    String line;
                    while ((line = in.readLine()) != null && line.equals(""));
                    return line;
                } finally {
                    socket.close();
                }
            }
        });
    }

    private Future<String> eventuallyBindAcceptAndReadLine(final int port, final int delay) {
        return Executors.newSingleThreadExecutor().submit(new Callable<String>() {
            @Override
            public String call() throws Exception {
                Thread.sleep(delay);
                serverSocket = new ServerSocket(port);
                Socket socket = serverSocket.accept();
                try {
                    return new BufferedReader(new InputStreamReader(socket.getInputStream())).readLine();
                } finally {
                    socket.close();
                }
            }
        });
    }

}

@@ -762,17 +762,18 @@
 Timeout(
-5
+10
 000);%0A%0A 
@@ -2525,16 +2525,17 @@
 Socket, 
+2
 500);%0A%0A 
@@ -2666,10 +2666,13 @@
 i %3C 
-32
+10000
 0; +
@@ -2822,16 +2822,17 @@
 ine.get(
+2
 700, Tim

Make test a bit more rigorous i guess