IllegalStateException в retryWhen с ReadStreamAdapter

Question

Я использую retryWhen, когда внешний HTTP-запрос к одной из моих внешних служб завершается с ошибкой. Проблема заключается в том, что I'm с помощью

RxHelper.toObservable(httpClient.request(method, url)) 

Чтобы получить мою наблюдаемую реакцию, и becuase VertX внутренне использовать ReadStreamAdapter я не могу использовать, потому что retryWhen подмигнули жалуются

java java.lang.IllegalStateException: Request already complete 

Вот пример кода:

RxHelper.toObservable(httpClient.request(method, url)) 
    .retryWhen(new ServiceExceptionRetry())  
    .subscribe(f -> replySuccess(eventMsg, event, f), t -> handleError(t, eventMsg, event)); 

Любая идея, как достичь этого?

Answer 1

Вы можете использовать defer чтобы создать Наблюдаемые из method и client каждый раз, как это:

Observable.defer(() -> RxHelper.toObservable(httpClient.request(method, url))) 
    .retryWhen(new ServiceExceptionRetry())  
    .subscribe(f -> replySuccess(eventMsg, event, f), t -> handleError(t, eventMsg, event));