@Test public void testServeTwoObservablesOnSamePort() throws InterruptedException { Observable<Integer> os1 = Observable.range(0, 101); Observable<String> os2 = Observable.from(new String[] {"a", "b", "c", "d", "e"}); PortSelectorWithinRange portSelector = new PortSelectorWithinRange(8000, 9000); int serverPort = portSelector.acquirePort(); RemoteRxServer server = new RemoteRxServer.Builder() .port(serverPort) .addObservable(new ServeObservable.Builder<Integer>() .name("ints") .encoder(Codecs.integer()) .observable(os1) .build()) .addObservable(new ServeObservable.Builder<String>() .name("strings") .encoder(Codecs.string()) .observable(os2) .build()) .build(); server.start(); Observable<Integer> ro1 = RemoteObservable.connect(new ConnectToObservable.Builder<Integer>() .host("localhost") .port(serverPort) .name("ints") .decoder(Codecs.integer()) .build()) .getObservable(); Observable<String> ro2 = RemoteObservable.connect(new ConnectToObservable.Builder<String>() .host("localhost") .port(serverPort) .name("strings") .decoder(Codecs.string()) .build()) .getObservable(); MathObservable.sumInteger(ro1).toBlocking().forEach(new Action1<Integer>() { @Override public void call(Integer t1) { Assert.assertEquals(5050, t1.intValue()); } }); ro2.reduce(new Func2<String, String, String>() { @Override public String call(String t1, String t2) { return t1 + t2; } }).toBlocking().forEach(new Action1<String>() { @Override public void call(String t1) { Assert.assertEquals("abcde", t1); } }); }
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);  }