@Test public void testServedMergedObservablesAddAfterServe() { Observable<Integer> os1 = Observable.range(0, 100); Observable<Integer> os2 = Observable.range(100, 100); ReplaySubject<Observable<Integer>> subject = ReplaySubject.create(); subject.onNext(os1); subject.onNext(os2); PortSelectorWithinRange portSelector = new PortSelectorWithinRange(8000, 9000); int serverPort = portSelector.acquirePort(); RemoteRxServer server = new RemoteRxServer.Builder() .port(serverPort) .addObservable(new ServeObservable.Builder<Integer>() .encoder(Codecs.integer()) .observable(Observable.merge(subject)) .build()) .build(); server.start(); Observable<Integer> os3 = Observable.range(200, 101); subject.onNext(os3); subject.onCompleted(); Observable<Integer> oc = RemoteObservable.connect("localhost", serverPort, Codecs.integer()); MathObservable.sumInteger(oc).toBlocking().forEach(new Action1<Integer>() { @Override public void call(Integer t1) { Assert.assertEquals(45150, t1.intValue()); } }); }
public static <T> RemoteRxConnection<T> connect(final ConnectToObservable<T> params) { final RxMetrics metrics = new RxMetrics(); return new RemoteRxConnection<T>(Observable.create(new OnSubscribe<T>() { @Override public void call(Subscriber<? super T> subscriber) { RemoteUnsubscribe remoteUnsubscribe = new RemoteUnsubscribe(params.getName()); subscriber.add(remoteUnsubscribe); createTcpConnectionToServer(params, remoteUnsubscribe, metrics, params.getConnectionDisconnectCallback(), params.getCloseTrigger()) .subscribe(subscriber); } }), metrics, params.getCloseTrigger()); }
RemoteObservable { public static <T> RemoteRxConnection<T> connect(final ConnectToObservable<T> params) { final RxMetrics metrics = new RxMetrics(); return new RemoteRxConnection<T>(Observable.create(new OnSubscribe<T>() { @Override public void call(Subscriber<? super T> subscriber) { RemoteUnsubscribe remoteUnsubscribe = new RemoteUnsubscribe(params.getName()); subscriber.add(remoteUnsubscribe); createTcpConnectionToServer(params, remoteUnsubscribe, metrics, params.getConnectionDisconnectCallback(), params.getCloseTrigger()) .subscribe(subscriber); } }), metrics, params.getCloseTrigger()); } }
RemoteObservable { public static <T> RemoteRxConnection<T> connect(final ConnectToObservable<T> params) { final RxMetrics metrics = new RxMetrics(); return new RemoteRxConnection<T>(Observable.create(new OnSubscribe<T>() { @Override public void call(Subscriber<? super T> subscriber) { RemoteUnsubscribe remoteUnsubscribe = new RemoteUnsubscribe(params.getName()); subscriber.add(remoteUnsubscribe); createTcpConnectionToServer(params, remoteUnsubscribe, metrics, params.getConnectionDisconnectCallback(), params.getCloseTrigger()) .subscribe(subscriber); } }), metrics, params.getCloseTrigger()); } private  RemoteObservable(); }
RemoteObservable { public static <T> RemoteRxConnection<T> connect(final ConnectToObservable<T> params) { final RxMetrics metrics = new RxMetrics(); return new RemoteRxConnection<T>(Observable.create(new OnSubscribe<T>() { @Override public void call(Subscriber<? super T> subscriber) { RemoteUnsubscribe remoteUnsubscribe = new RemoteUnsubscribe(params.getName()); subscriber.add(remoteUnsubscribe); createTcpConnectionToServer(params, remoteUnsubscribe, metrics, params.getConnectionDisconnectCallback(), params.getCloseTrigger()) .subscribe(subscriber); } }), metrics, params.getCloseTrigger()); } private  RemoteObservable(); static RemoteRxConnection<T> connect(final ConnectToObservable<T> params); static RemoteRxConnection<GroupedObservable<K, V>> connect(
            final ConnectToGroupedObservable<K, V> config); static RemoteRxConnection<MantisGroup<K, V>> connectToMGO(
            final ConnectToGroupedObservable<K, V> config, final SpscArrayQueue<MantisGroup<?, ?>> inputQueue); static Observable<T> connect(final String host, final int port, final Decoder<T> decoder); static RemoteRxServer serve(int port, final Observable<T> observable, final Encoder<T> encoder); static RemoteRxServer serve(int port, String name, final Observable<T> observable, final Encoder<T> encoder); }
RemoteObservable { public static <T> RemoteRxConnection<T> connect(final ConnectToObservable<T> params) { final RxMetrics metrics = new RxMetrics(); return new RemoteRxConnection<T>(Observable.create(new OnSubscribe<T>() { @Override public void call(Subscriber<? super T> subscriber) { RemoteUnsubscribe remoteUnsubscribe = new RemoteUnsubscribe(params.getName()); subscriber.add(remoteUnsubscribe); createTcpConnectionToServer(params, remoteUnsubscribe, metrics, params.getConnectionDisconnectCallback(), params.getCloseTrigger()) .subscribe(subscriber); } }), metrics, params.getCloseTrigger()); } private  RemoteObservable(); static RemoteRxConnection<T> connect(final ConnectToObservable<T> params); static RemoteRxConnection<GroupedObservable<K, V>> connect(
            final ConnectToGroupedObservable<K, V> config); static RemoteRxConnection<MantisGroup<K, V>> connectToMGO(
            final ConnectToGroupedObservable<K, V> config, final SpscArrayQueue<MantisGroup<?, ?>> inputQueue); static Observable<T> connect(final String host, final int port, final Decoder<T> decoder); static RemoteRxServer serve(int port, final Observable<T> observable, final Encoder<T> encoder); static RemoteRxServer serve(int port, String name, final Observable<T> observable, final Encoder<T> encoder);  }