@Test @Ignore public void test() { final InstanceManager<Integer> instances = InstanceManager.create(); final LoadBalancer<Client> lb = LoadBalancer .fromSource(instances.map(Client.connector())) .withQuarantiner(Client.incarnationFactory(), Delays.fixed(1, TimeUnit.SECONDS), scheduler) .build(RoundRobinLoadBalancer.<Client>create()); instances.add(1); instances.add(2); List<String> result = Observable .interval(100, TimeUnit.MILLISECONDS) .concatMap(new Func1<Long, Observable<String>>() { @Override public Observable<String> call(final Long counter) { return Observable.just(lb.next()) .concatMap(new Func1<Client, Observable<String>>() { @Override public Observable<String> call(Client instance) { instance.onBegin(); if (1 == instance.getValue()) { instance.onFailed(); return Observable.error(new Exception("Failed")); } instance.onSuccess(); return Observable.just(instance + "-" + counter); } }) .retry(1); } }) .take(10) .toList() .toBlocking() .first() ; }
public static <T> InstanceQuarantiner<T> create( IncarnationFactory<T> factory, DelayStrategy backoffStrategy, Scheduler scheduler) { return new InstanceQuarantiner<T>(factory, backoffStrategy, scheduler); }
InstanceQuarantiner implements Func1<Instance<T>, Observable<Instance<T>>> { public static <T> InstanceQuarantiner<T> create( IncarnationFactory<T> factory, DelayStrategy backoffStrategy, Scheduler scheduler) { return new InstanceQuarantiner<T>(factory, backoffStrategy, scheduler); } }
InstanceQuarantiner implements Func1<Instance<T>, Observable<Instance<T>>> { public static <T> InstanceQuarantiner<T> create( IncarnationFactory<T> factory, DelayStrategy backoffStrategy, Scheduler scheduler) { return new InstanceQuarantiner<T>(factory, backoffStrategy, scheduler); } InstanceQuarantiner(IncarnationFactory<T> factory, DelayStrategy backoffStrategy, Scheduler scheduler); }
InstanceQuarantiner implements Func1<Instance<T>, Observable<Instance<T>>> { public static <T> InstanceQuarantiner<T> create( IncarnationFactory<T> factory, DelayStrategy backoffStrategy, Scheduler scheduler) { return new InstanceQuarantiner<T>(factory, backoffStrategy, scheduler); } InstanceQuarantiner(IncarnationFactory<T> factory, DelayStrategy backoffStrategy, Scheduler scheduler); static InstanceQuarantiner<T> create(
            IncarnationFactory<T> factory,
            DelayStrategy         backoffStrategy, 
            Scheduler             scheduler); @Override Observable<Instance<T>> call(final Instance<T> primaryInstance); }
InstanceQuarantiner implements Func1<Instance<T>, Observable<Instance<T>>> { public static <T> InstanceQuarantiner<T> create( IncarnationFactory<T> factory, DelayStrategy backoffStrategy, Scheduler scheduler) { return new InstanceQuarantiner<T>(factory, backoffStrategy, scheduler); } InstanceQuarantiner(IncarnationFactory<T> factory, DelayStrategy backoffStrategy, Scheduler scheduler); static InstanceQuarantiner<T> create(
            IncarnationFactory<T> factory,
            DelayStrategy         backoffStrategy, 
            Scheduler             scheduler); @Override Observable<Instance<T>> call(final Instance<T> primaryInstance);  }