<commit_before>package com.github.davidmoten.rx.internal.operators;

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 com.github.davidmoten.rx.Actions;
import com.github.davidmoten.rx.Bytes;
import com.github.davidmoten.rx.Checked;
import com.github.davidmoten.rx.Checked.F0;
import com.github.davidmoten.rx.Functions;

import rx.Observable;
import rx.Observer;
import rx.functions.Action2;
import rx.functions.Func0;
import rx.functions.Func1;
import rx.observables.SyncOnSubscribe;

public final class ObservableServerSocket {

    private ObservableServerSocket() {
        // prevent instantiation
    }

    public static Observable<Observable<byte[]>> create(final int port, final int timeoutMs,
            final int bufferSize) {
        return Observable.using( //
                createServerSocketFactory(port, timeoutMs), //
                new Func1<ServerSocket, Observable<Observable<byte[]>>>() {
                    @Override
                    public Observable<Observable<byte[]>> call(ServerSocket serverSocket) {
                        return createServerSocketObservable(serverSocket, timeoutMs, bufferSize);
                    }
                }, //
                Actions.close(), //
                true);
    }

    private static Func0<ServerSocket> createServerSocketFactory(final int port,
            final int timeoutMs) {
        return Checked.f0(new F0<ServerSocket>() {
            @Override
            public ServerSocket call() throws Exception {
                return createServerSocket(port, timeoutMs);
            }
        });
    }

    private static ServerSocket createServerSocket(int port, long timeoutMs) throws IOException {
        ServerSocket s = new ServerSocket(port);
        s.setSoTimeout((int) timeoutMs);
        return s;
    }

    private static Observable<Observable<byte[]>> createServerSocketObservable(
            ServerSocket serverSocket, final long timeoutMs, final int bufferSize) {
        return Observable.create( //
                SyncOnSubscribe.<ServerSocket, Observable<byte[]>> createSingleState( //
                        Functions.constant0(serverSocket), //
                        new Action2<ServerSocket, Observer<? super Observable<byte[]>>>() {

                            @Override
                            public void call(ServerSocket ss,
                                    Observer<? super Observable<byte[]>> observer) {
                                acceptConnection(timeoutMs, bufferSize, ss, observer);
                            }}
                        ));
    }

    private static void acceptConnection(long timeoutMs, int bufferSize, ServerSocket ss,
            Observer<? super Observable<byte[]>> observer) {
        Socket socket;
        while (true) {
            try {
                socket = ss.accept();
                observer.onNext(createSocketObservable(socket, timeoutMs, bufferSize));
                break;
            } catch (SocketTimeoutException e) {
                // timed out so will continue waiting
            } catch (IOException e) {
                // unknown problem
                observer.onError(e);
                break;
            }
        }
    }
<commit_msg>fix generics so compile in java 6<commit_after>package com.github.davidmoten.rx.internal.operators;

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 com.github.davidmoten.rx.Actions;
import com.github.davidmoten.rx.Bytes;
import com.github.davidmoten.rx.Checked;
import com.github.davidmoten.rx.Checked.F0;
import com.github.davidmoten.rx.Functions;

import rx.Observable;
import rx.Observer;
import rx.functions.Action2;
import rx.functions.Func0;
import rx.functions.Func1;
import rx.observables.SyncOnSubscribe;

public final class ObservableServerSocket {

    private ObservableServerSocket() {
        // prevent instantiation
    }

    public static Observable<Observable<byte[]>> create(final int port, final int timeoutMs,
            final int bufferSize) {
        Func1<ServerSocket, Observable<? extends Observable<byte[]>>> observableFactory = createObservableFactory(
                timeoutMs, bufferSize);
        return Observable.using( //
                createServerSocketFactory(port, timeoutMs), //
                observableFactory, //
                Actions.close(), //
                true);
    }

    private static Func0<ServerSocket> createServerSocketFactory(final int port,
            final int timeoutMs) {
        return Checked.f0(new F0<ServerSocket>() {
            @Override
            public ServerSocket call() throws Exception {
                return createServerSocket(port, timeoutMs);
            }
        });
    }

    private static ServerSocket createServerSocket(int port, long timeoutMs) throws IOException {
        ServerSocket s = new ServerSocket(port);
        s.setSoTimeout((int) timeoutMs);
        return s;
    }

    private static Func1<ServerSocket, Observable<? extends Observable<byte[]>>> createObservableFactory(
            final int timeoutMs, final int bufferSize) {
        return new Func1<ServerSocket, Observable<? extends Observable<byte[]>>>() {
            @Override
            public Observable<Observable<byte[]>> call(ServerSocket serverSocket) {
                return createServerSocketObservable(serverSocket, timeoutMs, bufferSize);
            }
        };
    }

    private static Observable<Observable<byte[]>> createServerSocketObservable(
            ServerSocket serverSocket, final long timeoutMs, final int bufferSize) {
        return Observable.create( //
                SyncOnSubscribe.<ServerSocket, Observable<byte[]>> createSingleState( //
                        Functions.constant0(serverSocket), //
                        new Action2<ServerSocket, Observer<? super Observable<byte[]>>>() {

                            @Override
                            public void call(ServerSocket ss,
                                    Observer<? super Observable<byte[]>> observer) {
                                acceptConnection(timeoutMs, bufferSize, ss, observer);
                            }
                        }));
    }

    private static void acceptConnection(long timeoutMs, int bufferSize, ServerSocket ss,
            Observer<? super Observable<byte[]>> observer) {
        Socket socket;
        while (true) {
            try {
                socket = ss.accept();
                observer.onNext(createSocketObservable(socket, timeoutMs, bufferSize));
                break;
            } catch (SocketTimeoutException e) {
                // timed out so will continue waiting
            } catch (IOException e) {
                // unknown problem
                observer.onError(e);
                break;
            }
        }
    }

{'fraction_non_alphanumeric': 0.08958598276105695, 'fraction_numerical': 0.0032499646742970185, 'mean_word_length': 2.088132635253054, 'pattern_counts': {'":': 0, '<': 49, '<?xml version=': 0, '>': 49, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4944346', 'n_tokens_mistral': 1746, 'n_tokens_neox': 1665, 'n_words': 407}