// Path: src/main/java/com/github/davidmoten/rx2/internal/flowable/FlowableServerSocket.java
// 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;
//                 }
//             }
//         }
//     }
// 
//     @VisibleForTesting
//     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);
//         }
//     }
// 
// }


import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.concurrent.Callable;
import com.github.davidmoten.rx2.internal.flowable.FlowableServerSocket;
import io.reactivex.Flowable;
import io.reactivex.functions.Action;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Predicate;
import io.reactivex.internal.functions.Functions;
        public ServerSocketBuilder(final Callable<? extends ServerSocket> serverSocketFactory) {
            this.serverSocketFactory = serverSocketFactory;
        }

        public ServerSocketBuilder readTimeoutMs(int readTimeoutMs) {
            this.readTimeoutMs = readTimeoutMs;
            return this;
        }

        public ServerSocketBuilder bufferSize(int bufferSize) {
            this.bufferSize = bufferSize;
            return this;
        }

        public ServerSocketBuilder preAcceptAction(Action action) {
            this.preAcceptAction = action;
            return this;
        }

        public ServerSocketBuilder acceptTimeoutMs(int acceptTimeoutMs) {
            this.acceptTimeoutMs = acceptTimeoutMs;
            return this;
        }

        public ServerSocketBuilder acceptSocketIf(Predicate<? super Socket> acceptSocket) {
            this.acceptSocket = acceptSocket;
            return this;
        }

        public Flowable<Flowable<byte[]>> create() {
// Path: src/main/java/com/github/davidmoten/rx2/internal/flowable/FlowableServerSocket.java
// 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;
//                 }
//             }
//         }
//     }
// 
//     @VisibleForTesting
//     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);
//         }
//     }
// 
// }


// Path: src/main/java/com/github/davidmoten/rx2/IO.java
import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.concurrent.Callable;
import com.github.davidmoten.rx2.internal.flowable.FlowableServerSocket;
import io.reactivex.Flowable;
import io.reactivex.functions.Action;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Predicate;
import io.reactivex.internal.functions.Functions;

        public ServerSocketBuilder(final Callable<? extends ServerSocket> serverSocketFactory) {
            this.serverSocketFactory = serverSocketFactory;
        }

        public ServerSocketBuilder readTimeoutMs(int readTimeoutMs) {
            this.readTimeoutMs = readTimeoutMs;
            return this;
        }

        public ServerSocketBuilder bufferSize(int bufferSize) {
            this.bufferSize = bufferSize;
            return this;
        }

        public ServerSocketBuilder preAcceptAction(Action action) {
            this.preAcceptAction = action;
            return this;
        }

        public ServerSocketBuilder acceptTimeoutMs(int acceptTimeoutMs) {
            this.acceptTimeoutMs = acceptTimeoutMs;
            return this;
        }

        public ServerSocketBuilder acceptSocketIf(Predicate<? super Socket> acceptSocket) {
            this.acceptSocket = acceptSocket;
            return this;
        }

        public Flowable<Flowable<byte[]>> create() {
            return FlowableServerSocket.create(serverSocketFactory, readTimeoutMs, bufferSize, preAcceptAction,