package com.github.davidmoten.rx2.internal.flowable;

import java.io.IOException;
import java.io.InputStream;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketException;
import java.net.SocketTimeoutException;
import java.util.concurrent.Callable;

import com.github.davidmoten.rx2.Bytes;
import com.github.davidmoten.rx2.Consumers;

import io.reactivex.Emitter;
import io.reactivex.Flowable;
import io.reactivex.functions.Action;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.functions.Predicate;

public final class FlowableServerSocket {

    private FlowableServerSocket() {
        // prevent instantiation
    }

    public static Flowable<Flowable<byte[]>> create(
            final Callable<? extends ServerSocket> serverSocketFactory, final int timeoutMs,
            final int bufferSize, Action preAcceptAction, int acceptTimeoutMs,
            Predicate<? super Socket> acceptSocket) {
        Function<ServerSocket, Flowable<Flowable<byte[]>>> FlowableFactory = createFlowableFactory(
                timeoutMs, bufferSize, preAcceptAction, acceptSocket);
        return Flowable.<Flowable<byte[]>, ServerSocket>using( //
                createServerSocketFactory(serverSocketFactory, acceptTimeoutMs), //
                FlowableFactory, //
                new Consumer<ServerSocket>() {
                    //Note that in java 1.6, ServerSocket does not implement Closeable
                    @Override
                    public void accept(ServerSocket ss) throws Exception {
                        ss.close();
                    }}, //
                true);
    }

    private static Callable<ServerSocket> createServerSocketFactory(
            final Callable<? extends ServerSocket> serverSocketFactory, final int acceptTimeoutMs) {
        return new Callable<ServerSocket>() {
            @Override
            public ServerSocket call() throws Exception {
                return createServerSocket(serverSocketFactory, acceptTimeoutMs);
            }
        };
    }

    private static ServerSocket createServerSocket(
            Callable<? extends ServerSocket> serverSocketCreator, long timeoutMs) throws Exception {
        ServerSocket s = serverSocketCreator.call();
        s.setSoTimeout((int) timeoutMs);
        return s;
    }

    private static Function<ServerSocket, Flowable<Flowable<byte[]>>> createFlowableFactory(
            final int timeoutMs, final int bufferSize, final Action preAcceptAction,
            final Predicate<? super Socket> acceptSocket) {
        return new Function<ServerSocket, Flowable<Flowable<byte[]>>>() {
            @Override
            public Flowable<Flowable<byte[]>> apply(ServerSocket serverSocket) {
                return createServerSocketFlowable(serverSocket, timeoutMs, bufferSize,
                        preAcceptAction, acceptSocket);
            }
        };
    }

    private static Flowable<Flowable<byte[]>> createServerSocketFlowable(
            final ServerSocket serverSocket, final long timeoutMs, final int bufferSize,
            final Action preAcceptAction, final Predicate<? super Socket> acceptSocket) {
        return Flowable.generate( //
                new Consumer<Emitter<Flowable<byte[]>>>() {
                    @Override
                    public void accept(Emitter<Flowable<byte[]>> emitter) throws Exception {
                        acceptConnection(timeoutMs, bufferSize, serverSocket, emitter,
                                preAcceptAction, acceptSocket);
                    }
                });
    }

    private static void acceptConnection(long timeoutMs, int bufferSize, ServerSocket ss,
            Emitter<Flowable<byte[]>> emitter, Action preAcceptAction,
            Predicate<? super Socket> acceptSocket) {
        Socket socket;
        while (true) {
            try {
                preAcceptAction.run();
                socket = ss.accept();
                if (!acceptSocket.test(socket)) {
                    closeQuietly(socket);
                } else {
                    emitter.onNext(createSocketFlowable(socket, timeoutMs, bufferSize));
                    break;
                }
            } catch (SocketTimeoutException e) {
                // timed out so will loop around again
            } catch (Throwable e) {
                // if the server socket has been closed then this is most likely
                // an unsubscribe so we don't try to report an error which would
                // just end up in RxJavaPlugins.onError as a stack trace in the
                // console.
                if (e instanceof SocketException && ("Socket closed".equals(e.getMessage())
                        || "Socket operation on nonsocket: configureBlocking"
                                .equals(e.getMessage()))) {
                    break;
                } else {
                    // unknown problem
                    emitter.onError(e);
                    break;
                }
            }
        }
    }

    // visible for testing
    static void closeQuietly(Socket socket) {
        try {
            socket.close();
        } catch (IOException e) {
            // ignore exception
        }
    }

    private static Flowable<byte[]> createSocketFlowable(final Socket socket, long timeoutMs,
            final int bufferSize) {
        setTimeout(socket, timeoutMs);
        return Flowable.using( //
                new Callable<InputStream>() {
                    @Override
                    public InputStream call() throws Exception {
                        return socket.getInputStream();
                    }
                }, //
                new Function<InputStream, Flowable<byte[]>>() {
                    @Override
                    public Flowable<byte[]> apply(InputStream is) {
                        return Bytes.from(is, bufferSize);
                    }
                }, //
                Consumers.close(), //
                true);
    }

    private static void setTimeout(Socket socket, long timeoutMs) {
        try {
            socket.setSoTimeout((int) timeoutMs);
        } catch (SocketException e) {
            throw new RuntimeException(e);
        }
    }

}

@@ -1231,16 +1231,17 @@
 rSocket%3E
+ 
 using( /
@@ -1235,32 +1235,32 @@
 ket%3E using( //%0D%0A
-
                 
@@ -1435,16 +1435,17 @@
       //
+ 
 Note tha
@@ -1490,16 +1490,40 @@
 mplement
+%0D%0A                    //
  Closeab
@@ -1662,32 +1662,32 @@
    ss.close();%0D%0A
-
                 
@@ -1691,16 +1691,34 @@
        %7D
+%0D%0A                
 %7D, //%0D%0A 

fix 1.6 compilation error