@Test public void testDelayedOnError() throws Exception { final AtomicLong start = new AtomicLong(); final AtomicLong end = new AtomicLong(); final long delay = 100; final int repeat = 20; final CountDownLatch done = new CountDownLatch(repeat); ClientResumePolicy<String, String> policy = ClientResumePolicies.delayed(new Func0<Long>() { @Override public Long call() { return delay; } }, TimeUnit.MILLISECONDS); ServerClientContext<String, String> context = new ServerClientContext<>(new ServerInfo("localhost", 1000), client, factory, observer); start.set(System.currentTimeMillis()); end.set(0); for (int i = 0; i < repeat; ++i) { Observable<HttpClientResponse<String>> resumedOnCompleted = policy.onError(context, i, new Throwable("error")); resumedOnCompleted.subscribe(new Subscriber<HttpClientResponse<String>>() { @Override public void onCompleted() { end.getAndAdd(System.currentTimeMillis() - start.get()); done.countDown(); } @Override public void onError(Throwable e) { } @Override public void onNext(HttpClientResponse<String> stringHttpClientResponse) { } }); } long wait = 5; if (!done.await(5, TimeUnit.SECONDS)) { fail("It should take far less than " + wait + " seconds to run the test. "); } long elapsed = end.get(); long maxDelay = delay + delay / 2; assertTrue(String.format("The delay should be more than %d millionseconds, but no more than %d millionseconds. The actual: %d", repeat * delay, repeat * maxDelay, elapsed), elapsed >= repeat * delay && elapsed <= repeat * maxDelay); }
public static <R, E> ClientResumePolicy<R, E> delayed(final Func0<Long> delayFunc, final TimeUnit unit) { return new ClientResumePolicy<R, E>() { @Override public Observable<HttpClientResponse<E>> onError(ServerClientContext<R, E> clientContext, int attempts, Throwable error) { return createDelayedResponse(clientContext); } @Override public Observable<HttpClientResponse<E>> onCompleted(ServerClientContext<R, E> clientContext, int attempts) { return createDelayedResponse(clientContext); } private Observable<HttpClientResponse<E>> createDelayedResponse(ServerClientContext<R, E> clientContext) { return clientContext.newResponse() .delaySubscription(delayFunc.call(), unit); } }; }
ClientResumePolicies { public static <R, E> ClientResumePolicy<R, E> delayed(final Func0<Long> delayFunc, final TimeUnit unit) { return new ClientResumePolicy<R, E>() { @Override public Observable<HttpClientResponse<E>> onError(ServerClientContext<R, E> clientContext, int attempts, Throwable error) { return createDelayedResponse(clientContext); } @Override public Observable<HttpClientResponse<E>> onCompleted(ServerClientContext<R, E> clientContext, int attempts) { return createDelayedResponse(clientContext); } private Observable<HttpClientResponse<E>> createDelayedResponse(ServerClientContext<R, E> clientContext) { return clientContext.newResponse() .delaySubscription(delayFunc.call(), unit); } }; } }
ClientResumePolicies { public static <R, E> ClientResumePolicy<R, E> delayed(final Func0<Long> delayFunc, final TimeUnit unit) { return new ClientResumePolicy<R, E>() { @Override public Observable<HttpClientResponse<E>> onError(ServerClientContext<R, E> clientContext, int attempts, Throwable error) { return createDelayedResponse(clientContext); } @Override public Observable<HttpClientResponse<E>> onCompleted(ServerClientContext<R, E> clientContext, int attempts) { return createDelayedResponse(clientContext); } private Observable<HttpClientResponse<E>> createDelayedResponse(ServerClientContext<R, E> clientContext) { return clientContext.newResponse() .delaySubscription(delayFunc.call(), unit); } }; }  }
ClientResumePolicies { public static <R, E> ClientResumePolicy<R, E> delayed(final Func0<Long> delayFunc, final TimeUnit unit) { return new ClientResumePolicy<R, E>() { @Override public Observable<HttpClientResponse<E>> onError(ServerClientContext<R, E> clientContext, int attempts, Throwable error) { return createDelayedResponse(clientContext); } @Override public Observable<HttpClientResponse<E>> onCompleted(ServerClientContext<R, E> clientContext, int attempts) { return createDelayedResponse(clientContext); } private Observable<HttpClientResponse<E>> createDelayedResponse(ServerClientContext<R, E> clientContext) { return clientContext.newResponse() .delaySubscription(delayFunc.call(), unit); } }; }  static ClientResumePolicy<R, E> maxRepeat(final int maxRepeat); static ClientResumePolicy<R, E> noRepeat(); static ClientResumePolicy<R, E> delayed(final Func0<Long> delayFunc, final TimeUnit unit); static ClientResumePolicy<R, E> maxRepeat(final ClientResumePolicy<R, E> policy, final int maxRepeat); }
ClientResumePolicies { public static <R, E> ClientResumePolicy<R, E> delayed(final Func0<Long> delayFunc, final TimeUnit unit) { return new ClientResumePolicy<R, E>() { @Override public Observable<HttpClientResponse<E>> onError(ServerClientContext<R, E> clientContext, int attempts, Throwable error) { return createDelayedResponse(clientContext); } @Override public Observable<HttpClientResponse<E>> onCompleted(ServerClientContext<R, E> clientContext, int attempts) { return createDelayedResponse(clientContext); } private Observable<HttpClientResponse<E>> createDelayedResponse(ServerClientContext<R, E> clientContext) { return clientContext.newResponse() .delaySubscription(delayFunc.call(), unit); } }; }  static ClientResumePolicy<R, E> maxRepeat(final int maxRepeat); static ClientResumePolicy<R, E> noRepeat(); static ClientResumePolicy<R, E> delayed(final Func0<Long> delayFunc, final TimeUnit unit); static ClientResumePolicy<R, E> maxRepeat(final ClientResumePolicy<R, E> policy, final int maxRepeat);  }