diff --git a/pom.xml b/pom.xml
index d09073b..6cb199a 100644
--- a/pom.xml
+++ b/pom.xml
@@ -129,7 +129,7 @@
it.tdlight
tdlight-java
- [3.171.31,)
+ [3.171.36,)
it.cavallium
diff --git a/src/main/java/it/tdlight/tdlibsession/EventBusFlux.java b/src/main/java/it/tdlight/tdlibsession/EventBusFlux.java
index 805f586..d5153e3 100644
--- a/src/main/java/it/tdlight/tdlibsession/EventBusFlux.java
+++ b/src/main/java/it/tdlight/tdlibsession/EventBusFlux.java
@@ -2,14 +2,24 @@ package it.tdlight.tdlibsession;
import io.vertx.core.eventbus.DeliveryOptions;
import io.vertx.core.eventbus.EventBus;
+import io.vertx.core.eventbus.Message;
import io.vertx.core.eventbus.MessageCodec;
import io.vertx.core.eventbus.MessageConsumer;
+import io.vertx.core.eventbus.ReplyException;
+import it.tdlight.utils.MonoUtils;
import java.time.Duration;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import org.jetbrains.annotations.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import reactor.core.Disposable;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
+import reactor.core.publisher.Sinks;
+import reactor.core.publisher.Sinks.One;
+import reactor.util.function.Tuple2;
+import reactor.util.function.Tuples;
public class EventBusFlux {
private static final Logger logger = LoggerFactory.getLogger(EventBusFlux.class);
@@ -27,7 +37,11 @@ public class EventBusFlux {
}
}
- public static Mono serve(Flux flux,
+ /**
+ *
+ * @return tuple. T1 = flux served, T2 = error that caused cancelling of the subscription
+ */
+ public static Tuple2, Mono> serve(Flux flux,
EventBus eventBus,
String fluxAddress,
DeliveryOptions baseDeliveryOptions,
@@ -40,7 +54,8 @@ public class EventBusFlux {
var signalDeliveryOptions = new DeliveryOptions(deliveryOptions)
.setCodecName(signalsCodec.name());
AtomicInteger subscriptionsCount = new AtomicInteger();
- return Mono.create(sink -> {
+ One fatalErrorSink = Sinks.one();
+ var servedMono = Mono.create(sink -> {
MessageConsumer subscribe = eventBus.consumer(fluxAddress + ".subscribe");
subscribe.handler(msg -> {
@@ -62,47 +77,52 @@ public class EventBusFlux {
subscriptionReady.handler(subscriptionReadyMsg -> {
subscriptionReady.unregister(subscriptionReadyUnregistered -> {
if (subscriptionReadyUnregistered.succeeded()) {
- var subscription = flux.subscribe(item -> {
- eventBus.request(subscriptionAddress + ".signal", SignalMessage.onNext(item), signalDeliveryOptions, msg2 -> {
- if (msg2.failed()) {
- logger.error("Failed to send onNext signal", msg2.cause());
- }
- });
- }, error -> {
- eventBus.request(subscriptionAddress + ".signal", SignalMessage.onError(error), signalDeliveryOptions, msg2 -> {
- logger.info("Errored flux \"" + fluxAddress + "\"");
- if (msg2.failed()) {
- logger.error("Failed to send onError signal", msg2.cause());
- }
- });
- }, () -> {
- eventBus.request(subscriptionAddress + ".signal", SignalMessage.onComplete(), signalDeliveryOptions, msg2 -> {
- logger.info("Completed flux \"" + fluxAddress + "\"");
- if (msg2.failed()) {
- logger.error("Failed to send onComplete signal", msg2.cause());
- }
- });
- });
+ AtomicReference atomicSubscription = new AtomicReference<>(null);
+ var subscription = flux
+ .onErrorResume(error -> Mono
+ .>create(errorSink -> {
+ var responseHandler = MonoUtils.toHandler(errorSink);
+ eventBus.request(subscriptionAddress + ".signal", SignalMessage.onError(error), signalDeliveryOptions, responseHandler);
+ })
+ .then(Mono.empty())
+ )
+ .flatMap(item -> Mono.>create(itemSink -> {
+ var responseHandler = MonoUtils.toHandler(itemSink);
+ eventBus.request(subscriptionAddress + ".signal", SignalMessage.onNext(item), signalDeliveryOptions, responseHandler);
+ })).subscribe(response -> {}, error -> {
+ if (error instanceof ReplyException) {
+ var errorMessage = error.getMessage();
+ if (errorMessage != null && errorMessage.contains("NO_HANDLERS")) {
+ logger.error("Can't send a signal of flux \"" + fluxAddress + "\" because the connection was lost");
+ } else {
+ logger.error("Error when sending a signal of flux \"" + fluxAddress + "\": {}", error.getLocalizedMessage());
+ }
+ } else {
+ logger.error("Error when sending a signal of flux \"" + fluxAddress + "\"", error);
+ }
+ fatalErrorSink.tryEmitValue(error);
+ disposeFlux(atomicSubscription.get(), fatalErrorSink, cancel, dispose, fluxAddress, () -> {
+ logger.warn("Forcefully disposed \"" + fluxAddress + "\" caused by the previous error");
+ });
+ }, () -> {
+ eventBus.request(subscriptionAddress + ".signal", SignalMessage.onComplete(), signalDeliveryOptions, msg2 -> {
+ logger.info("Completed flux \"" + fluxAddress + "\"");
+ if (msg2.failed()) {
+ logger.error("Failed to send onComplete signal", msg2.cause());
+ fatalErrorSink.tryEmitValue(msg2.cause());
+ }
+ });
+ });
+ atomicSubscription.set(subscription);
cancel.handler(msg3 -> {
- logger.warn("Cancelling flux \"" + fluxAddress + "\"");
+ logger.trace("Cancelling flux \"" + fluxAddress + "\"");
subscription.dispose();
+ logger.debug("Cancelled flux \"" + fluxAddress + "\"");
msg3.reply(EMPTY, deliveryOptions);
});
dispose.handler(msg2 -> {
- logger.warn("Disposing flux \"" + fluxAddress + "\"");
- subscription.dispose();
- cancel.unregister(v -> {
- if (v.failed()) {
- logger.error("Failed to unregister cancel", v.cause());
- }
- dispose.unregister(v2 -> {
- if (v.failed()) {
- logger.error("Failed to unregister dispose", v2.cause());
- }
- msg2.reply(EMPTY);
- });
- });
+ disposeFlux(subscription, fatalErrorSink, cancel, dispose, fluxAddress, () -> msg2.reply(EMPTY));
});
cancel.completionHandler(h -> {
@@ -148,6 +168,33 @@ public class EventBusFlux {
}
});
});
+
+ return Tuples.of(servedMono, fatalErrorSink.asMono());
+ }
+
+ private static void disposeFlux(@Nullable Disposable subscription,
+ One fatalErrorSink,
+ MessageConsumer cancel,
+ MessageConsumer dispose,
+ String fluxAddress,
+ Runnable after) {
+ logger.trace("Disposing flux \"" + fluxAddress + "\"");
+ fatalErrorSink.tryEmitEmpty();
+ if (subscription != null) {
+ subscription.dispose();
+ }
+ cancel.unregister(v -> {
+ if (v.failed()) {
+ logger.error("Failed to unregister cancel", v.cause());
+ }
+ dispose.unregister(v2 -> {
+ if (v.failed()) {
+ logger.error("Failed to unregister dispose", v2.cause());
+ }
+ logger.debug("Disposed flux \"" + fluxAddress + "\"");
+ after.run();
+ });
+ });
}
public static Flux connect(EventBus eventBus,
diff --git a/src/main/java/it/tdlight/tdlibsession/remoteclient/TDLibRemoteClient.java b/src/main/java/it/tdlight/tdlibsession/remoteclient/TDLibRemoteClient.java
index 0cf71e8..9836664 100644
--- a/src/main/java/it/tdlight/tdlibsession/remoteclient/TDLibRemoteClient.java
+++ b/src/main/java/it/tdlight/tdlibsession/remoteclient/TDLibRemoteClient.java
@@ -3,6 +3,7 @@ package it.tdlight.tdlibsession.remoteclient;
import io.vertx.core.AsyncResult;
import io.vertx.core.Future;
import io.vertx.core.Handler;
+import io.vertx.core.Promise;
import io.vertx.core.eventbus.Message;
import io.vertx.core.net.JksOptions;
import io.vertx.core.shareddata.AsyncMap;
@@ -17,6 +18,7 @@ import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
import org.apache.logging.log4j.LogManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -37,6 +39,7 @@ public class TDLibRemoteClient implements AutoCloseable {
private final Set membersAddresses;
private final Many clusterManager = Sinks.many().replay().latest();
private final Scheduler deploymentScheduler = Schedulers.newSingle("TDLib", false);
+ private final AtomicInteger statsActiveDeployments = new AtomicInteger();
public TDLibRemoteClient(SecurityInfo securityInfo, String masterHostname, String netInterface, int port, Set membersAddresses) {
this.securityInfo = securityInfo;
@@ -124,7 +127,7 @@ public class TDLibRemoteClient implements AutoCloseable {
if (mapResult.succeeded()) {
var deployableBotAddresses = mapResult.result();
- clusterManager.getSharedData().getLockWithTimeout("deployment", 15000, lockAcquisitionResult -> {
+ clusterManager.getSharedData().getLock("deployment", lockAcquisitionResult -> {
if (lockAcquisitionResult.succeeded()) {
var deploymentLock = lockAcquisitionResult.result();
putAllAsync(deployableBotAddresses, botAddresses.values(), (AsyncResult putAllResult) -> {
@@ -168,7 +171,7 @@ public class TDLibRemoteClient implements AutoCloseable {
AsyncMap deployableBotAddresses) {
clusterManager.getEventBus().consumer("tdlib.remoteclient.clients.deploy", (Message msg) -> {
- clusterManager.getSharedData().getLockWithTimeout("deployment", 15000, lockAcquisitionResult -> {
+ clusterManager.getSharedData().getLock("deployment", lockAcquisitionResult -> {
if (lockAcquisitionResult.succeeded()) {
var deploymentLock = lockAcquisitionResult.result();
var botAddress = msg.body();
@@ -229,14 +232,23 @@ public class TDLibRemoteClient implements AutoCloseable {
}
private void deployBot(TdClusterManager clusterManager, String botAddress, Handler> deploymentHandler) {
+ logger.info("Active deployments: " + statsActiveDeployments.incrementAndGet());
AsyncTdMiddleEventBusServer verticle = new AsyncTdMiddleEventBusServer(clusterManager);
- verticle.onBeforeStop(handler -> {
- clusterManager.getSharedData().getLockWithTimeout("deployment", 30000, lockAcquisitionResult -> {
+ var afterStopPromise = Promise.promise();
+ if (verticle.onAfterStop(handler -> {
+ afterStopPromise.complete();
+ handler.complete();
+ }, true).isFailure()) {
+ deploymentHandler.handle(Future.failedFuture(new IllegalStateException("Failed to register to event onAfterStop")));
+ return;
+ }
+ if (verticle.onBeforeStop(handler -> {
+ clusterManager.getSharedData().getLock("deployment", lockAcquisitionResult -> {
if (lockAcquisitionResult.succeeded()) {
var deploymentLock = lockAcquisitionResult.result();
- verticle.onAfterStop(handler2 -> {
+ afterStopPromise.future().onComplete(handler2 -> {
deploymentLock.release();
- handler2.complete();
+ logger.info("Active deployments: " + statsActiveDeployments.decrementAndGet());
});
clusterManager.getSharedData().getClusterWideMap("runningBotAddresses", (AsyncResult> mapResult) -> {
if (mapResult.succeeded()) {
@@ -260,7 +272,10 @@ public class TDLibRemoteClient implements AutoCloseable {
handler.fail(lockAcquisitionResult.cause());
}
});
- });
+ }, false).isFailure()) {
+ deploymentHandler.handle(Future.failedFuture(new IllegalStateException("Failed to register to event onBeforeStop")));
+ return;
+ }
verticle.start(botAddress, botAddress, false).doOnError(error -> {
logger.error("Can't deploy bot \"" + botAddress + "\"", error);
}).subscribeOn(deploymentScheduler).subscribe(v -> {}, err -> {
diff --git a/src/main/java/it/tdlight/tdlibsession/td/direct/AsyncTdDirectImpl.java b/src/main/java/it/tdlight/tdlibsession/td/direct/AsyncTdDirectImpl.java
index 85c2984..abc3753 100644
--- a/src/main/java/it/tdlight/tdlibsession/td/direct/AsyncTdDirectImpl.java
+++ b/src/main/java/it/tdlight/tdlibsession/td/direct/AsyncTdDirectImpl.java
@@ -10,7 +10,6 @@ import it.tdlight.jni.TdApi.Ok;
import it.tdlight.jni.TdApi.UpdateAuthorizationState;
import it.tdlight.tdlibsession.td.TdResult;
import it.tdlight.tdlight.ClientManager;
-import java.time.Duration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.publisher.Flux;
@@ -36,7 +35,7 @@ public class AsyncTdDirectImpl implements AsyncTdDirect {
@Override
public Mono> execute(Function request, boolean synchronous) {
if (synchronous) {
- return td.asMono().flatMap(td -> Mono.fromCallable(() -> {
+ return td.asMono().single().flatMap(td -> Mono.fromCallable(() -> {
if (td != null) {
return TdResult.of(td.execute(request));
} else {
@@ -47,18 +46,17 @@ public class AsyncTdDirectImpl implements AsyncTdDirect {
}
}).publishOn(Schedulers.boundedElastic()).single()).subscribeOn(tdScheduler);
} else {
- return td.asMono().flatMap(td -> Mono.>create(sink -> {
+ return td.asMono().single().flatMap(td -> Mono.>create(sink -> {
if (td != null) {
- try {
- td.send(request, v -> sink.success(TdResult.of(v)), sink::error);
- } catch (Throwable t) {
- sink.error(t);
- }
+ td.send(request, v -> sink.success(TdResult.of(v)), sink::error);
} else {
if (request.getConstructor() == Close.CONSTRUCTOR) {
+ logger.trace("Sending close success to sink " + sink.toString());
sink.success(TdResult.of(new Ok()));
+ } else {
+ logger.trace("Sending close error to sink " + sink.toString());
+ sink.error(new IllegalStateException("TDLib client is destroyed"));
}
- sink.error(new IllegalStateException("TDLib client is destroyed"));
}
})).single().subscribeOn(tdScheduler);
}
@@ -66,15 +64,17 @@ public class AsyncTdDirectImpl implements AsyncTdDirect {
@Override
public Flux> receive(AsyncTdDirectOptions options) {
+ // If closed it will be either true or false
+ final One closedFromTd = Sinks.one();
return Flux.>create(emitter -> {
- One closedFromTd = Sinks.one();
var client = ClientManager.create((Object object) -> {
emitter.next(TdResult.of(object));
// Close the emitter if receive closed state
if (object.getConstructor() == UpdateAuthorizationState.CONSTRUCTOR
&& ((UpdateAuthorizationState) object).authorizationState.getConstructor()
== AuthorizationStateClosed.CONSTRUCTOR) {
- closedFromTd.tryEmitValue(new java.lang.Object());
+ logger.debug("Received closed status from tdlib");
+ closedFromTd.tryEmitValue(true);
emitter.complete();
}
}, emitter::error, emitter::error);
@@ -86,10 +86,21 @@ public class AsyncTdDirectImpl implements AsyncTdDirect {
// Send close if the stream is disposed before tdlib is closed
emitter.onDispose(() -> {
- closedFromTd.asMono().take(Duration.ofMillis(10)).switchIfEmpty(Mono.fromRunnable(() -> client.send(new Close(),
- result -> logger.warn("Close result: {}", result),
- ex -> logger.error("Error when disposing td client", ex)
- ))).subscribeOn(tdScheduler).subscribe();
+ // Try to emit false, so that if it has not been closed from tdlib, now it is explicitly false.
+ closedFromTd.tryEmitValue(false);
+
+ closedFromTd.asMono()
+ .doOnNext(isClosedFromTd -> {
+ if (!isClosedFromTd) {
+ logger.warn("The stream has been disposed without closing tdlib. Sending TdApi.Close()...");
+ client.send(new Close(),
+ result -> logger.warn("Close result: {}", result),
+ ex -> logger.error("Error when disposing td client", ex)
+ );
+ }
+ })
+ .subscribeOn(tdScheduler)
+ .subscribe();
});
}).subscribeOn(tdScheduler);
}
diff --git a/src/main/java/it/tdlight/tdlibsession/td/easy/AsyncTdEasy.java b/src/main/java/it/tdlight/tdlibsession/td/easy/AsyncTdEasy.java
index c77a35a..801c7e0 100644
--- a/src/main/java/it/tdlight/tdlibsession/td/easy/AsyncTdEasy.java
+++ b/src/main/java/it/tdlight/tdlibsession/td/easy/AsyncTdEasy.java
@@ -66,25 +66,33 @@ public class AsyncTdEasy {
private final One fatalError = Sinks.one();
private final AsyncTdMiddle td;
private final String logName;
- private final Flux incomingUpdatesCo;
+ private final Flux incomingUpdates;
public AsyncTdEasy(AsyncTdMiddle td, String logName) {
this.td = td;
this.logName = logName;
// todo: use Duration.ZERO instead of 10ms interval
- this.incomingUpdatesCo = td.receive()
- .filterWhen(update -> Mono.from(requestedDefinitiveExit).map(requestedDefinitiveExit -> !requestedDefinitiveExit))
+ this.incomingUpdates = td.receive()
.flatMap(this::preprocessUpdates)
.flatMap(update -> Mono.from(this.getState()).single().map(state -> new AsyncTdUpdateObj(state, update)))
.filter(upd -> upd.getState().getConstructor() == AuthorizationStateReady.CONSTRUCTOR)
.map(upd -> (TdApi.Update) upd.getUpdate())
.doOnError(ex -> {
logger.error(ex.getLocalizedMessage(), ex);
- }).doOnNext(v -> {
- if (logger.isDebugEnabled()) logger.debug(v.toString());
}).doOnComplete(() -> {
- authState.onNext(new AuthorizationStateClosed());
+ authState.asFlux().take(1).single().subscribe(authState -> {
+ if (authState.getConstructor() != AuthorizationStateClosed.CONSTRUCTOR) {
+ logger.warn("Updates stream has closed while"
+ + " the current authorization state is"
+ + " still {}. Setting authorization state as closed!", authState.getClass().getSimpleName());
+ this.authState.onNext(new AuthorizationStateClosed());
+ }
+ });
+ })
+ .doOnTerminate(() -> {
+ logger.debug("Incoming updates flux terminated. Setting requestedDefinitiveExit: true");
+ requestedDefinitiveExit.onNext(true);
})
.subscribeOn(scheduler)
.publish().refCount(1);
@@ -130,7 +138,7 @@ public class AsyncTdEasy {
}
private Flux getIncomingUpdates(boolean includePreAuthUpdates) {
- return Flux.from(incomingUpdatesCo).doFinally(s -> requestedDefinitiveExit.onNext(true)).subscribeOn(scheduler);
+ return incomingUpdates.subscribeOn(scheduler);
}
/**
@@ -291,19 +299,27 @@ public class AsyncTdEasy {
return true;
}
})
- .then(Mono.from(requestedDefinitiveExit).single())
+ .then(requestedDefinitiveExit.asFlux().take(1).single())
.filter(closeRequested -> !closeRequested)
- .doOnSuccess(v -> requestedDefinitiveExit.onNext(true))
- .then(td.execute(new TdApi.Close(), false))
- .doOnNext(ok -> {
- logger.debug("Received Ok after TdApi.Close");
+ .doOnSuccess(s -> {
+ logger.debug("Setting requestedDefinitiveExit: true");
+ requestedDefinitiveExit.onNext(true);
})
+ .then(td.execute(new TdApi.Close(), false).doOnSubscribe(s -> {
+ logger.debug("Sending TdApi.Close");
+ }))
+ .doOnNext(closeResponse -> logger.debug("TdApi.Close response is: \"{}\"",
+ closeResponse.toString().replace('\n', ' ')
+ ))
.then(authState
.filter(authorizationState -> authorizationState.getConstructor() == AuthorizationStateClosed.CONSTRUCTOR)
.take(1)
.singleOrEmpty())
.doOnNext(ok -> {
- logger.info("Received AuthorizationStateClosed after TdApi.Close");
+ logger.debug("Received AuthorizationStateClosed after TdApi.Close");
+ })
+ .doOnSuccess(s -> {
+ logger.info("AsyncTdEasy closed successfully");
})
.then()
.subscribeOn(scheduler);
@@ -495,14 +511,18 @@ public class AsyncTdEasy {
this.authState.onNext(new AuthorizationStateReady());
return Mono.empty();
}
+ case AuthorizationStateClosing.CONSTRUCTOR:
+ logger.debug("Received AuthorizationStateClosing from td");
+ return Mono.empty();
case AuthorizationStateClosed.CONSTRUCTOR:
+ logger.debug("Received AuthorizationStateClosed from td");
return Mono.from(requestedDefinitiveExit).doOnNext(closeRequested -> {
if (closeRequested) {
- logger.info("AsyncTdEasy closed successfully");
+ logger.debug("td closed successfully");
} else {
- logger.warn("AsyncTdEasy closed unexpectedly: " + logName);
- authState.onNext(obj);
+ logger.warn("td closed unexpectedly: {}", logName);
}
+ authState.onNext(obj);
}).flatMap(closeRequested -> {
if (closeRequested) {
return Mono
diff --git a/src/main/java/it/tdlight/tdlibsession/td/middle/client/AsyncTdMiddleEventBusClient.java b/src/main/java/it/tdlight/tdlibsession/td/middle/client/AsyncTdMiddleEventBusClient.java
index 73f5858..89aefbb 100644
--- a/src/main/java/it/tdlight/tdlibsession/td/middle/client/AsyncTdMiddleEventBusClient.java
+++ b/src/main/java/it/tdlight/tdlibsession/td/middle/client/AsyncTdMiddleEventBusClient.java
@@ -37,6 +37,7 @@ import org.slf4j.LoggerFactory;
import org.warp.commonutils.error.InitializationException;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
+import reactor.core.publisher.SignalType;
import reactor.core.publisher.Sinks;
import reactor.core.publisher.Sinks.Many;
@@ -47,6 +48,7 @@ public class AsyncTdMiddleEventBusClient extends AbstractVerticle implements Asy
public static final boolean OUTPUT_REQUESTS = false;
public static final byte[] EMPTY = new byte[0];
+ private final Many tdCloseRequested = Sinks.many().replay().latestOrDefault(false);
private final Many tdClosed = Sinks.many().replay().latestOrDefault(false);
private final DeliveryOptions deliveryOptions;
private final DeliveryOptions deliveryOptionsWithTimeout;
@@ -172,9 +174,30 @@ public class AsyncTdMiddleEventBusClient extends AbstractVerticle implements Asy
@Override
public void stop(Promise stopPromise) {
+ logger.debug("Stopping AsyncTdMiddle client verticle...");
readyToStartConsumer.unregister(result -> {
- tdClosed.tryEmitNext(true);
- stopPromise.complete();
+ if (result.failed()) {
+ logger.error("Failed to unregister readyToStart consumer");
+ } else {
+ logger.debug("Unregistered readyToStart consumer");
+ }
+ tdCloseRequested.asFlux().take(1).single().flatMap(closeRequested -> {
+ if (!closeRequested) {
+ logger.warn("Verticle is being stopped before closing TDLib with Close()! Sending Close() before stopping...");
+ return this.execute(new TdApi.Close(), false).doOnTerminate(() -> {
+ logger.debug("Close() sent to td");
+ markCloseRequested();
+ }).then();
+ } else {
+ return Mono.empty();
+ }
+ }).thenMany(tdClosed.asFlux()).filter(closed -> closed).take(1).subscribe(v -> {}, cause -> {
+ logger.debug("Failed to stop AsyncTdMiddle client verticle");
+ stopPromise.fail(cause);
+ }, () -> {
+ logger.debug("Stopped AsyncTdMiddle client verticle");
+ stopPromise.complete();
+ });
});
}
@@ -204,7 +227,7 @@ public class AsyncTdMiddleEventBusClient extends AbstractVerticle implements Asy
@Override
public Flux receive() {
var fluxCodec = new TdResultListMessageCodec();
- return Mono.from(tdClosed.asFlux()).single().filter(tdClosed -> !tdClosed).flatMapMany(_closed -> EventBusFlux
+ return tdCloseRequested.asFlux().take(1).single().filter(close -> !close).flatMapMany(_closed -> EventBusFlux
.connect(cluster.getEventBus(),
botAddress + ".updates",
deliveryOptions,
@@ -235,12 +258,38 @@ public class AsyncTdMiddleEventBusClient extends AbstractVerticle implements Asy
if (item.getConstructor() == UpdateAuthorizationState.CONSTRUCTOR) {
var state = (UpdateAuthorizationState) item;
if (state.authorizationState.getConstructor() == AuthorizationStateClosed.CONSTRUCTOR) {
-
// Send tdClosed early to avoid errors
- tdClosed.tryEmitNext(true);
+ logger.debug("Received AuthorizationStateClosed from td. Marking td as closed");
+ markCloseRequested();
+ markClosed();
}
}
- }));
+ })).doFinally(s -> {
+ if (s == SignalType.ON_ERROR) {
+ // Send tdClosed early to avoid errors
+ logger.debug("Updates flux terminated with an error signal. Marking td as closed");
+ markCloseRequested();
+ markClosed();
+ }
+ });
+ }
+
+ private void markCloseRequested() {
+ if (tdCloseRequested.tryEmitNext(true).isFailure()) {
+ logger.error("Failed to set tdCloseRequested");
+ if (tdCloseRequested.tryEmitComplete().isFailure()) {
+ logger.error("Failed to complete tdCloseRequested");
+ }
+ }
+ }
+
+ private void markClosed() {
+ if (tdClosed.tryEmitNext(true).isFailure()) {
+ logger.error("Failed to set tdClosed");
+ if (tdClosed.tryEmitComplete().isFailure()) {
+ logger.error("Failed to complete tdClosed");
+ }
+ }
}
@Override
@@ -255,7 +304,7 @@ public class AsyncTdMiddleEventBusClient extends AbstractVerticle implements Asy
.replace(" = ", "="));
}
- return Mono.from(tdClosed.asFlux()).single().filter(tdClosed -> !tdClosed).>flatMap((_x) -> Mono.create(sink -> {
+ return tdCloseRequested.asFlux().take(1).single().filter(close -> !close).>flatMap((_x) -> Mono.create(sink -> {
try {
cluster
.getEventBus()
diff --git a/src/main/java/it/tdlight/tdlibsession/td/middle/direct/AsyncTdMiddleLocal.java b/src/main/java/it/tdlight/tdlibsession/td/middle/direct/AsyncTdMiddleLocal.java
index 8b73f20..e21a339 100644
--- a/src/main/java/it/tdlight/tdlibsession/td/middle/direct/AsyncTdMiddleLocal.java
+++ b/src/main/java/it/tdlight/tdlibsession/td/middle/direct/AsyncTdMiddleLocal.java
@@ -37,7 +37,7 @@ public class AsyncTdMiddleLocal implements AsyncTdMiddle {
try {
return AsyncTdMiddleEventBusClient.getAndDeployInstance(masterClusterManager, botAlias, botAddress, true).doOnNext(cli -> {
this.cli.onNext(cli);
- }).doOnError(error -> this.cli.onError(error)).doFinally(_v -> this.cli.onComplete());
+ }).doOnError(error -> this.cli.onError(error)).doOnSuccess(_v -> this.cli.onComplete());
} catch (InitializationException e) {
this.cli.onError(e);
return Mono.error(e);
diff --git a/src/main/java/it/tdlight/tdlibsession/td/middle/server/AsyncTdMiddleEventBusServer.java b/src/main/java/it/tdlight/tdlibsession/td/middle/server/AsyncTdMiddleEventBusServer.java
index 37fbbc4..daecdde 100644
--- a/src/main/java/it/tdlight/tdlibsession/td/middle/server/AsyncTdMiddleEventBusServer.java
+++ b/src/main/java/it/tdlight/tdlibsession/td/middle/server/AsyncTdMiddleEventBusServer.java
@@ -10,6 +10,10 @@ import io.vertx.core.eventbus.Message;
import io.vertx.core.eventbus.MessageConsumer;
import it.tdlight.common.ConstructorDetector;
import it.tdlight.jni.TdApi;
+import it.tdlight.jni.TdApi.AuthorizationStateClosed;
+import it.tdlight.jni.TdApi.Error;
+import it.tdlight.jni.TdApi.Update;
+import it.tdlight.jni.TdApi.UpdateAuthorizationState;
import it.tdlight.tdlibsession.EventBusFlux;
import it.tdlight.tdlibsession.td.TdResult;
import it.tdlight.tdlibsession.td.TdResultMessage;
@@ -27,39 +31,41 @@ import java.time.Duration;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
-import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
+import reactor.core.publisher.Sinks;
+import reactor.core.publisher.Sinks.EmitResult;
+import reactor.core.publisher.Sinks.Many;
+import reactor.core.publisher.Sinks.One;
+import reactor.util.function.Tuples;
public class AsyncTdMiddleEventBusServer {
+ // Static values
private static final Logger logger = LoggerFactory.getLogger(AsyncTdMiddleEventBusServer.class);
+ public static final byte[] EMPTY = new byte[0];
+ public static final Duration WAIT_DURATION = Duration.ofSeconds(1);
- private static final byte[] EMPTY = new byte[0];
- // todo: restore duration to 2 seconds instead of 10 millis, when the bug of tdlight double queue wait is fixed
- public static final Duration WAIT_DURATION = Duration.ofSeconds(1);// Duration.ofMillis(10);
- // If you enable this the poll will wait up to 100 additional milliseconds between each poll, if the server is remote
- private static final boolean ENABLE_MINIMUM_POLL_WAIT_INTERVAL = false;
-
+ // Values configured from constructor
private final TdClusterManager cluster;
private final AsyncTdDirectOptions tdOptions;
- private String botAlias;
- private String botAddress;
- private boolean local;
+ // Variables configured by the user at startup
+ private final One botAlias = Sinks.one();
+ private final One botAddress = Sinks.one();
+ private final One local = Sinks.one();
+ private final Many>> onBeforeStopListeners = Sinks.many().replay().all();
+ private final Many>> onAfterStopListeners = Sinks.many().replay().all();
- protected AsyncTdDirectImpl td;
- /**
- * Value is not important, emits when a request is received
- */
- private final List>> onBeforeStopListeners = new CopyOnWriteArrayList<>();
- private final List>> onAfterStopListeners = new CopyOnWriteArrayList<>();
- private MessageConsumer startConsumer;
- private MessageConsumer isWorkingConsumer;
- private MessageConsumer executeConsumer;
+ // Variables configured at startup
+ private final One td = Sinks.one();
+ private final One> startConsumer = Sinks.one();
+ private final One> isWorkingConsumer = Sinks.one();
+ private final One> executeConsumer = Sinks.one();
@SuppressWarnings({"unchecked", "rawtypes"})
public AsyncTdMiddleEventBusServer(TdClusterManager clusterManager) {
@@ -74,60 +80,139 @@ public class AsyncTdMiddleEventBusServer {
}
}
- public Mono start(String botAddress, String botAlias, boolean local) {
+ public Mono start(String botAddress, String botAlias, boolean local) {
return Mono.create(sink -> {
if (botAddress == null || botAddress.isEmpty()) {
sink.error(new IllegalArgumentException("botAddress is not set!"));
+ return;
+ }
+ if (this.botAddress.tryEmitValue(botAddress).isFailure()) {
+ sink.error(new IllegalStateException("Failed to set botAddress"));
+ return;
}
- this.botAddress = botAddress;
if (botAlias == null || botAlias.isEmpty()) {
sink.error(new IllegalArgumentException("botAlias is not set!"));
+ return;
+ }
+ if (this.botAlias.tryEmitValue(botAlias).isFailure()) {
+ sink.error(new IllegalStateException("Failed to set botAlias"));
+ return;
+ }
+ if (this.local.tryEmitValue(local).isFailure()) {
+ sink.error(new IllegalStateException("Failed to set local"));
+ return;
+ }
+ var td = new AsyncTdDirectImpl(botAlias);
+ if (this.td.tryEmitValue(td).isFailure()) {
+ sink.error(new IllegalStateException("Failed to set td instance"));
+ return;
+ }
+ if (this.onBeforeStopListeners.tryEmitComplete().isFailure()) {
+ sink.error(new IllegalStateException("Failed to finalize \"before stop\" listeners"));
+ return;
+ }
+ if (this.onAfterStopListeners.tryEmitComplete().isFailure()) {
+ sink.error(new IllegalStateException("Failed to finalize \"after stop\" listeners"));
+ return;
}
- this.botAlias = botAlias;
- this.local = local;
- this.td = new AsyncTdDirectImpl(botAlias);
AtomicBoolean alreadyDeployed = new AtomicBoolean(false);
- this.startConsumer = cluster.getEventBus().consumer(botAddress + ".start", (Message msg) -> {
- if (alreadyDeployed.compareAndSet(false, true)) {
- this.listen().then(this.pipe()).then(Mono.create(registrationSink -> {
- this.isWorkingConsumer = cluster.getEventBus().consumer(botAddress + ".isWorking", (Message workingMsg) -> {
- workingMsg.reply(EMPTY, cluster.newDeliveryOpts().setLocalOnly(local));
+ var startConsumer = cluster.getEventBus().consumer(botAddress + ".start");
+ if (this.startConsumer.tryEmitValue(startConsumer).isSuccess()) {
+ startConsumer.handler((Message msg) -> {
+ if (alreadyDeployed.compareAndSet(false, true)) {
+ startConsumer.unregister(startConsumerUnregistered -> {
+ if (startConsumerUnregistered.succeeded()) {
+ onSuccessfulStartRequest(msg, td, botAddress, botAlias, local);
+ } else {
+ logger.error("Failed to unregister start consumer");
+ }
});
- this.isWorkingConsumer.completionHandler(MonoUtils.toHandler(registrationSink));
- }))
- .subscribe(v -> {}, ex -> {
- logger.info(botAddress + " server deployed and started. succeeded: false");
- logger.error(ex.getLocalizedMessage(), ex);
- msg.fail(500, ex.getLocalizedMessage());
- }, () -> {
- logger.info(botAddress + " server deployed and started. succeeded: true");
- msg.reply(EMPTY);
- });
- } else {
- msg.reply(EMPTY);
- }
- });
- startConsumer.completionHandler(h -> {
- logger.info(botAddress + " server deployed. succeeded: " + h.succeeded());
- if (h.succeeded()) {
- logger.debug("Sending " + botAddress + ".readyToStart");
- cluster.getEventBus().request(botAddress + ".readyToStart", EMPTY, cluster.newDeliveryOpts().setSendTimeout(30000), msg -> {
- sink.success();
- });
- } else {
- sink.error(h.cause());
- }
- });
+ } else {
+ msg.reply(EMPTY);
+ }
+ });
+ startConsumer.completionHandler(h -> {
+ logger.info(botAddress + " server deployed. succeeded: " + h.succeeded());
+ if (h.succeeded()) {
+ var readyToStartOpts = cluster.newDeliveryOpts().setSendTimeout(30000);
+ logger.debug("Sending " + botAddress + ".readyToStart");
+ cluster.getEventBus().request(botAddress + ".readyToStart", EMPTY, readyToStartOpts, msg -> sink.success());
+ } else {
+ sink.error(h.cause());
+ }
+ });
+ } else {
+ sink.error(new IllegalStateException("Failed to set startConsumer"));
+ }
+ }).thenReturn(this);
+ }
+
+ private void onSuccessfulStartRequest(Message msg, AsyncTdDirectImpl td, String botAddress, String botAlias, boolean local) {
+ this.listen(td, botAddress, botAlias, local).then(this.pipe(td, botAddress, botAlias, local)).then(Mono.create(registrationSink -> {
+ var isWorkingConsumer = cluster.getEventBus().consumer(botAddress + ".isWorking");
+ if (this.isWorkingConsumer.tryEmitValue(isWorkingConsumer).isSuccess()) {
+ isWorkingConsumer.handler((Message workingMsg) -> {
+ workingMsg.reply(EMPTY, cluster.newDeliveryOpts().setLocalOnly(local));
+ });
+ isWorkingConsumer.completionHandler(MonoUtils.toHandler(registrationSink));
+ } else {
+ logger.error("Failed to set isWorkingConsumer");
+ msg.fail(500, "Failed to set isWorkingConsumer");
+ }
+ })).subscribe(v -> {}, ex -> {
+ logger.error("Deploy and start of bot \"" + botAlias + "\": ❌ Failed", ex);
+ msg.fail(500, ex.getLocalizedMessage());
+ }, () -> {
+ logger.info("Deploy and start of bot \"" + botAlias + "\": ✅ Succeeded");
+ msg.reply(EMPTY);
});
}
- public void onBeforeStop(Consumer> r) {
- this.onBeforeStopListeners.add(r);
+ /**
+ * Register a before stop listener
+ * @param eventListener listener
+ * @return success if the listener has been registered correctly
+ */
+ public EmitResult onBeforeStop(Consumer> eventListener, boolean crashOnFail) {
+ if (crashOnFail) {
+ return this.onBeforeStopListeners.tryEmitNext(eventListener);
+ } else {
+ return this.onBeforeStopListeners.tryEmitNext(promise -> {
+ Promise falliblePromise = Promise.promise();
+ falliblePromise.future().onComplete(result -> {
+ if (result.failed()) {
+ logger.warn("A beforeStop listener failed. Ignored error", result.cause());
+ }
+ promise.complete();
+ });
+ eventListener.accept(falliblePromise);
+
+ });
+ }
}
- public void onAfterStop(Consumer> r) {
- this.onAfterStopListeners.add(r);
+ /**
+ * Register an after stop listener
+ * @param eventListener listener
+ * @return success if the listener has been registered correctly
+ */
+ public EmitResult onAfterStop(Consumer> eventListener, boolean crashOnFail) {
+ if (crashOnFail) {
+ return this.onAfterStopListeners.tryEmitNext(eventListener);
+ } else {
+ return this.onAfterStopListeners.tryEmitNext(promise -> {
+ Promise falliblePromise = Promise.promise();
+ falliblePromise.future().onComplete(result -> {
+ if (result.failed()) {
+ logger.warn("An afterStop listener failed. Ignored error", result.cause());
+ }
+ promise.complete();
+ });
+ eventListener.accept(falliblePromise);
+
+ });
+ }
}
private void runAll(List>> actions, Handler> resultHandler) {
@@ -147,115 +232,160 @@ public class AsyncTdMiddleEventBusServer {
}
}
- private Mono listen() {
+ private Mono listen(AsyncTdDirectImpl td, String botAddress, String botAlias, boolean local) {
return Mono.create(registrationSink -> {
+ MessageConsumer executeConsumer = cluster.getEventBus().consumer(botAddress + ".execute");
+ if (this.executeConsumer.tryEmitValue(executeConsumer).isFailure()) {
+ registrationSink.error(new IllegalStateException("Failed to set executeConsumer"));
+ return;
+ }
- this.executeConsumer = cluster.getEventBus().consumer(botAddress + ".execute", (Message msg) -> {
- try {
+ Flux.>create(sink -> {
+ executeConsumer.handler(sink::next);
+ executeConsumer.completionHandler(MonoUtils.toHandler(registrationSink));
+ })
+ .doOnNext(msg -> {
+ if (OUTPUT_REQUESTS) {
+ System.out.println(":=> " + msg
+ .body()
+ .getRequest()
+ .toString()
+ .replace("\n", " ")
+ .replace("\t", "")
+ .replace(" ", "")
+ .replace(" = ", "="));
+ }
+ })
+ .flatMap(msg -> td
+ .execute(msg.body().getRequest(), msg.body().isExecuteDirectly())
+ .map(result -> Tuples.of(msg, result)))
+ .handle((tuple, sink) -> {
+ var msg = tuple.getT1();
+ var response = tuple.getT2();
+ var replyOpts = cluster.newDeliveryOpts().setLocalOnly(local);
+ var replyValue = new TdResultMessage(response.result(), response.cause());
+ try {
+ msg.reply(replyValue, replyOpts);
+ sink.next(response);
+ } catch (Exception ex) {
+ msg.fail(500, ex.getLocalizedMessage());
+ sink.error(ex);
+ }
+ })
+ .then()
+ .doOnError(ex -> {
+ logger.error("Error when processing a request", ex);
+ })
+ .subscribe();
+ });
+ }
+
+ private void undeploy(Runnable whenUndeployed) {
+ botAlias.asMono().single().flatMap(botAlias -> {
+ logger.info("Undeploy of bot \"" + botAlias + "\": stopping");
+ return onBeforeStopListeners.asFlux()
+ .collectList()
+ .single()
+ .flatMap(onBeforeStopListeners ->
+ Mono.create(sink -> runAll(onBeforeStopListeners, MonoUtils.toHandler(sink)))
+ .doOnError(ex -> logger.error("A beforeStop listener failed", ex)))
+ .then(Flux
+ .merge(unregisterConsumerOrLog(this.isWorkingConsumer.asMono(), "isWorkingConsumer"),
+ unregisterConsumerOrLog(this.startConsumer.asMono(), "isWorkingConsumer"),
+ unregisterConsumerOrLog(this.executeConsumer.asMono(), "isWorkingConsumer"))
+ .then())
+ .then(onAfterStopListeners.asFlux().collectList())
+ .single()
+ .doOnNext(onAfterStopListeners -> {
+ logger.info("TdMiddle verticle \"" + botAddress + "\" stopped");
+
+ runAll(onAfterStopListeners, onAfterStopHandler -> {
+ if (onAfterStopHandler.failed()) {
+ logger.error("An afterStop listener failed: " + onAfterStopHandler.cause());
+ }
+
+ logger.info("Undeploy of bot \"" + botAlias + "\": stopped");
+ whenUndeployed.run();
+ });
+ });
+ })
+ .doOnError(ex -> logger.error("Error when stopping", ex))
+ .subscribe();
+ }
+
+ private Mono unregisterConsumerOrLog(Mono> consumerMono, String consumerName) {
+ return consumerMono
+ .flatMap(consumer -> Mono
+ .create(sink -> consumer.unregister(MonoUtils.toHandler(sink)))
+ .onErrorResume(ex -> Mono.fromRunnable(() -> {
+ logger.error("Can't unregister consumer \"" + consumerName + "\"", ex);
+ })));
+ }
+
+ private Mono pipe(AsyncTdDirectImpl td, String botAddress, String botAlias, boolean local) {
+ var updatesFlux = td
+ .receive(tdOptions)
+ .doOnNext(update -> {
if (OUTPUT_REQUESTS) {
- System.out.println(":=> " + msg
- .body()
- .getRequest()
+ System.out.println("<=: " + update
.toString()
.replace("\n", " ")
.replace("\t", "")
.replace(" ", "")
.replace(" = ", "="));
}
- td
- .execute(msg.body().getRequest(), msg.body().isExecuteDirectly())
- .switchIfEmpty(Mono.fromSupplier(() -> {
- return TdResult.failed(new TdApi.Error(500, "Received null response"));
- }))
- .handle((response, sink) -> {
- try {
- msg.reply(new TdResultMessage(response.result(), response.cause()),
- cluster.newDeliveryOpts().setLocalOnly(local)
- );
- sink.next(response);
- } catch (Exception ex) {
- sink.error(ex);
+ })
+ .flatMap(item -> Mono.>create(sink -> {
+ if (item.succeeded()) {
+ var tdObject = item.result();
+ if (tdObject instanceof Update) {
+ var tdUpdate = (Update) tdObject;
+ if (tdUpdate.getConstructor() == UpdateAuthorizationState.CONSTRUCTOR) {
+ var tdUpdateAuthorizationState = (UpdateAuthorizationState) tdUpdate;
+ if (tdUpdateAuthorizationState.authorizationState.getConstructor()
+ == AuthorizationStateClosed.CONSTRUCTOR) {
+ logger.debug("Undeploying after receiving AuthorizationStateClosed");
+ this.undeploy(() -> sink.success(item));
+ return;
}
- })
- .subscribe(response -> {}, ex -> {
- logger.error("Error when processing a request", ex);
- msg.fail(500, ex.getLocalizedMessage());
- });
- } catch (ClassCastException ex) {
- logger.error("Error when deserializing a request", ex);
- msg.fail(500, ex.getMessage());
- }
- });
- executeConsumer.completionHandler(MonoUtils.toHandler(registrationSink));
- });
- }
-
- private void undeploy(Runnable whenUndeployed) {
- runAll(onBeforeStopListeners, onBeforeStopHandler -> {
- if (onBeforeStopHandler.failed()) {
- logger.error("A beforeStop listener failed: "+ onBeforeStopHandler.cause());
- }
-
- Mono.create(sink -> this.isWorkingConsumer.unregister(result -> {
- if (result.failed()) {
- logger.error("Can't unregister consumer", result.cause());
- }
- this.startConsumer.unregister(result2 -> {
- if (result2.failed()) {
- logger.error("Can't unregister consumer", result2.cause());
- }
-
- this.executeConsumer.unregister(result4 -> {
- if (result4.failed()) {
- logger.error("Can't unregister consumer", result4.cause());
+ }
+ } else if (tdObject instanceof Error) {
+ // An error in updates means that a fatal error occurred
+ logger.debug("Undeploying after receiving a fatal error");
+ this.undeploy(() -> sink.success(item));
+ return;
}
-
- sink.success();
- });
- });
- })).doFinally(signalType -> {
- logger.info("TdMiddle verticle \"" + botAddress + "\" stopped");
-
- runAll(onAfterStopListeners, onAfterStopHandler -> {
- if (onAfterStopHandler.failed()) {
- logger.error("An afterStop listener failed: " + onAfterStopHandler.cause());
+ sink.success(item);
}
-
- whenUndeployed.run();
- });
- }).subscribe(v -> {}, ex -> {
- logger.error("Error when stopping", ex);
- }, () -> {});
- });
- }
-
- private Mono pipe() {
- var updatesFlux = td.receive(tdOptions).doOnNext(update -> {
- if (OUTPUT_REQUESTS) {
- System.out.println("<=: " + update
- .toString()
- .replace("\n", " ")
- .replace("\t", "")
- .replace(" ", "")
- .replace(" = ", "="));
- }
- }).bufferTimeout(tdOptions.getEventsSize(), local ? Duration.ofMillis(1) : Duration.ofMillis(100))
- .filter(l -> !l.isEmpty())
+ }))
+ .bufferTimeout(tdOptions.getEventsSize(), local ? Duration.ofMillis(1) : Duration.ofMillis(100))
.windowTimeout(1, Duration.ofSeconds(5))
.flatMap(w -> w.defaultIfEmpty(Collections.emptyList()))
- .map(TdResultList::new).doFinally(s -> {
+ .map(TdResultList::new).doOnTerminate(() -> {
if (OUTPUT_REQUESTS) {
System.out.println("<=: end (3)");
}
- this.undeploy(() -> {});
});
var fluxCodec = new TdResultListMessageCodec();
- return EventBusFlux.serve(updatesFlux,
+ var tuple = EventBusFlux.serve(updatesFlux,
cluster.getEventBus(),
botAddress + ".updates",
cluster.newDeliveryOpts().setLocalOnly(local),
fluxCodec,
Duration.ofSeconds(30)
);
+ var served = tuple.getT1();
+ var fatalError = tuple.getT2();
+ //noinspection CallingSubscribeInNonBlockingScope
+ fatalError
+ .doOnNext(e -> logger.warn("Undeploying after a fatal error in a served flux"))
+ .flatMap(error -> td.execute(new TdApi.Close(), false))
+ .doOnError(ex -> logger.error("Unexpected error", ex))
+ .subscribe();
+ return served.doOnSubscribe(s -> {
+ logger.debug("Preparing to serve bot \"" + botAlias + "\" updates flux...");
+ }).doOnSuccess(v -> {
+ logger.debug("Ready to serve bot \"" + botAlias + "\" updates flux");
+ });
}
}
diff --git a/src/main/java/it/tdlight/utils/MonoUtils.java b/src/main/java/it/tdlight/utils/MonoUtils.java
index 9648479..4b566a1 100644
--- a/src/main/java/it/tdlight/utils/MonoUtils.java
+++ b/src/main/java/it/tdlight/utils/MonoUtils.java
@@ -8,7 +8,6 @@ import it.tdlight.jni.TdApi;
import it.tdlight.jni.TdApi.Object;
import it.tdlight.tdlibsession.td.TdError;
import it.tdlight.tdlibsession.td.TdResult;
-import it.tdlight.tdlibsession.td.middle.direct.AsyncTdMiddleDirect;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.function.BiConsumer;
diff --git a/src/main/resources/tdlib-session-container-log4j2.xml b/src/main/resources/tdlib-session-container-log4j2.xml
index 922d77c..5d083ff 100644
--- a/src/main/resources/tdlib-session-container-log4j2.xml
+++ b/src/main/resources/tdlib-session-container-log4j2.xml
@@ -7,7 +7,7 @@
.
-->
-
+
@@ -27,7 +27,7 @@
-
+