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

Code cleanup

Browse Source
This commit is contained in:
Andrea Cavalli 2022-10-08 13:44:32 +02:00
parent 1eb1d7b95f
commit 59027e8e62
2 changed files with 17 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -34,6 +34,8 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import reactor.core.Disposable; import reactor.core.Disposable;
import reactor.core.publisher.Mono; import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoSink;
import reactor.core.publisher.Sinks;
import reactor.core.publisher.Sinks.EmitFailureHandler; import reactor.core.publisher.Sinks.EmitFailureHandler;
import reactor.core.publisher.Sinks.Many; import reactor.core.publisher.Sinks.Many;
import reactor.core.scheduler.Schedulers; import reactor.core.scheduler.Schedulers;
@ -41,14 +43,12 @@ import reactor.core.scheduler.Schedulers;
abstract class BaseAtomixReactiveApiClient implements ReactiveApiMultiClient { abstract class BaseAtomixReactiveApiClient implements ReactiveApiMultiClient {
private static final Logger LOG = LoggerFactory.getLogger(BaseAtomixReactiveApiClient.class); private static final Logger LOG = LoggerFactory.getLogger(BaseAtomixReactiveApiClient.class);
private static final Duration HUNDRED_MS = Duration.ofMillis(100);
private static final long EMPTY_USER_ID = 0; private static final long EMPTY_USER_ID = 0;
// Temporary id used to make requests // Temporary id used to make requests
private final long clientId; private final long clientId;
private final Consumer<OnRequest<?>> requests; private final Consumer<OnRequest<?>> requests;
private final Map<Long, CompletableFuture<Timestamped<OnResponse<TdApi.Object>>>> responses private final Map<Long, MonoSink<Timestamped<OnResponse<Object>>>> responses
= new ConcurrentHashMap<>(); = new ConcurrentHashMap<>();
private final AtomicLong requestId = new AtomicLong(0); private final AtomicLong requestId = new AtomicLong(0);
private final Disposable subscription; private final Disposable subscription;
@ -63,7 +63,7 @@ abstract class BaseAtomixReactiveApiClient implements ReactiveApiMultiClient {
LOG.debug("Bot received a response for an unknown request id: {}", response.data().requestId()); LOG.debug("Bot received a response for an unknown request id: {}", response.data().requestId());
return; return;
} }
responseSink.complete(response); responseSink.success(response);
}).subscribeOn(Schedulers.parallel()) }).subscribeOn(Schedulers.parallel())
.subscribe(v -> {}, ex -> LOG.error("Reactive api client responses flux has failed unexpectedly!", ex)); .subscribe(v -> {}, ex -> LOG.error("Reactive api client responses flux has failed unexpectedly!", ex));
} }
@ -78,9 +78,14 @@ abstract class BaseAtomixReactiveApiClient implements ReactiveApiMultiClient {
if (timeoutDuration.isNegative() || timeoutDuration.isZero()) { if (timeoutDuration.isNegative() || timeoutDuration.isZero()) {
return Mono.error(timeoutError); return Mono.error(timeoutError);
} }
var cf = new CompletableFuture<Timestamped<OnResponse<TdApi.Object>>>(); Mono<T> response = Mono
this.responses.put(requestId, cf); .<Timestamped<OnResponse<TdApi.Object>>>create(sink -> {
Mono<T> response = Mono.fromFuture(() -> cf) sink.onDispose(() -> this.responses.remove(requestId, sink));
var prev = this.responses.putIfAbsent(requestId, sink);
if (prev != null) {
sink.error(new IllegalStateException("Can't call the same request twice: " + requestId));
}
})
.timeout(timeoutDuration, Mono.fromSupplier(() -> new Timestamped<>(requestTimestamp.toEpochMilli(), .timeout(timeoutDuration, Mono.fromSupplier(() -> new Timestamped<>(requestTimestamp.toEpochMilli(),
new Response<>(clientId, requestId, userId, new TdApi.Error(408, "Request Timeout"))))) new Response<>(clientId, requestId, userId, new TdApi.Error(408, "Request Timeout")))))
.<T>handle((responseObj, sink) -> { .<T>handle((responseObj, sink) -> {
@ -100,8 +105,7 @@ abstract class BaseAtomixReactiveApiClient implements ReactiveApiMultiClient {
} else { } else {
sink.error(new UnsupportedOperationException("Unknown response type: " + responseObj.data().getClass())); sink.error(new UnsupportedOperationException("Unknown response type: " + responseObj.data().getClass()));
} }
}) });
.doFinally(s -> this.responses.remove(requestId));
requests.accept(new Request<>(userId, clientId, requestId, request, timeout)); requests.accept(new Request<>(userId, clientId, requestId, request, timeout));
return response; return response;
}); });
@ -139,7 +143,7 @@ abstract class BaseAtomixReactiveApiClient implements ReactiveApiMultiClient {
return Mono.fromRunnable(() -> { return Mono.fromRunnable(() -> {
subscription.dispose(); subscription.dispose();
long now = System.currentTimeMillis(); long now = System.currentTimeMillis();
responses.forEach((requestId, cf) -> cf.complete(new Timestamped<>(now, responses.forEach((requestId, cf) -> cf.success(new Timestamped<>(now,
new Response<>(clientId, requestId, EMPTY_USER_ID, new Error(408, "Request Timeout")) new Response<>(clientId, requestId, EMPTY_USER_ID, new Error(408, "Request Timeout"))
))); )));
responses.clear(); responses.clear();

4
src/main/java/it/tdlight/reactiveapi/TdlibChannelsSharedReceive.java
View File

@ -64,7 +64,9 @@ public class TdlibChannelsSharedReceive implements Closeable {
.subscribeOn(Schedulers.parallel()) .subscribeOn(Schedulers.parallel())
.subscribe(n -> {}, ex -> { .subscribe(n -> {}, ex -> {
LOG.error("An error when handling requests killed the requests subscriber!", ex); LOG.error("An error when handling requests killed the requests subscriber!", ex);
requests.emitError(ex, EmitFailureHandler.busyLooping(Duration.ofMillis(100))); synchronized (requests) {
requests.emitError(ex, EmitFailureHandler.FAIL_FAST);
}
}); });
} }