package com.github.davidmoten.rx.internal.operators;

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

import java.io.Closeable;
import java.io.IOException;
import java.io.OutputStream;
import java.net.BindException;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.UnknownHostException;
import java.nio.charset.StandardCharsets;
import java.util.HashSet;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ConcurrentSkipListSet;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;

import org.junit.Assert;
import org.junit.Test;

import com.github.davidmoten.junit.Asserts;
import com.github.davidmoten.rx.Actions;
import com.github.davidmoten.rx.Bytes;
import com.github.davidmoten.rx.IO;

import rx.AsyncEmitter.BackpressureMode;
import rx.Observable;
import rx.observers.TestSubscriber;
import rx.schedulers.Schedulers;

public final class ObservableServerSocketTest {

    private static final int PORT = 12345;
    private static final String TEXT = "hello there";

    @Test
    public void serverSocketReadsTcpPushWhenBufferIsSmallerThanInput()
            throws UnknownHostException, IOException, InterruptedException {
        checkServerSocketReadsTcpPushWhenBufferSizeIs(TEXT, 4);
    }

    @Test
    public void serverSocketReadsTcpPushWhenBufferIsBiggerThanInput()
            throws UnknownHostException, IOException, InterruptedException {
        checkServerSocketReadsTcpPushWhenBufferSizeIs(TEXT, 8192);
    }

    @Test
    public void serverSocketReadsTcpPushWhenBufferIsSameSizeAsInput()
            throws UnknownHostException, IOException, InterruptedException {
        checkServerSocketReadsTcpPushWhenBufferSizeIs(TEXT, TEXT.length());
    }

    @Test
    public void serverSocketReadsTcpPushWhenInputIsEmpty()
            throws UnknownHostException, IOException, InterruptedException {
        checkServerSocketReadsTcpPushWhenBufferSizeIs("", 4);
    }

    @Test
    public void serverSocketReadsTcpPushWhenInputIsOneCharacter()
            throws UnknownHostException, IOException, InterruptedException {
        checkServerSocketReadsTcpPushWhenBufferSizeIs("a", 4);
    }

    @Test
    public void errorEmittedIfServerSocketBusy() throws IOException {

        TestSubscriber<Object> ts = TestSubscriber.create();
        try (ServerSocket socket = new ServerSocket(PORT)) {
            IO.serverSocket(PORT, 10, TimeUnit.SECONDS, 5, BackpressureMode.BUFFER).subscribe(ts);
            ts.assertNoValues();
            ts.assertNotCompleted();
            ts.assertTerminalEvent();
            assertTrue(ts.getOnErrorEvents().get(0).getCause() instanceof BindException);
        }
    }

    @Test
    public void isUtilityClass() {
        Asserts.assertIsUtilityClass(ObservableServerSocket.class);
    }

    @Test
    public void isUtilityClassIO() {
        Asserts.assertIsUtilityClass(IO.class);
    }

    @Test
    public void testCloserWhenDoesNotThrow() {
        AtomicBoolean called = new AtomicBoolean();
        Closeable c = new Closeable() {

            @Override
            public void close() throws IOException {
                called.set(true);
            }
        };
        ObservableServerSocket.closer().call(c);
        assertTrue(called.get());
    }

    @Test
    public void testCloserWhenThrows() {
        IOException ex = new IOException();
        Closeable c = new Closeable() {

            @Override
            public void close() throws IOException {
                throw ex;
            }
        };
        try {
            ObservableServerSocket.closer().call(c);
            Assert.fail();
        } catch (RuntimeException e) {
            assertTrue(ex == e.getCause());
        }
    }

    @Test
    public void testAsynchronousDelivery() throws InterruptedException {
        TestSubscriber<String> ts = TestSubscriber.create();
        try {
            int bufferSize = 4;
            IO.serverSocket(PORT, 10, TimeUnit.SECONDS, bufferSize, BackpressureMode.BUFFER) //
                    .flatMap(g -> g //
                            .compose(Bytes.collect()) //
                            .onErrorResumeNext(Observable.empty()), 1) //
                    .map(bytes -> new String(bytes, StandardCharsets.UTF_8)) //
                    .doOnNext(System.out::println) //
                    .doOnError(e -> e.printStackTrace()) //
                    .subscribe(ts);
            TestSubscriber<Object> ts2 = TestSubscriber.create();
            Set<String> messages = new ConcurrentSkipListSet<>();
            int messageBlocks = 10;
            int numMessages = 1000;
            Observable.range(1, numMessages).flatMap(n -> {
                return Observable.defer(() -> {
                    System.out.println(Thread.currentThread().getName() + " - writing message");
                    String id = UUID.randomUUID().toString();
                    StringBuilder s = new StringBuilder();
                    for (int i = 0; i < messageBlocks; i++) {
                        s.append(id);
                    }
                    messages.add(s.toString());
                    try {
                        Socket socket = new Socket("localhost", PORT);
                        OutputStream out = socket.getOutputStream();
                        for (int i = 0; i < messageBlocks; i++) {
                            out.write(id.getBytes(StandardCharsets.UTF_8));
                        }
                        out.close();
                        socket.close();
                    } catch (Exception e) {
                        throw new RuntimeException(e);
                    }
                    return Observable.just(1);
                }).subscribeOn(Schedulers.computation());
            } , 1).subscribe(ts2);
            ts2.awaitTerminalEvent();
            ts2.assertCompleted();
            // allow server to complete processing
            Thread.sleep(1000);
            assertEquals(messages, new HashSet<String>(ts.getOnNextEvents()));
        } finally {
            ts.unsubscribe();
        }
    }

    private void checkServerSocketReadsTcpPushWhenBufferSizeIs(String text, int bufferSize)
            throws UnknownHostException, IOException, InterruptedException {
        TestSubscriber<Object> ts = TestSubscriber.create();
        AtomicReference<byte[]> result = new AtomicReference<byte[]>();
        try {
            IO.serverSocket(PORT, 10, TimeUnit.SECONDS, bufferSize, BackpressureMode.BUFFER) //
                    .flatMap(g -> g //
                            .compose(Bytes.collect()) //
                            .doOnNext(Actions.setAtomic(result)) //
                            .doOnNext(bytes -> System.out.println(
                                    Thread.currentThread().getName() + ": " + new String(bytes))) //
                            .onErrorResumeNext(Observable.empty()))
                    .subscribe(ts);
            Socket socket = new Socket("localhost", PORT);
            OutputStream out = socket.getOutputStream();
            out.write(text.getBytes());
            out.close();
            socket.close();
            Thread.sleep(1000);
            assertEquals(text, new String(result.get(), StandardCharsets.UTF_8));
        } finally {
            // will close server socket
            ts.unsubscribe();
        }
    }

}

@@ -502,24 +502,63 @@
 kipListSet;%0A
+import java.util.concurrent.Executors;%0A
 import java.
@@ -955,24 +955,45 @@
 Observable;%0A
+import rx.Scheduler;%0A
 import rx.ob
@@ -4718,24 +4718,37 @@
 istSet%3C%3E();%0A
+            %0A
             
@@ -4807,16 +4807,135 @@
 = 1000;%0A
+            %0A            //sender%0A            Scheduler scheduler = Schedulers.from(Executors.newFixedThreadPool(50));%0A
         
@@ -6032,25 +6032,25 @@
 cribeOn(
-S
+s
 cheduler
 s.comput
@@ -6045,23 +6045,8 @@
 uler
-s.computation()
 );%0A 

use 50 threads for wrting