andreacavalli/tdlib-session-container
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Implement timeouts properly

Browse Source
This commit is contained in:
Andrea Cavalli 2022-06-28 00:11:34 +02:00
parent 07c22d39f2
commit d931217e81
1 changed files with 5 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
src/main/java/it/tdlight/reactiveapi/BaseAtomixReactiveApiClient.java
View File

@ -102,14 +102,16 @@ abstract class BaseAtomixReactiveApiClient implements ReactiveApiClient, AutoClo
return Mono.defer(() -> { return Mono.defer(() -> {
var requestId = this.requestId.getAndIncrement(); var requestId = this.requestId.getAndIncrement();
var timeoutError = new TdError(408, "Request Timeout"); var timeoutError = new TdError(408, "Request Timeout");
Mono<T> timeoutErrorMono = Mono.error(timeoutError); var requestTimestamp = Instant.now();
var timeoutDuration = Duration.between(Instant.now(), timeout); var timeoutDuration = Duration.between(requestTimestamp, timeout);
if (timeoutDuration.isNegative() || timeoutDuration.isZero()) { if (timeoutDuration.isNegative() || timeoutDuration.isZero()) {
return timeoutErrorMono; return Mono.error(timeoutError);
} }
var cf = new CompletableFuture<Timestamped<OnResponse<TdApi.Object>>>(); var cf = new CompletableFuture<Timestamped<OnResponse<TdApi.Object>>>();
this.responses.put(requestId, cf); this.responses.put(requestId, cf);
Mono<T> response = Mono.fromFuture(cf) Mono<T> response = Mono.fromFuture(cf)
.timeout(timeoutDuration, Mono.fromSupplier(() -> new Timestamped<>(requestTimestamp.toEpochMilli(),
new Response<>(clientId, requestId, userId, new TdApi.Error(408, "Request Timeout")))))
.<T>handle((responseObj, sink) -> { .<T>handle((responseObj, sink) -> {
if (Instant.ofEpochMilli(responseObj.timestamp()).compareTo(timeout) > 0) { if (Instant.ofEpochMilli(responseObj.timestamp()).compareTo(timeout) > 0) {
sink.error(new TdError(408, "Request Timeout")); sink.error(new TdError(408, "Request Timeout"));
@ -128,7 +130,6 @@ abstract class BaseAtomixReactiveApiClient implements ReactiveApiClient, AutoClo
sink.error(new UnsupportedOperationException("Unknown response type: " + responseObj.data().getClass())); sink.error(new UnsupportedOperationException("Unknown response type: " + responseObj.data().getClass()));
} }
}) })
.timeout(timeoutDuration, timeoutErrorMono)
.doFinally(s -> this.responses.remove(requestId)); .doFinally(s -> this.responses.remove(requestId));
requests.emitNext(new Request<>(clientId, requestId, request, timeout), EmitFailureHandler.busyLooping(TEN_MS)); requests.emitNext(new Request<>(clientId, requestId, request, timeout), EmitFailureHandler.busyLooping(TEN_MS));
return response; return response;