@Test public void dynamicConnectionTest() throws InterruptedException { PortSelectorWithinRange portSelector = new PortSelectorWithinRange(8000, 9000); final int serverPort1 = portSelector.acquirePort(); final int serverPort2 = portSelector.acquirePort(); RemoteRxServer server1 = RemoteObservable.serve(serverPort1, Observable.range(1, 100), Codecs.integer()); server1.start(); RemoteRxServer server2 = RemoteObservable.serve(serverPort2, Observable.range(101, 100), Codecs.integer()); server2.start(); Observable<Endpoint> endpointSwitch = Observable.create(new OnSubscribe<Endpoint>() { @Override public void call(Subscriber<? super Endpoint> subscriber) { subscriber.onNext(new Endpoint("localhost", serverPort1)); subscriber.onNext(new Endpoint("localhost", serverPort2)); } }); ConnectToObservable.Builder<Integer> config = new ConnectToObservable.Builder<Integer>() .decoder(Codecs.integer()); DynamicConnection<Integer> connection = DynamicConnection.create(config, endpointSwitch); MathObservable.sumInteger(connection.observable()) .last() .subscribe(new Action1<Integer>() { @Override public void call(Integer t1) { Assert.assertEquals(20100, t1.intValue()); } }); Thread.sleep(1000); connection.close(); }
DynamicConnection(Func1<Endpoint, Observable<T>> toObservableFunc, Observable<Endpoint> changeEndpointObservable) { this.changeEndpointObservable = changeEndpointObservable; this.toObservableFunc = toObservableFunc; }
DynamicConnection { DynamicConnection(Func1<Endpoint, Observable<T>> toObservableFunc, Observable<Endpoint> changeEndpointObservable) { this.changeEndpointObservable = changeEndpointObservable; this.toObservableFunc = toObservableFunc; } }
DynamicConnection { DynamicConnection(Func1<Endpoint, Observable<T>> toObservableFunc, Observable<Endpoint> changeEndpointObservable) { this.changeEndpointObservable = changeEndpointObservable; this.toObservableFunc = toObservableFunc; } DynamicConnection(Func1<Endpoint, Observable<T>> toObservableFunc, Observable<Endpoint> changeEndpointObservable); }
DynamicConnection { DynamicConnection(Func1<Endpoint, Observable<T>> toObservableFunc, Observable<Endpoint> changeEndpointObservable) { this.changeEndpointObservable = changeEndpointObservable; this.toObservableFunc = toObservableFunc; } DynamicConnection(Func1<Endpoint, Observable<T>> toObservableFunc, Observable<Endpoint> changeEndpointObservable); static DynamicConnection<GroupedObservable<K, V>> create(
            final ConnectToGroupedObservable.Builder<K, V> config, Observable<Endpoint> endpoints); static DynamicConnection<T> create(
            final ConnectToObservable.Builder<T> config, Observable<Endpoint> endpoints); void close(); Observable<T> observable(); }
DynamicConnection { DynamicConnection(Func1<Endpoint, Observable<T>> toObservableFunc, Observable<Endpoint> changeEndpointObservable) { this.changeEndpointObservable = changeEndpointObservable; this.toObservableFunc = toObservableFunc; } DynamicConnection(Func1<Endpoint, Observable<T>> toObservableFunc, Observable<Endpoint> changeEndpointObservable); static DynamicConnection<GroupedObservable<K, V>> create(
            final ConnectToGroupedObservable.Builder<K, V> config, Observable<Endpoint> endpoints); static DynamicConnection<T> create(
            final ConnectToObservable.Builder<T> config, Observable<Endpoint> endpoints); void close(); Observable<T> observable();  }