@@ -0,0 +1,1962 @@
+package au.gov.amsa.ais.rx;%0A%0Aimport java.nio.charset.StandardCharsets;%0Aimport java.util.Arrays;%0Aimport java.util.List;%0A%0Aimport rx.Observable;%0Aimport rx.functions.Func1;%0Aimport rx.schedulers.Schedulers;%0Aimport rx.subjects.PublishSubject;%0Aimport au.gov.amsa.streams.HostPort;%0Aimport au.gov.amsa.streams.StringSockets;%0Aimport au.gov.amsa.streams.Strings;%0A%0Apublic class AisServers %7B%0A%0A    // used to inject a STOP sentinel message into the stream so that we can%0A    // stop the webapp at will%0A    private final PublishSubject%3CString%3E stopper = PublishSubject.create();%0A%0A    private final List%3CHostPort%3E hostPorts;%0A%0A    private static final String STOP_SENTINEL = %22STOP_SENTINEL%22;%0A%0A    public AisServers(List%3CHostPort%3E hostPorts) %7B%0A        this.hostPorts = hostPorts;%0A    %7D%0A%0A    public Observable%3CString%3E nmea() %7B%0A        Observable%3CObservable%3CString%3E%3E ais = Observable.from(hostPorts)%0A        //%0A                .map(hp -%3E StringSockets%0A                        .from(hp.getHost(), hp.getPort(), hp.getQuietTimeoutMs(),%0A                                hp.getReconnectDelayMs(), StandardCharsets.UTF_8)%0A                        // split by new line character%0A                        .compose(o -%3E Strings.split(o, %22%5Cn%22))%0A                        // make asynchronous%0A                        .subscribeOn(Schedulers.io()));%0A%0A        return Observable.merge(stopper.serialize().nest().concatWith(ais))%0A        // check each line to see if we wish to stop%0A                .takeWhile(isNotStopSentinel());%0A%0A    %7D%0A%0A    private static Func1%3CString, Boolean%3E isNotStopSentinel() %7B%0A        return t -%3E !STOP_SENTINEL.equals(t);%0A    %7D%0A%0A    public void stop() %7B%0A        stopper.onNext(STOP_SENTINEL);%0A    %7D%0A%0A    public static void main(String%5B%5D args) %7B%0A        List%3CHostPort%3E hostPorts = Arrays.asList(HostPort.create(%22sarapps%22, 9010, 1000, 1000),%0A                HostPort.create(%22sarapps%22, 9100, 1000, 1000));%0A        new AisServers(hostPorts).nmea().toBlocking().last();%0A    %7D%0A%7D%0A

add AisServers to properly merge two sources