+ * Many operations in Vert.x APIs provide results back by passing an instance of this in a {@link io.vertx.core.Handler}.
+ *
+ * The result can either have failed or succeeded.
+ *
+ * If it failed then the cause of the failure is available with {@link #cause}.
+ *
+ * If it succeeded then the actual result is available with {@link #result}
+ *
+ * @author Tim Fox
+ */
+public interface TdResult {
+
+ /**
+ * The result of the operation. This will be null if the operation failed.
+ *
+ * @return the result or null if the operation failed.
+ */
+ T result();
+
+ /**
+ * The result of the operation. This will throw CompletionException if the operation failed.
+ *
+ * @return the result.
+ */
+ T orElseThrow() throws CompletionException;
+
+ /**
+ * A TdApi.Error describing failure. This will be null if the operation succeeded.
+ *
+ * @return the cause or null if the operation succeeded.
+ */
+ TdApi.Error cause();
+
+ /**
+ * Did it succeed?
+ *
+ * @return true if it succeded or false otherwise
+ */
+ boolean succeeded();
+
+ /**
+ * Did it fail?
+ *
+ * @return true if it failed or false otherwise
+ */
+ boolean failed();
+
+ /**
+ * Apply a {@code mapper} function on this async result.
+ *
+ * The {@code mapper} is called with the completed value and this mapper returns a value. This value will complete the result returned by this method call.
+ *
+ * When this async result is failed, the failure will be propagated to the returned async result and the {@code mapper} will not be called.
+ *
+ * @param mapper the mapper function
+ * @return the mapped async result
+ */
+ default TdResult map(Function mapper) {
+ if (mapper == null) {
+ throw new NullPointerException();
+ }
+ return new TdResult() {
+ @Override
+ public U result() {
+ if (succeeded()) {
+ return mapper.apply(TdResult.this.result());
+ } else {
+ return null;
+ }
+ }
+
+ @Override
+ public U orElseThrow() throws CompletionException {
+ if (succeeded()) {
+ return mapper.apply(TdResult.this.orElseThrow());
+ } else {
+ return null;
+ }
+ }
+
+ @Override
+ public TdApi.Error cause() {
+ return TdResult.this.cause();
+ }
+
+ @Override
+ public boolean succeeded() {
+ return TdResult.this.succeeded();
+ }
+
+ @Override
+ public boolean failed() {
+ return TdResult.this.failed();
+ }
+ };
+ }
+
+ /**
+ * Map the result of this async result to a specific {@code value}.
+ *
+ * When this async result succeeds, this {@code value} will succeeed the async result returned by this method call.
+ *
+ * When this async result fails, the failure will be propagated to the returned async result.
+ *
+ * @param value the value that eventually completes the mapped async result
+ * @return the mapped async result
+ */
+ default TdResult map(V value) {
+ return map((Function) t -> value);
+ }
+
+ /**
+ * Map the result of this async result to {@code null}.
+ *
+ * This is a convenience for {@code TdResult.map((T) null)} or {@code TdResult.map((Void) null)}.
+ *
+ * When this async result succeeds, {@code null} will succeeed the async result returned by this method call.
+ *
+ * When this async result fails, the failure will be propagated to the returned async result.
+ *
+ * @return the mapped async result
+ */
+ default TdResult mapEmpty() {
+ return map((V)null);
+ }
+
+ /**
+ * Apply a {@code mapper} function on this async result.
+ *
+ * The {@code mapper} is called with the failure and this mapper returns a value. This value will complete the result returned by this method call.
+ *
+ * When this async result is succeeded, the value will be propagated to the returned async result and the {@code mapper} will not be called.
+ *
+ * @param mapper the mapper function
+ * @return the mapped async result
+ */
+ default TdResult otherwise(Function mapper) {
+ if (mapper == null) {
+ throw new NullPointerException();
+ }
+ return new TdResult() {
+ @Override
+ public T result() {
+ if (TdResult.this.succeeded()) {
+ return TdResult.this.result();
+ } else if (TdResult.this.failed()) {
+ return mapper.apply(TdResult.this.cause());
+ } else {
+ return null;
+ }
+ }
+ @Override
+ public T orElseThrow() {
+ if (TdResult.this.succeeded()) {
+ return TdResult.this.orElseThrow();
+ } else if (TdResult.this.failed()) {
+ return mapper.apply(TdResult.this.cause());
+ } else {
+ return null;
+ }
+ }
+
+ @Override
+ public TdApi.Error cause() {
+ return null;
+ }
+
+ @Override
+ public boolean succeeded() {
+ return TdResult.this.succeeded() || TdResult.this.failed();
+ }
+
+ @Override
+ public boolean failed() {
+ return false;
+ }
+ };
+ }
+
+ /**
+ * Map the failure of this async result to a specific {@code value}.
+ *
+ * When this async result fails, this {@code value} will succeeed the async result returned by this method call.
+ *
+ * When this async succeeds, the result will be propagated to the returned async result.
+ *
+ * @param value the value that eventually completes the mapped async result
+ * @return the mapped async result
+ */
+ default TdResult otherwise(T value) {
+ return otherwise(err -> value);
+ }
+
+ /**
+ * Map the failure of this async result to {@code null}.
+ *
+ * This is a convenience for {@code TdResult.otherwise((T) null)}.
+ *
+ * When this async result fails, the {@code null} will succeeed the async result returned by this method call.
+ *
+ * When this async succeeds, the result will be propagated to the returned async result.
+ *
+ * @return the mapped async result
+ */
+ default TdResult otherwiseEmpty() {
+ return otherwise(err -> null);
+ }
+
+ static TdResult succeeded(@NotNull T value) {
+ return new TdResultImpl(value, null);
+ }
+
+ static TdResult failed(@NotNull TdApi.Error error) {
+ return new TdResultImpl(null, error);
+ }
+
+ static TdResult of(@NotNull TdApi.Object resultOrError) {
+ if (resultOrError.getConstructor() == TdApi.Error.CONSTRUCTOR) {
+ return failed((TdApi.Error) resultOrError);
+ } else {
+ //noinspection unchecked
+ return succeeded((T) resultOrError);
+ }
+ }
+
+ class TdResultImpl implements TdResult {
+
+ private final U value;
+ private final Error error;
+
+ public TdResultImpl(U value, Error error) {
+ this.value = value;
+ this.error = error;
+
+ assert (value == null) != (error == null);
+ }
+
+ @Override
+ public U result() {
+ return value;
+ }
+
+ @Override
+ public U orElseThrow() {
+ if (error != null) {
+ throw new TdError(error.code, error.message);
+ }
+ return value;
+ }
+
+ @Override
+ public Error cause() {
+ return error;
+ }
+
+ @Override
+ public boolean succeeded() {
+ return value != null;
+ }
+
+ @Override
+ public boolean failed() {
+ return error != null;
+ }
+ }
+}
diff --git a/src/main/java/it/tdlight/tdlibsession/td/TdResultMessage.java b/src/main/java/it/tdlight/tdlibsession/td/TdResultMessage.java
new file mode 100644
index 0000000..84fadd1
--- /dev/null
+++ b/src/main/java/it/tdlight/tdlibsession/td/TdResultMessage.java
@@ -0,0 +1,24 @@
+package it.tdlight.tdlibsession.td;
+
+import it.tdlight.jni.TdApi;
+import it.tdlight.jni.TdApi.Error;
+import it.tdlight.jni.TdApi.Object;
+
+public class TdResultMessage {
+ public final TdApi.Object value;
+ public final TdApi.Error cause;
+
+ public TdResultMessage(Object value, Error cause) {
+ this.value = value;
+ this.cause = cause;
+ }
+
+ public TdResult toTdResult() {
+ if (value != null) {
+ //noinspection unchecked
+ return TdResult.succeeded((T) value);
+ } else {
+ return TdResult.failed(cause);
+ }
+ }
+}
diff --git a/src/main/java/it/tdlight/tdlibsession/td/direct/AsyncTdDirect.java b/src/main/java/it/tdlight/tdlibsession/td/direct/AsyncTdDirect.java
new file mode 100644
index 0000000..fe9c0f4
--- /dev/null
+++ b/src/main/java/it/tdlight/tdlibsession/td/direct/AsyncTdDirect.java
@@ -0,0 +1,43 @@
+package it.tdlight.tdlibsession.td.direct;
+
+import io.vertx.core.AsyncResult;
+import it.tdlight.jni.TdApi;
+import it.tdlight.jni.TdApi.Function;
+import it.tdlight.jni.TdApi.Update;
+import it.tdlight.tdlibsession.td.TdResult;
+import java.time.Duration;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+public interface AsyncTdDirect {
+
+ /**
+ * Receives incoming updates and request responses from TDLib. May be called from any thread, but
+ * shouldn't be called simultaneously from two different threads.
+ *
+ * @param receiveDuration Maximum number of seconds allowed for this function to wait for new records. Default: 1 sec
+ * @param eventsSize Maximum number of events allowed in list. Default: 350 events
+ * @return An incoming update or request response list. The object returned in the response may be
+ * an empty list if the timeout expires.
+ */
+ Flux>> getUpdates(Duration receiveDuration, int eventsSize);
+
+ /**
+ * Sends request to TDLib. May be called from any thread.
+ *
+ * @param request Request to TDLib.
+ * @param synchronous Execute synchronously.
+ * @return The request response or {@link it.tdlight.jni.TdApi.Error}.
+ */
+ Mono> execute(Function request, boolean synchronous);
+
+ /**
+ * Initializes the client and TDLib instance.
+ */
+ Mono initializeClient();
+
+ /**
+ * Destroys the client and TDLib instance.
+ */
+ Mono destroyClient();
+}
diff --git a/src/main/java/it/tdlight/tdlibsession/td/direct/AsyncTdDirectImpl.java b/src/main/java/it/tdlight/tdlibsession/td/direct/AsyncTdDirectImpl.java
new file mode 100644
index 0000000..7866650
--- /dev/null
+++ b/src/main/java/it/tdlight/tdlibsession/td/direct/AsyncTdDirectImpl.java
@@ -0,0 +1,113 @@
+package it.tdlight.tdlibsession.td.direct;
+
+import io.vertx.core.AsyncResult;
+import io.vertx.core.Future;
+import it.tdlight.common.TelegramClient;
+import it.tdlight.jni.TdApi;
+import it.tdlight.jni.TdApi.AuthorizationStateClosed;
+import it.tdlight.jni.TdApi.Function;
+import it.tdlight.jni.TdApi.Object;
+import it.tdlight.jni.TdApi.Update;
+import it.tdlight.jni.TdApi.UpdateAuthorizationState;
+import it.tdlight.tdlibsession.td.TdResult;
+import it.tdlight.tdlight.ClientManager;
+import java.time.Duration;
+import java.util.concurrent.atomic.AtomicReference;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import reactor.core.publisher.EmitterProcessor;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Scheduler;
+import reactor.core.scheduler.Schedulers;
+
+public class AsyncTdDirectImpl implements AsyncTdDirect {
+
+ private static final Logger logger = LoggerFactory.getLogger(AsyncTdDirect.class);
+
+ private final AtomicReference td = new AtomicReference<>();
+ private final Scheduler tdScheduler = Schedulers.newSingle("TdMain");
+ private final Scheduler tdPollScheduler = Schedulers.newSingle("TdPoll");
+ private final Scheduler tdUpdatesScheduler = Schedulers.newSingle("TdUpdate");
+ private final Scheduler tdResponsesScheduler = Schedulers.newSingle("TdResponse");
+
+ private final EmitterProcessor>> updatesProcessor = EmitterProcessor.create();
+ private final String botAlias;
+
+ public AsyncTdDirectImpl(String botAlias) {
+ this.botAlias = botAlias;
+ }
+
+ @Override
+ public Mono> execute(Function request, boolean synchronous) {
+ if (synchronous) {
+ return Mono.just(TdResult.of(this.td.get().execute(request)));
+ } else {
+ return Mono.>create(sink -> {
+ try {
+ this.td.get().send(request, v -> {
+ sink.success(TdResult.of(v));
+ }, sink::error);
+ } catch (Throwable t) {
+ sink.error(t);
+ }
+ }).subscribeOn(tdResponsesScheduler);
+ }
+ }
+
+ @Override
+ public Flux>> getUpdates(Duration receiveDuration, int eventsSize) {
+ return Flux.from(updatesProcessor.subscribeOn(tdUpdatesScheduler));
+ }
+
+ public Scheduler getTdUpdatesScheduler() {
+ return tdUpdatesScheduler;
+ }
+
+ public Scheduler getTdResponsesScheduler() {
+ return tdResponsesScheduler;
+ }
+
+ @Override
+ public Mono initializeClient() {
+ return Mono.create(sink -> {
+ Flux.>>create(emitter -> {
+ var client = ClientManager.create((Object object) -> {
+ emitter.next(Future.succeededFuture(TdResult.of(object)));
+ // Close the emitter if receive closed state
+ if (object.getConstructor() == UpdateAuthorizationState.CONSTRUCTOR
+ && ((UpdateAuthorizationState) object).authorizationState.getConstructor()
+ == AuthorizationStateClosed.CONSTRUCTOR) {
+ emitter.complete();
+ }
+ }, updateError -> {
+ emitter.next(Future.failedFuture(updateError));
+ }, error -> {
+ emitter.next(Future.failedFuture(error));
+ });
+ this.td.set(client);
+
+ emitter.onDispose(() -> {
+ this.td.set(null);
+ });
+ }).subscribeOn(tdPollScheduler).subscribe(next -> {
+ updatesProcessor.onNext(next);
+ sink.success(true);
+ }, error -> {
+ updatesProcessor.onError(error);
+ sink.error(error);
+ }, () -> {
+ updatesProcessor.onComplete();
+ sink.success(true);
+ });
+ }).single().then().subscribeOn(tdScheduler);
+ }
+
+ @Override
+ public Mono destroyClient() {
+ return Mono.fromCallable(() -> {
+ // do nothing
+ return (Void) null;
+ }).single().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
new file mode 100644
index 0000000..24a2280
--- /dev/null
+++ b/src/main/java/it/tdlight/tdlibsession/td/easy/AsyncTdEasy.java
@@ -0,0 +1,460 @@
+package it.tdlight.tdlibsession.td.easy;
+
+import it.tdlight.common.utils.ScannerUtils;
+import it.tdlight.jni.TdApi;
+import it.tdlight.jni.TdApi.AuthorizationState;
+import it.tdlight.jni.TdApi.AuthorizationStateClosed;
+import it.tdlight.jni.TdApi.AuthorizationStateClosing;
+import it.tdlight.jni.TdApi.AuthorizationStateReady;
+import it.tdlight.jni.TdApi.AuthorizationStateWaitCode;
+import it.tdlight.jni.TdApi.AuthorizationStateWaitEncryptionKey;
+import it.tdlight.jni.TdApi.AuthorizationStateWaitPassword;
+import it.tdlight.jni.TdApi.AuthorizationStateWaitPhoneNumber;
+import it.tdlight.jni.TdApi.AuthorizationStateWaitRegistration;
+import it.tdlight.jni.TdApi.AuthorizationStateWaitTdlibParameters;
+import it.tdlight.jni.TdApi.CheckAuthenticationBotToken;
+import it.tdlight.jni.TdApi.CheckAuthenticationPassword;
+import it.tdlight.jni.TdApi.CheckDatabaseEncryptionKey;
+import it.tdlight.jni.TdApi.Error;
+import it.tdlight.jni.TdApi.Object;
+import it.tdlight.jni.TdApi.OptionValue;
+import it.tdlight.jni.TdApi.OptionValueBoolean;
+import it.tdlight.jni.TdApi.OptionValueEmpty;
+import it.tdlight.jni.TdApi.OptionValueInteger;
+import it.tdlight.jni.TdApi.OptionValueString;
+import it.tdlight.jni.TdApi.PhoneNumberAuthenticationSettings;
+import it.tdlight.jni.TdApi.RegisterUser;
+import it.tdlight.jni.TdApi.SetAuthenticationPhoneNumber;
+import it.tdlight.jni.TdApi.SetTdlibParameters;
+import it.tdlight.jni.TdApi.TdlibParameters;
+import it.tdlight.jni.TdApi.Update;
+import it.tdlight.jni.TdApi.UpdateAuthorizationState;
+import it.tdlight.tdlibsession.FatalErrorType;
+import it.tdlight.tdlibsession.td.TdResult;
+import it.tdlight.tdlibsession.td.middle.AsyncTdMiddle;
+import it.tdlight.utils.MonoUtils;
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Comparator;
+import java.util.Set;
+import org.reactivestreams.Publisher;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.warp.commonutils.error.InitializationException;
+import reactor.core.publisher.EmitterProcessor;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.core.publisher.ReplayProcessor;
+import reactor.core.scheduler.Schedulers;
+
+public class AsyncTdEasy {
+
+ private static final Logger logger = LoggerFactory.getLogger(AsyncTdEasy.class);
+
+ private final ReplayProcessor authState = ReplayProcessor.cacheLastOrDefault(new AuthorizationStateClosed());
+ private final ReplayProcessor requestedDefinitiveExit = ReplayProcessor.cacheLastOrDefault(false);
+ private final ReplayProcessor settings = ReplayProcessor.cacheLast();
+ private final EmitterProcessor globalErrors = EmitterProcessor.create();
+ private final EmitterProcessor fatalErrors = EmitterProcessor.create();
+ private final AsyncTdMiddle td;
+ private final String logName;
+ private final Flux incomingUpdatesCo;
+
+ public AsyncTdEasy(AsyncTdMiddle td, String logName) {
+ this.td = td;
+ this.logName = logName;
+
+ var sch = Schedulers.newSingle("TdEasyUpdates");
+
+ // todo: use Duration.ZERO instead of 10ms interval
+ this.incomingUpdatesCo = td.getUpdates()
+ .filterWhen(update -> Mono.from(requestedDefinitiveExit).map(requestedDefinitiveExit -> !requestedDefinitiveExit))
+ .subscribeOn(sch)
+ .publishOn(sch)
+ .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());
+ })
+ .publish().refCount(1);
+ }
+
+ public Mono create(TdEasySettings settings) {
+ return Mono
+ .fromCallable(() -> {
+ // Create session directories
+ if (Files.notExists(Path.of(settings.databaseDirectory))) {
+ try {
+ Files.createDirectories(Path.of(settings.databaseDirectory));
+ } catch (IOException ex) {
+ throw new InitializationException(ex);
+ }
+ }
+ return true;
+ })
+ .subscribeOn(Schedulers.boundedElastic())
+ .flatMap(_v -> {
+ this.settings.onNext(settings);
+ return Mono.empty();
+ });
+ }
+
+ /**
+ * Get TDLib state
+ */
+ public Flux getState() {
+ return Flux.from(authState);
+ }
+
+ /**
+ * Get incoming updates from TDLib.
+ */
+ public Flux getIncomingUpdates() {
+ return getIncomingUpdates(false);
+ }
+
+ private Flux getIncomingUpdates(boolean includePreAuthUpdates) {
+ return Flux.from(incomingUpdatesCo);
+ }
+
+ /**
+ * Get generic error updates from TDLib (When they are not linked to a precise request).
+ */
+ public Flux getIncomingErrors() {
+ return Flux.from(globalErrors);
+ }
+
+ /**
+ * Receives fatal errors from TDLib.
+ */
+ public Flux getFatalErrors() {
+ return Flux.from(fatalErrors);
+ }
+
+ /**
+ * Sends request to TDLib.
+ * @return The response or {@link TdApi.Error}.
+ */
+ public Mono> send(TdApi.Function request) {
+ return td.execute(request, false);
+ }
+
+ private Mono> sendDirectly(TdApi.Function obj) {
+ return td.execute(obj, false);
+ }
+
+ /**
+ * Set verbosity level
+ * @param i level
+ */
+ public Mono setVerbosityLevel(int i) {
+ return sendDirectly(new TdApi.SetLogVerbosityLevel(i)).then();
+ }
+
+ /**
+ * Clear option on TDLib
+ * @param name option name
+ */
+ public Mono clearOption(String name) {
+ return sendDirectly(new TdApi.SetOption(name, new TdApi.OptionValueEmpty())).then();
+ }
+
+ /**
+ * Set option on TDLib
+ * @param name option name
+ * @param value option value
+ */
+ public Mono setOptionString(String name, String value) {
+ return sendDirectly(new TdApi.SetOption(name, new TdApi.OptionValueString(value))).then();
+ }
+
+ /**
+ * Set option on TDLib
+ * @param name option name
+ * @param value option value
+ */
+ public Mono setOptionInteger(String name, long value) {
+ return sendDirectly(new TdApi.SetOption(name, new TdApi.OptionValueInteger(value))).then();
+ }
+
+ /**
+ * Set option on TDLib
+ * @param name option name
+ * @param value option value
+ */
+ public Mono setOptionBoolean(String name, boolean value) {
+ return sendDirectly(new TdApi.SetOption(name, new TdApi.OptionValueBoolean(value))).then();
+ }
+
+ /**
+ * Get option from TDLib
+ * @param name option name
+ * @return The value or nothing
+ */
+ public Mono getOptionString(String name) {
+ return this.sendDirectly(new TdApi.GetOption(name)).handle(MonoUtils::orElseThrow).flatMap((TdApi.OptionValue value) -> {
+ switch (value.getConstructor()) {
+ case OptionValueString.CONSTRUCTOR:
+ return Mono.just(((OptionValueString) value).value);
+ case OptionValueEmpty.CONSTRUCTOR:
+ return Mono.empty();
+ default:
+ return Mono.error(new UnsupportedOperationException("The option " + name + " is of type "
+ + value.getClass().getSimpleName()));
+ }
+ });
+ }
+
+ /**
+ * Get option from TDLib
+ * @param name option name
+ * @return The value or nothing
+ */
+ public Mono getOptionInteger(String name) {
+ return this.sendDirectly(new TdApi.GetOption(name)).handle(MonoUtils::orElseThrow).flatMap((TdApi.OptionValue value) -> {
+ switch (value.getConstructor()) {
+ case OptionValueInteger.CONSTRUCTOR:
+ return Mono.just(((OptionValueInteger) value).value);
+ case OptionValueEmpty.CONSTRUCTOR:
+ return Mono.empty();
+ default:
+ return Mono.error(new UnsupportedOperationException("The option " + name + " is of type "
+ + value.getClass().getSimpleName()));
+ }
+ });
+ }
+
+ /**
+ * Get option from TDLib
+ * @param name option name
+ * @return The value or nothing
+ */
+ public Mono getOptionBoolean(String name) {
+ return this.sendDirectly(new TdApi.GetOption(name)).handle(MonoUtils::orElseThrow).flatMap((TdApi.OptionValue value) -> {
+ switch (value.getConstructor()) {
+ case OptionValueBoolean.CONSTRUCTOR:
+ return Mono.just(((OptionValueBoolean) value).value);
+ case OptionValueEmpty.CONSTRUCTOR:
+ return Mono.empty();
+ default:
+ return Mono.error(new UnsupportedOperationException("The option " + name + " is of type "
+ + value.getClass().getSimpleName()));
+ }
+ });
+ }
+
+ /**
+ * Synchronously executes TDLib requests. Only a few requests can be executed synchronously. May
+ * be called from any thread.
+ *
+ * @param request Request to the TDLib.
+ * @return The request response.
+ */
+ public Mono> execute(TdApi.Function request) {
+ return td.execute(request, true);
+ }
+
+ /**
+ * Set if skip updates or not
+ */
+ public Mono setSkipUpdates(boolean skipUpdates) { //todo: do this
+ return null;
+ }
+
+ /**
+ * Closes the client gracefully by sending {@link TdApi.Close}.
+ */
+ public Mono close() {
+ return Mono.from(getState())
+ .filter(state -> {
+ switch (state.getConstructor()) {
+ case AuthorizationStateClosing.CONSTRUCTOR:
+ case AuthorizationStateClosed.CONSTRUCTOR:
+ return false;
+ default:
+ return true;
+ }
+ })
+ .then(Mono.from(requestedDefinitiveExit).single())
+ .filter(closeRequested -> !closeRequested)
+ .doOnSuccess(v -> requestedDefinitiveExit.onNext(true))
+ .then(td.execute(new TdApi.Close(), false))
+ .then();
+ }
+
+ /**
+ *
+ * @param timeout Timeout in seconds when reading data
+ */
+ public void setReadTimeout(int timeout) {
+ //todo: do this
+ }
+
+ /**
+ *
+ * @param timeout Timeout in seconds when listening methods or connecting
+ */
+ public void setMethodTimeout(int timeout) {
+ //todo: do this
+ }
+
+ private Mono catchErrors(Object obj) {
+ if (obj.getConstructor() == Error.CONSTRUCTOR) {
+ var error = (Error) obj;
+
+ switch (error.message) {
+ case "PHONE_CODE_INVALID":
+ globalErrors.onNext(error);
+ return Mono.just(new AuthorizationStateWaitCode());
+ case "PASSWORD_HASH_INVALID":
+ globalErrors.onNext(error);
+ return Mono.just(new AuthorizationStateWaitPassword());
+ case "PHONE_NUMBER_INVALID":
+ fatalErrors.onNext(FatalErrorType.PHONE_NUMBER_INVALID);
+ break;
+ case "ACCESS_TOKEN_INVALID":
+ fatalErrors.onNext(FatalErrorType.ACCESS_TOKEN_INVALID);
+ break;
+ case "CONNECTION_KILLED":
+ fatalErrors.onNext(FatalErrorType.CONNECTION_KILLED);
+ break;
+ default:
+ globalErrors.onNext(error);
+ break;
+ }
+ return Mono.empty();
+ }
+ return Mono.just(obj);
+ }
+
+ public Mono isBot() {
+ return Mono.from(settings).single().map(TdEasySettings::isBotTokenSet);
+ }
+
+ private Publisher preprocessUpdates(Update updateObj) {
+ return Mono
+ .just(updateObj)
+ .flatMap(this::catchErrors)
+ .filter(obj -> obj.getConstructor() == UpdateAuthorizationState.CONSTRUCTOR)
+ .map(obj -> ((UpdateAuthorizationState) obj).authorizationState)
+ .flatMap(obj -> {
+ this.authState.onNext(new AuthorizationStateReady());
+ switch (obj.getConstructor()) {
+ case AuthorizationStateWaitTdlibParameters.CONSTRUCTOR:
+ return Mono.from(this.settings).map(settings -> {
+ var parameters = new TdlibParameters();
+ parameters.useTestDc = settings.useTestDc;
+ parameters.databaseDirectory = settings.databaseDirectory;
+ parameters.filesDirectory = settings.filesDirectory;
+ parameters.useFileDatabase = settings.useFileDatabase;
+ parameters.useChatInfoDatabase = settings.useChatInfoDatabase;
+ parameters.useMessageDatabase = settings.useMessageDatabase;
+ parameters.useSecretChats = false;
+ parameters.apiId = settings.apiId;
+ parameters.apiHash = settings.apiHash;
+ parameters.systemLanguageCode = settings.systemLanguageCode;
+ parameters.deviceModel = settings.deviceModel;
+ parameters.systemVersion = settings.systemVersion;
+ parameters.applicationVersion = settings.applicationVersion;
+ parameters.enableStorageOptimizer = settings.enableStorageOptimizer;
+ parameters.ignoreFileNames = settings.ignoreFileNames;
+ return new SetTdlibParameters(parameters);
+ }).flatMap(this::sendDirectly).then();
+ case AuthorizationStateWaitEncryptionKey.CONSTRUCTOR:
+ return sendDirectly(new CheckDatabaseEncryptionKey()).then();
+ case AuthorizationStateWaitPhoneNumber.CONSTRUCTOR:
+ return Mono.from(this.settings).flatMap(settings -> {
+ if (settings.isPhoneNumberSet()) {
+ return sendDirectly(new SetAuthenticationPhoneNumber(String.valueOf(settings.getPhoneNumber()),
+ new PhoneNumberAuthenticationSettings(false, false, false)
+ ));
+ } else if (settings.isBotTokenSet()) {
+ return sendDirectly(new CheckAuthenticationBotToken(settings.getBotToken()));
+ } else {
+ return Mono.error(new IllegalArgumentException("A bot is neither an user or a bot"));
+ }
+ }).then();
+ case AuthorizationStateWaitRegistration.CONSTRUCTOR:
+ var authorizationStateWaitRegistration = (AuthorizationStateWaitRegistration) obj;
+ RegisterUser registerUser = new RegisterUser();
+ if (authorizationStateWaitRegistration.termsOfService != null
+ && authorizationStateWaitRegistration.termsOfService.text != null && !authorizationStateWaitRegistration.termsOfService.text.text.isBlank()) {
+ logger.info("Telegram Terms of Service:\n" + authorizationStateWaitRegistration.termsOfService.text.text);
+ }
+
+ while (registerUser.firstName == null || registerUser.firstName.length() <= 0
+ || registerUser.firstName.length() > 64 || registerUser.firstName.isBlank()) {
+ registerUser.firstName = ScannerUtils.askParameter(this.logName, "Enter First Name").trim();
+ }
+ while (registerUser.lastName == null || registerUser.firstName.length() > 64) {
+ registerUser.lastName = ScannerUtils.askParameter(this.logName, "Enter Last Name").trim();
+ }
+
+ return sendDirectly(registerUser).then();
+ case AuthorizationStateWaitPassword.CONSTRUCTOR:
+ var authorizationStateWaitPassword = (AuthorizationStateWaitPassword) obj;
+ String passwordMessage = "Password authorization of '" + this.logName + "':";
+ if (authorizationStateWaitPassword.passwordHint != null && !authorizationStateWaitPassword.passwordHint.isBlank()) {
+ passwordMessage += "\n\tHint: " + authorizationStateWaitPassword.passwordHint;
+ }
+ logger.info(passwordMessage);
+
+ var password = ScannerUtils.askParameter(this.logName, "Enter your password");
+
+ return sendDirectly(new CheckAuthenticationPassword(password)).then();
+ case AuthorizationStateReady.CONSTRUCTOR: {
+ return Mono.empty();
+ }
+ case AuthorizationStateClosed.CONSTRUCTOR:
+ return Mono.from(requestedDefinitiveExit).doOnNext(closeRequested -> {
+ if (closeRequested) {
+ logger.info("AsyncTdEasy closed successfully");
+ } else {
+ logger.warn("AsyncTdEasy closed unexpectedly: " + logName);
+ }
+ }).flatMap(closeRequested -> {
+ if (closeRequested) {
+ return Mono
+ .from(settings)
+ .map(settings -> settings.databaseDirectory)
+ .map(Path::of)
+ .flatMapIterable(sessionPath -> Set.of(sessionPath.resolve("media"),
+ sessionPath.resolve("passport"),
+ sessionPath.resolve("profile_photos"),
+ sessionPath.resolve("stickers"),
+ sessionPath.resolve("temp"),
+ sessionPath.resolve("thumbnails"),
+ sessionPath.resolve("wallpapers")
+ ))
+ .doOnNext(directory -> {
+ try {
+ if (!Files.walk(directory)
+ .sorted(Comparator.reverseOrder())
+ .map(Path::toFile)
+ .allMatch(File::delete)) {
+ throw new IOException("Can't delete a file!");
+ }
+ } catch (IOException e) {
+ logger.error("Can't delete temporary session subdirectory", e);
+ }
+ })
+ .then(Mono.just(closeRequested));
+ } else {
+ return Mono.just(closeRequested);
+ }
+ }).then();
+ default:
+ return Mono.empty();
+ }
+ })
+ .thenReturn(updateObj);
+ }
+}
diff --git a/src/main/java/it/tdlight/tdlibsession/td/easy/AsyncTdUpdateObj.java b/src/main/java/it/tdlight/tdlibsession/td/easy/AsyncTdUpdateObj.java
new file mode 100644
index 0000000..dab815e
--- /dev/null
+++ b/src/main/java/it/tdlight/tdlibsession/td/easy/AsyncTdUpdateObj.java
@@ -0,0 +1,49 @@
+package it.tdlight.tdlibsession.td.easy;
+
+import it.tdlight.jni.TdApi;
+import it.tdlight.jni.TdApi.AuthorizationState;
+import java.util.Objects;
+import java.util.StringJoiner;
+
+public class AsyncTdUpdateObj {
+ private final AuthorizationState state;
+ private final TdApi.Object update;
+
+ public AsyncTdUpdateObj(AuthorizationState state, TdApi.Object update) {
+ this.state = state;
+ this.update = update;
+ }
+
+ public AuthorizationState getState() {
+ return state;
+ }
+
+ public TdApi.Object getUpdate() {
+ return update;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ AsyncTdUpdateObj that = (AsyncTdUpdateObj) o;
+ return Objects.equals(state, that.state) && Objects.equals(update, that.update);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(state, update);
+ }
+
+ @Override
+ public String toString() {
+ return new StringJoiner(", ", AsyncTdUpdateObj.class.getSimpleName() + "[", "]")
+ .add("state=" + state)
+ .add("update=" + update)
+ .toString();
+ }
+}
diff --git a/src/main/java/it/tdlight/tdlibsession/td/easy/TdEasySettings.java b/src/main/java/it/tdlight/tdlibsession/td/easy/TdEasySettings.java
new file mode 100644
index 0000000..6e3d1b1
--- /dev/null
+++ b/src/main/java/it/tdlight/tdlibsession/td/easy/TdEasySettings.java
@@ -0,0 +1,262 @@
+package it.tdlight.tdlibsession.td.easy;
+
+import java.util.Objects;
+import org.jetbrains.annotations.Nullable;
+
+public class TdEasySettings {
+ public final boolean useTestDc;
+ public final String databaseDirectory;
+ public final String filesDirectory;
+ public final boolean useFileDatabase;
+ public final boolean useChatInfoDatabase;
+ public final boolean useMessageDatabase;
+ public final int apiId;
+ public final String apiHash;
+ public final String systemLanguageCode;
+ public final String deviceModel;
+ public final String systemVersion;
+ public final String applicationVersion;
+ public final boolean enableStorageOptimizer;
+ public final boolean ignoreFileNames;
+ private final Long phoneNumber;
+ private final String botToken;
+
+ public TdEasySettings(boolean useTestDc,
+ String databaseDirectory,
+ String filesDirectory,
+ boolean useFileDatabase,
+ boolean useChatInfoDatabase,
+ boolean useMessageDatabase,
+ int apiId,
+ String apiHash,
+ String systemLanguageCode,
+ String deviceModel,
+ String systemVersion,
+ String applicationVersion,
+ boolean enableStorageOptimizer,
+ boolean ignoreFileNames,
+ @Nullable Long phoneNumber,
+ @Nullable String botToken) {
+ this.useTestDc = useTestDc;
+ this.databaseDirectory = databaseDirectory;
+ this.filesDirectory = filesDirectory;
+ this.useFileDatabase = useFileDatabase;
+ this.useChatInfoDatabase = useChatInfoDatabase;
+ this.useMessageDatabase = useMessageDatabase;
+ this.apiId = apiId;
+ this.apiHash = apiHash;
+ this.systemLanguageCode = systemLanguageCode;
+ this.deviceModel = deviceModel;
+ this.systemVersion = systemVersion;
+ this.applicationVersion = applicationVersion;
+ this.enableStorageOptimizer = enableStorageOptimizer;
+ this.ignoreFileNames = ignoreFileNames;
+ this.phoneNumber = phoneNumber;
+ this.botToken = botToken;
+ if ((phoneNumber == null) == (botToken == null)) {
+ throw new IllegalArgumentException("You must set a phone number or a bot token");
+ }
+ }
+
+ public boolean isPhoneNumberSet() {
+ return phoneNumber != null;
+ }
+
+ public long getPhoneNumber() {
+ return Objects.requireNonNull(phoneNumber, "You must set a phone number");
+ }
+
+ public boolean isBotTokenSet() {
+ return botToken != null;
+ }
+
+ public String getBotToken() {
+ return Objects.requireNonNull(botToken, "You must set a bot token");
+ }
+
+ public static Builder newBuilder() {
+ return new Builder();
+ }
+
+
+ public static class Builder {
+ private boolean useTestDc = false;
+ private String databaseDirectory = "jtdlib-database";
+ private String filesDirectory = "jtdlib-files";
+ private boolean useFileDatabase = true;
+ private boolean useChatInfoDatabase = true;
+ private boolean useMessageDatabase = true;
+ private int apiId = 376588;
+ private String apiHash = "2143fdfc2bbba3ec723228d2f81336c9";
+ private String systemLanguageCode = "en";
+ private String deviceModel = "JTDLib";
+ private String systemVersion = "JTDLib";
+ private String applicationVersion = "1.0";
+ private boolean enableStorageOptimizer = false;
+ private boolean ignoreFileNames = false;
+ @Nullable
+ private Long phoneNumber = null;
+ @Nullable
+ private String botToken = null;
+
+ private Builder() {
+
+ }
+
+ public boolean isUseTestDc() {
+ return useTestDc;
+ }
+
+ public Builder setUseTestDc(boolean useTestDc) {
+ this.useTestDc = useTestDc;
+ return this;
+ }
+
+ public String getDatabaseDirectory() {
+ return databaseDirectory;
+ }
+
+ public Builder setDatabaseDirectory(String databaseDirectory) {
+ this.databaseDirectory = databaseDirectory;
+ return this;
+ }
+
+ public String getFilesDirectory() {
+ return filesDirectory;
+ }
+
+ public Builder setFilesDirectory(String filesDirectory) {
+ this.filesDirectory = filesDirectory;
+ return this;
+ }
+
+ public boolean isUseFileDatabase() {
+ return useFileDatabase;
+ }
+
+ public Builder setUseFileDatabase(boolean useFileDatabase) {
+ this.useFileDatabase = useFileDatabase;
+ return this;
+ }
+
+ public boolean isUseChatInfoDatabase() {
+ return useChatInfoDatabase;
+ }
+
+ public Builder setUseChatInfoDatabase(boolean useChatInfoDatabase) {
+ this.useChatInfoDatabase = useChatInfoDatabase;
+ return this;
+ }
+
+ public boolean isUseMessageDatabase() {
+ return useMessageDatabase;
+ }
+
+ public Builder setUseMessageDatabase(boolean useMessageDatabase) {
+ this.useMessageDatabase = useMessageDatabase;
+ return this;
+ }
+
+ public int getApiId() {
+ return apiId;
+ }
+
+ public Builder setApiId(int apiId) {
+ this.apiId = apiId;
+ return this;
+ }
+
+ public String getApiHash() {
+ return apiHash;
+ }
+
+ public Builder setApiHash(String apiHash) {
+ this.apiHash = apiHash;
+ return this;
+ }
+
+ public String getSystemLanguageCode() {
+ return systemLanguageCode;
+ }
+
+ public Builder setSystemLanguageCode(String systemLanguageCode) {
+ this.systemLanguageCode = systemLanguageCode;
+ return this;
+ }
+
+ public String getDeviceModel() {
+ return deviceModel;
+ }
+
+ public Builder setDeviceModel(String deviceModel) {
+ this.deviceModel = deviceModel;
+ return this;
+ }
+
+ public String getSystemVersion() {
+ return systemVersion;
+ }
+
+ public Builder setSystemVersion(String systemVersion) {
+ this.systemVersion = systemVersion;
+ return this;
+ }
+
+ public String getApplicationVersion() {
+ return applicationVersion;
+ }
+
+ public Builder setApplicationVersion(String applicationVersion) {
+ this.applicationVersion = applicationVersion;
+ return this;
+ }
+
+ public boolean isEnableStorageOptimizer() {
+ return enableStorageOptimizer;
+ }
+
+ public Builder setEnableStorageOptimizer(boolean enableStorageOptimizer) {
+ this.enableStorageOptimizer = enableStorageOptimizer;
+ return this;
+ }
+
+ public boolean isIgnoreFileNames() {
+ return ignoreFileNames;
+ }
+
+ public Builder setIgnoreFileNames(boolean ignoreFileNames) {
+ this.ignoreFileNames = ignoreFileNames;
+ return this;
+ }
+
+ public Builder setPhoneNumber(long phoneNumber) {
+ this.phoneNumber = phoneNumber;
+ return this;
+ }
+
+ public Builder setBotToken(String botToken) {
+ this.botToken = botToken;
+ return this;
+ }
+
+ public TdEasySettings build() {
+ return new TdEasySettings(useTestDc,
+ databaseDirectory,
+ filesDirectory,
+ useFileDatabase,
+ useChatInfoDatabase,
+ useMessageDatabase,
+ apiId,
+ apiHash,
+ systemLanguageCode,
+ deviceModel,
+ systemVersion,
+ applicationVersion,
+ enableStorageOptimizer,
+ ignoreFileNames,
+ phoneNumber,
+ botToken
+ );
+ }
+ }
+}
diff --git a/src/main/java/it/tdlight/tdlibsession/td/middle/AsyncTdMiddle.java b/src/main/java/it/tdlight/tdlibsession/td/middle/AsyncTdMiddle.java
new file mode 100644
index 0000000..82aae3c
--- /dev/null
+++ b/src/main/java/it/tdlight/tdlibsession/td/middle/AsyncTdMiddle.java
@@ -0,0 +1,24 @@
+package it.tdlight.tdlibsession.td.middle;
+
+import it.tdlight.jni.TdApi;
+import it.tdlight.tdlibsession.td.TdResult;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+public interface AsyncTdMiddle {
+
+ /**
+ * Receives incoming updates from TDLib.
+ *
+ * @return Updates
+ */
+ Flux getUpdates();
+
+ /**
+ * Sends request to TDLib. May be called from any thread.
+ *
+ * @param request Request to TDLib.
+ * @param executeDirectly Execute the function synchronously.
+ */
+ Mono> execute(TdApi.Function request, boolean executeDirectly);
+}
diff --git a/src/main/java/it/tdlight/tdlibsession/td/middle/AsyncTdMiddleCommon.java b/src/main/java/it/tdlight/tdlibsession/td/middle/AsyncTdMiddleCommon.java
new file mode 100644
index 0000000..8a7984a
--- /dev/null
+++ b/src/main/java/it/tdlight/tdlibsession/td/middle/AsyncTdMiddleCommon.java
@@ -0,0 +1,10 @@
+package it.tdlight.tdlibsession.td.middle;
+
+import io.vertx.core.AbstractVerticle;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class AsyncTdMiddleCommon extends AbstractVerticle {
+
+ private static final Logger logger = LoggerFactory.getLogger(AsyncTdMiddleCommon.class);
+}
diff --git a/src/main/java/it/tdlight/tdlibsession/td/middle/ExecuteObject.java b/src/main/java/it/tdlight/tdlibsession/td/middle/ExecuteObject.java
new file mode 100644
index 0000000..9766758
--- /dev/null
+++ b/src/main/java/it/tdlight/tdlibsession/td/middle/ExecuteObject.java
@@ -0,0 +1,55 @@
+package it.tdlight.tdlibsession.td.middle;
+
+import it.tdlight.jni.TdApi;
+import java.util.Objects;
+import java.util.StringJoiner;
+
+public class ExecuteObject {
+ private final boolean executeDirectly;
+ private final TdApi.Function request;
+
+ public ExecuteObject(boolean executeDirectly, TdApi.Function request) {
+ this.executeDirectly = executeDirectly;
+ this.request = request;
+ }
+
+ public boolean isExecuteDirectly() {
+ return executeDirectly;
+ }
+
+ public TdApi.Function getRequest() {
+ return request;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+
+ ExecuteObject that = (ExecuteObject) o;
+
+ if (executeDirectly != that.executeDirectly) {
+ return false;
+ }
+ return Objects.equals(request, that.request);
+ }
+
+ @Override
+ public int hashCode() {
+ int result = (executeDirectly ? 1 : 0);
+ result = 31 * result + (request != null ? request.hashCode() : 0);
+ return result;
+ }
+
+ @Override
+ public String toString() {
+ return new StringJoiner(", ", ExecuteObject.class.getSimpleName() + "[", "]")
+ .add("executeDirectly=" + executeDirectly)
+ .add("request=" + request)
+ .toString();
+ }
+}
diff --git a/src/main/java/it/tdlight/tdlibsession/td/middle/TdClusterManager.java b/src/main/java/it/tdlight/tdlibsession/td/middle/TdClusterManager.java
new file mode 100644
index 0000000..e9cb4ef
--- /dev/null
+++ b/src/main/java/it/tdlight/tdlibsession/td/middle/TdClusterManager.java
@@ -0,0 +1,227 @@
+package it.tdlight.tdlibsession.td.middle;
+
+import com.hazelcast.config.Config;
+import com.hazelcast.config.EvictionPolicy;
+import com.hazelcast.config.GroupConfig;
+import com.hazelcast.config.MapConfig;
+import com.hazelcast.config.MaxSizeConfig;
+import com.hazelcast.config.MaxSizeConfig.MaxSizePolicy;
+import com.hazelcast.config.MergePolicyConfig;
+import com.hazelcast.config.SemaphoreConfig;
+import io.vertx.core.Handler;
+import io.vertx.core.Vertx;
+import io.vertx.core.VertxOptions;
+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.http.ClientAuth;
+import io.vertx.core.net.JksOptions;
+import io.vertx.core.spi.cluster.ClusterManager;
+import io.vertx.spi.cluster.hazelcast.HazelcastClusterManager;
+import java.nio.channels.AlreadyBoundException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Map;
+import java.util.Random;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.jetbrains.annotations.Nullable;
+import it.tdlight.utils.MonoUtils;
+import reactor.core.publisher.Mono;
+
+public class TdClusterManager {
+
+ private static final AtomicBoolean definedMasterCluster = new AtomicBoolean(false);
+ private static final AtomicBoolean definedNodesCluster = new AtomicBoolean(false);
+ private final ClusterManager mgr;
+ private final VertxOptions vertxOptions;
+ private final Vertx vertx;
+ private final EventBus eb;
+
+ public TdClusterManager(ClusterManager mgr, VertxOptions vertxOptions, Vertx vertx, EventBus eventBus) {
+ this.mgr = mgr;
+ this.vertxOptions = vertxOptions;
+ this.vertx = vertx;
+ this.eb = eventBus;
+ }
+
+ public static Mono ofMaster(JksOptions keyStoreOptions, JksOptions trustStoreOptions, boolean onlyLocal, String masterHostname, String netInterface, int port, Set nodesAddresses) {
+ if (definedMasterCluster.compareAndSet(false, true)) {
+ var vertxOptions = new VertxOptions();
+ netInterface = onlyLocal ? "127.0.0.1" : netInterface;
+ Config cfg;
+ if (!onlyLocal) {
+ cfg = new Config();
+ cfg.setInstanceName("Master");
+ } else {
+ cfg = null;
+ }
+ return of(cfg,
+ vertxOptions,
+ keyStoreOptions, trustStoreOptions, masterHostname, netInterface, port, nodesAddresses);
+ } else {
+ return Mono.error(new AlreadyBoundException());
+ }
+ }
+
+ public static Mono ofNodes(JksOptions keyStoreOptions, JksOptions trustStoreOptions, boolean onlyLocal, String masterHostname, String netInterface, int port, Set nodesAddresses) {
+ if (definedNodesCluster.compareAndSet(false, true)) {
+ var vertxOptions = new VertxOptions();
+ netInterface = onlyLocal ? "127.0.0.1" : netInterface;
+ Config cfg;
+ if (!onlyLocal) {
+ cfg = new Config();
+ cfg.setInstanceName("Node-" + new Random().nextLong());
+ } else {
+ cfg = null;
+ }
+ return of(cfg, vertxOptions, keyStoreOptions, trustStoreOptions, masterHostname, netInterface, port, nodesAddresses);
+ } else {
+ return Mono.error(new AlreadyBoundException());
+ }
+ }
+
+ public static Mono of(@Nullable Config cfg,
+ VertxOptions vertxOptions,
+ JksOptions keyStoreOptions,
+ JksOptions trustStoreOptions,
+ String masterHostname,
+ String netInterface,
+ int port,
+ Set nodesAddresses) {
+ ClusterManager mgr;
+ if (cfg != null) {
+ cfg.getNetworkConfig().setPortCount(1);
+ cfg.getNetworkConfig().setPort(port);
+ cfg.getNetworkConfig().setPortAutoIncrement(false);
+ cfg.getPartitionGroupConfig().setEnabled(false);
+ cfg.addMapConfig(new MapConfig()
+ .setName("__vertx.subs")
+ .setBackupCount(1)
+ .setTimeToLiveSeconds(0)
+ .setMaxIdleSeconds(0)
+ .setEvictionPolicy(EvictionPolicy.NONE)
+ .setMaxSizeConfig(new MaxSizeConfig().setMaxSizePolicy(MaxSizePolicy.PER_NODE).setSize(0))
+ .setMergePolicyConfig(new MergePolicyConfig().setPolicy("com.hazelcast.map.merge.LatestUpdateMapMergePolicy")));
+ cfg.setSemaphoreConfigs(Map.of("__vertx.*", new SemaphoreConfig().setInitialPermits(1)));
+ cfg.getNetworkConfig().getJoin().getMulticastConfig().setEnabled(false);
+ cfg.getNetworkConfig().getJoin().getAwsConfig().setEnabled(false);
+ cfg.getNetworkConfig().getJoin().getTcpIpConfig().setEnabled(true);
+ var addresses = new ArrayList<>(nodesAddresses);
+ cfg.getNetworkConfig().getJoin().getTcpIpConfig().setMembers(addresses);
+ cfg.getNetworkConfig().getInterfaces().clear();
+ cfg.getNetworkConfig().getInterfaces().setInterfaces(Collections.singleton(netInterface)).setEnabled(true);
+ cfg.getNetworkConfig().setOutboundPorts(Collections.singleton(0));
+
+ cfg.setProperty("hazelcast.logging.type", "slf4j");
+ cfg.setProperty("hazelcast.wait.seconds.before.join", "0");
+ cfg.setProperty("hazelcast.tcp.join.port.try.count", "5");
+ cfg.setProperty("hazelcast.socket.bind.any", "false");
+ cfg.setGroupConfig(new GroupConfig().setName("dev").setPassword("HzPasswordsAreDeprecated"));
+ mgr = new HazelcastClusterManager(cfg);
+ vertxOptions.setClusterManager(mgr);
+ vertxOptions.getEventBusOptions().setConnectTimeout(120000);
+ //vertxOptions.getEventBusOptions().setIdleTimeout(60);
+ //vertxOptions.getEventBusOptions().setSsl(false);
+
+ vertxOptions.getEventBusOptions().setSslHandshakeTimeout(120000).setSslHandshakeTimeoutUnit(TimeUnit.MILLISECONDS);
+ vertxOptions.getEventBusOptions().setKeyStoreOptions(keyStoreOptions);
+ vertxOptions.getEventBusOptions().setTrustStoreOptions(trustStoreOptions);
+ vertxOptions.getEventBusOptions().setHost(masterHostname);
+ vertxOptions.getEventBusOptions().setPort(port + 1);
+ vertxOptions.getEventBusOptions().setSsl(true).setEnabledSecureTransportProtocols(Set.of("TLSv1.3", "TLSv1.2"));
+ vertxOptions.getEventBusOptions().setClientAuth(ClientAuth.REQUIRED);
+ } else {
+ mgr = null;
+ vertxOptions.setClusterManager(null);
+ vertxOptions.getEventBusOptions().setClustered(false);
+ }
+
+ return Mono
+ .create(sink -> {
+ if (mgr != null) {
+ Vertx.clusteredVertx(vertxOptions, MonoUtils.toHandler(sink));
+ } else {
+ sink.success(Vertx.vertx(vertxOptions));
+ }
+ })
+ .map(vertx -> new TdClusterManager(mgr, vertxOptions, vertx, vertx.eventBus()));
+ }
+
+ public Vertx getVertx() {
+ return vertx;
+ }
+
+ public EventBus getEventBus() {
+ return eb;
+ }
+
+ public VertxOptions getVertxOptions() {
+ return vertxOptions;
+ }
+
+ public DeliveryOptions newDeliveryOpts() {
+ return new DeliveryOptions().setSendTimeout(120000);
+ }
+
+ /**
+ *
+ * @param objectClass
+ * @param messageCodec
+ * @param
+ * @return true if registered, false if already registered
+ */
+ public boolean registerDefaultCodec(Class objectClass, MessageCodec messageCodec) {
+ try {
+ eb.registerDefaultCodec(objectClass, messageCodec);
+ return true;
+ } catch (IllegalStateException ex) {
+ if (ex.getMessage().startsWith("Already a default codec registered for class")) {
+ return false;
+ }
+ if (ex.getMessage().startsWith("Already a codec registered with name")) {
+ return false;
+ }
+ throw ex;
+ }
+ }
+
+ /**
+ * Create a message consumer against the specified address.
+ *