@Test public void testOnCompletedFromReplaySubject() { PublishSubject<Integer> subject = PublishSubject.create(); subject.onNext(1); subject.onNext(2); subject.onNext(3); subject.onCompleted(); 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(subject) .serverSideFilter(ServerSideFilters.oddsAndEvens()) .build()) .build(); server.start(); Observable<Integer> ro = RemoteObservable.connect("localhost", serverPort, Codecs.integer()); final MutableReference<Boolean> completed = new MutableReference<Boolean>(); ro.materialize().toBlocking().forEach(new Action1<Notification<Integer>>() { @Override public void call(Notification<Integer> notification) { if (notification.isOnCompleted()) { completed.setValue(true); } } }); Assert.assertEquals(true, completed.getValue()); }
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);  }