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

Replace verticles with normal class

Browse Source
This commit is contained in:
Andrea Cavalli 2021-01-13 19:46:46 +01:00
parent ed7d12e9ba
commit a3a2893fb8
6 changed files with 109 additions and 132 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
src/main/java/it/tdlight/tdlibsession/EventBusFlux.java
View File

@ -56,11 +56,23 @@ public class EventBusFlux {
MessageConsumer<byte[]> cancel = eventBus.consumer(subscriptionAddress + ".cancel"); MessageConsumer<byte[]> cancel = eventBus.consumer(subscriptionAddress + ".cancel");
var subscription = flux.subscribe(item -> { var subscription = flux.subscribe(item -> {
eventBus.send(subscriptionAddress + ".signal", SignalMessage.<T>onNext(item), signalDeliveryOptions); var request = eventBus.request(subscriptionAddress + ".signal", SignalMessage.<T>onNext(item), signalDeliveryOptions, msg2 -> {
if (msg2.failed()) {
logger.error("Failed to send onNext signal", msg2.cause());
}
});
}, error -> { }, error -> {
eventBus.send(subscriptionAddress + ".signal", SignalMessage.<T>onError(error), signalDeliveryOptions); eventBus.request(subscriptionAddress + ".signal", SignalMessage.<T>onError(error), signalDeliveryOptions, msg2 -> {
if (msg2.failed()) {
logger.error("Failed to send onNext signal", msg2.cause());
}
});
}, () -> { }, () -> {
eventBus.send(subscriptionAddress + ".signal", SignalMessage.<T>onComplete(), signalDeliveryOptions); eventBus.request(subscriptionAddress + ".signal", SignalMessage.<T>onComplete(), signalDeliveryOptions, msg2 -> {
if (msg2.failed()) {
logger.error("Failed to send onNext signal", msg2.cause());
}
});
}); });
cancel.handler(msg3 -> { cancel.handler(msg3 -> {

87
src/main/java/it/tdlight/tdlibsession/remoteclient/TDLibRemoteClient.java
View File

@ -4,7 +4,6 @@ import io.vertx.core.AsyncResult;
import io.vertx.core.Future; import io.vertx.core.Future;
import io.vertx.core.Handler; import io.vertx.core.Handler;
import io.vertx.core.eventbus.Message; import io.vertx.core.eventbus.Message;
import io.vertx.core.json.JsonObject;
import io.vertx.core.net.JksOptions; import io.vertx.core.net.JksOptions;
import io.vertx.core.shareddata.AsyncMap; import io.vertx.core.shareddata.AsyncMap;
import it.tdlight.common.Init; import it.tdlight.common.Init;
@ -37,7 +36,7 @@ public class TDLibRemoteClient implements AutoCloseable {
private final int port; private final int port;
private final Set<String> membersAddresses; private final Set<String> membersAddresses;
private final Many<TdClusterManager> clusterManager = Sinks.many().replay().latest(); private final Many<TdClusterManager> clusterManager = Sinks.many().replay().latest();
private final Scheduler vertxStatusScheduler = Schedulers.newSingle("VertxStatus", false); private final Scheduler deploymentScheduler = Schedulers.single();
public TDLibRemoteClient(SecurityInfo securityInfo, String masterHostname, String netInterface, int port, Set<String> membersAddresses) { public TDLibRemoteClient(SecurityInfo securityInfo, String masterHostname, String netInterface, int port, Set<String> membersAddresses) {
this.securityInfo = securityInfo; this.securityInfo = securityInfo;
@ -155,7 +154,9 @@ public class TDLibRemoteClient implements AutoCloseable {
}) })
.doOnError(ex -> { .doOnError(ex -> {
logger.error(ex.getLocalizedMessage(), ex); logger.error(ex.getLocalizedMessage(), ex);
}).subscribe(i -> {}, e -> {}, () -> startedEventHandler.handle(null)); }).subscribe(i -> {}, e -> {
logger.error("Remote client error", e);
}, () -> startedEventHandler.handle(null));
} catch (IOException ex) { } catch (IOException ex) {
logger.error("Remote client error", ex); logger.error("Remote client error", ex);
} }
@ -230,56 +231,43 @@ public class TDLibRemoteClient implements AutoCloseable {
private void deployBot(TdClusterManager clusterManager, String botAddress, Handler<AsyncResult<String>> deploymentHandler) { private void deployBot(TdClusterManager clusterManager, String botAddress, Handler<AsyncResult<String>> deploymentHandler) {
AsyncTdMiddleEventBusServer verticle = new AsyncTdMiddleEventBusServer(clusterManager); AsyncTdMiddleEventBusServer verticle = new AsyncTdMiddleEventBusServer(clusterManager);
verticle.onBeforeStop(handler -> { verticle.onBeforeStop(handler -> {
vertxStatusScheduler.schedule(() -> { clusterManager.getSharedData().getLockWithTimeout("deployment", 15000, lockAcquisitionResult -> {
clusterManager.getSharedData().getLockWithTimeout("deployment", 15000, lockAcquisitionResult -> { if (lockAcquisitionResult.succeeded()) {
if (lockAcquisitionResult.succeeded()) { var deploymentLock = lockAcquisitionResult.result();
var deploymentLock = lockAcquisitionResult.result(); verticle.onAfterStop(handler2 -> {
verticle.onAfterStop(handler2 -> { deploymentLock.release();
vertxStatusScheduler.schedule(() -> { handler2.complete();
deploymentLock.release(); });
handler2.complete(); clusterManager.getSharedData().getClusterWideMap("runningBotAddresses", (AsyncResult<AsyncMap<String, String>> mapResult) -> {
}); if (mapResult.succeeded()) {
}); var runningBotAddresses = mapResult.result();
clusterManager.getSharedData().getClusterWideMap("runningBotAddresses", (AsyncResult<AsyncMap<String, String>> mapResult) -> { runningBotAddresses.removeIfPresent(botAddress, netInterface, putResult -> {
if (mapResult.succeeded()) { if (putResult.succeeded()) {
var runningBotAddresses = mapResult.result(); if (putResult.result() != null) {
runningBotAddresses.removeIfPresent(botAddress, netInterface, putResult -> { handler.complete();
if (putResult.succeeded()) {
if (putResult.result() != null) {
handler.complete();
} else {
handler.fail("Can't destroy bot with address \"" + botAddress + "\" because it has been already destroyed");
}
} else { } else {
handler.fail(putResult.cause()); handler.fail("Can't destroy bot with address \"" + botAddress + "\" because it has been already destroyed");
} }
}); } else {
} else { handler.fail(putResult.cause());
handler.fail(mapResult.cause()); }
} });
}); } else {
} else { handler.fail(mapResult.cause());
handler.fail(lockAcquisitionResult.cause()); }
} });
}); } else {
handler.fail(lockAcquisitionResult.cause());
}
}); });
}); });
clusterManager verticle.start(botAddress, botAddress, false).doOnError(error -> {
.getVertx() logger.error("Can't deploy bot \"" + botAddress + "\"", error);
.deployVerticle(verticle, }).subscribeOn(deploymentScheduler).subscribe(v -> {}, err -> {
clusterManager deploymentHandler.handle(Future.failedFuture(err));
.newDeploymentOpts() }, () -> {
.setConfig(new JsonObject() deploymentHandler.handle(Future.succeededFuture());
.put("botAddress", botAddress) });
.put("botAlias", botAddress)
.put("local", false)),
(deployed) -> {
if (deployed.failed()) {
logger.error("Can't deploy bot \"" + botAddress + "\"", deployed.cause());
}
deploymentHandler.handle(deployed);
}
);
} }
private void putAllAsync(AsyncMap<Object, Object> sharedMap, private void putAllAsync(AsyncMap<Object, Object> sharedMap,
@ -307,6 +295,5 @@ public class TDLibRemoteClient implements AutoCloseable {
@Override @Override
public void close() { public void close() {
clusterManager.asFlux().blockFirst(); clusterManager.asFlux().blockFirst();
vertxStatusScheduler.dispose();
} }
} }

2
src/main/java/it/tdlight/tdlibsession/td/easy/AsyncTdEasy.java
View File

@ -58,7 +58,7 @@ public class AsyncTdEasy {
private static final Logger logger = LoggerFactory.getLogger(AsyncTdEasy.class); private static final Logger logger = LoggerFactory.getLogger(AsyncTdEasy.class);
private final Scheduler scheduler = Schedulers.newSingle("TdEasyUpdates"); private final Scheduler scheduler = Schedulers.single();
private final ReplayProcessor<AuthorizationState> authState = ReplayProcessor.create(1); private final ReplayProcessor<AuthorizationState> authState = ReplayProcessor.create(1);
private final ReplayProcessor<Boolean> requestedDefinitiveExit = ReplayProcessor.cacheLastOrDefault(false); private final ReplayProcessor<Boolean> requestedDefinitiveExit = ReplayProcessor.cacheLastOrDefault(false);
private final ReplayProcessor<TdEasySettings> settings = ReplayProcessor.cacheLast(); private final ReplayProcessor<TdEasySettings> settings = ReplayProcessor.cacheLast();

8
src/main/java/it/tdlight/tdlibsession/td/middle/client/AsyncTdMiddleEventBusClient.java
View File

@ -39,6 +39,8 @@ import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono; import reactor.core.publisher.Mono;
import reactor.core.publisher.Sinks; import reactor.core.publisher.Sinks;
import reactor.core.publisher.Sinks.Many; import reactor.core.publisher.Sinks.Many;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;
public class AsyncTdMiddleEventBusClient extends AbstractVerticle implements AsyncTdMiddle { public class AsyncTdMiddleEventBusClient extends AbstractVerticle implements AsyncTdMiddle {
@ -47,6 +49,7 @@ public class AsyncTdMiddleEventBusClient extends AbstractVerticle implements Asy
public static final boolean OUTPUT_REQUESTS = false; public static final boolean OUTPUT_REQUESTS = false;
public static final byte[] EMPTY = new byte[0]; public static final byte[] EMPTY = new byte[0];
private final Scheduler tdMiddleScheduler = Schedulers.single();
private final Many<Boolean> tdClosed = Sinks.many().replay().latestOrDefault(false); private final Many<Boolean> tdClosed = Sinks.many().replay().latestOrDefault(false);
private final DeliveryOptions deliveryOptions; private final DeliveryOptions deliveryOptions;
private final DeliveryOptions deliveryOptionsWithTimeout; private final DeliveryOptions deliveryOptionsWithTimeout;
@ -146,6 +149,7 @@ public class AsyncTdMiddleEventBusClient extends AbstractVerticle implements Asy
if (msg.succeeded()) { if (msg.succeeded()) {
this.listen() this.listen()
.timeout(Duration.ofSeconds(30)) .timeout(Duration.ofSeconds(30))
.subscribeOn(tdMiddleScheduler)
.subscribe(v -> {}, future::fail, future::complete); .subscribe(v -> {}, future::fail, future::complete);
} else { } else {
future.fail(msg.cause()); future.fail(msg.cause());
@ -241,7 +245,7 @@ public class AsyncTdMiddleEventBusClient extends AbstractVerticle implements Asy
tdClosed.tryEmitNext(true); tdClosed.tryEmitNext(true);
} }
} }
})); })).subscribeOn(tdMiddleScheduler);
} }
@Override @Override
@ -297,6 +301,6 @@ public class AsyncTdMiddleEventBusClient extends AbstractVerticle implements Asy
} }
}).switchIfEmpty(Mono.fromSupplier(() -> { }).switchIfEmpty(Mono.fromSupplier(() -> {
return TdResult.failed(new TdApi.Error(500, "Client is closed or response is empty")); return TdResult.failed(new TdApi.Error(500, "Client is closed or response is empty"));
})); })).subscribeOn(tdMiddleScheduler);
} }
} }

11
src/main/java/it/tdlight/tdlibsession/td/middle/direct/AsyncTdMiddleLocal.java
View File

@ -1,6 +1,5 @@
package it.tdlight.tdlibsession.td.middle.direct; package it.tdlight.tdlibsession.td.middle.direct;
import io.vertx.core.json.JsonObject;
import it.tdlight.jni.TdApi; import it.tdlight.jni.TdApi;
import it.tdlight.jni.TdApi.Function; import it.tdlight.jni.TdApi.Function;
import it.tdlight.jni.TdApi.Object; import it.tdlight.jni.TdApi.Object;
@ -10,7 +9,6 @@ import it.tdlight.tdlibsession.td.middle.AsyncTdMiddle;
import it.tdlight.tdlibsession.td.middle.TdClusterManager; import it.tdlight.tdlibsession.td.middle.TdClusterManager;
import it.tdlight.tdlibsession.td.middle.client.AsyncTdMiddleEventBusClient; import it.tdlight.tdlibsession.td.middle.client.AsyncTdMiddleEventBusClient;
import it.tdlight.tdlibsession.td.middle.server.AsyncTdMiddleEventBusServer; import it.tdlight.tdlibsession.td.middle.server.AsyncTdMiddleEventBusServer;
import it.tdlight.utils.MonoUtils;
import java.util.Objects; import java.util.Objects;
import org.warp.commonutils.error.InitializationException; import org.warp.commonutils.error.InitializationException;
import reactor.core.publisher.Flux; import reactor.core.publisher.Flux;
@ -35,14 +33,7 @@ public class AsyncTdMiddleLocal implements AsyncTdMiddle {
} }
public Mono<AsyncTdMiddleLocal> start() { public Mono<AsyncTdMiddleLocal> start() {
return Mono.<String>create(sink -> { return srv.start(botAddress, botAlias, true).onErrorMap(InitializationException::new).flatMap(_x -> {
masterClusterManager
.getVertx()
.deployVerticle(srv,
masterClusterManager.newDeploymentOpts().setConfig(new JsonObject().put("botAddress", botAddress).put("botAlias", botAlias).put("local", true)),
MonoUtils.toHandler(sink)
);
}).onErrorMap(InitializationException::new).flatMap(_x -> {
try { try {
return AsyncTdMiddleEventBusClient.getAndDeployInstance(masterClusterManager, botAlias, botAddress, true).doOnNext(cli -> { return AsyncTdMiddleEventBusClient.getAndDeployInstance(masterClusterManager, botAlias, botAddress, true).doOnNext(cli -> {
this.cli.onNext(cli); this.cli.onNext(cli);

115
src/main/java/it/tdlight/tdlibsession/td/middle/server/AsyncTdMiddleEventBusServer.java
View File

@ -2,7 +2,6 @@ package it.tdlight.tdlibsession.td.middle.server;
import static it.tdlight.tdlibsession.td.middle.client.AsyncTdMiddleEventBusClient.OUTPUT_REQUESTS; import static it.tdlight.tdlibsession.td.middle.client.AsyncTdMiddleEventBusClient.OUTPUT_REQUESTS;
import io.vertx.core.AbstractVerticle;
import io.vertx.core.AsyncResult; import io.vertx.core.AsyncResult;
import io.vertx.core.Future; import io.vertx.core.Future;
import io.vertx.core.Handler; import io.vertx.core.Handler;
@ -37,7 +36,7 @@ import reactor.core.publisher.Mono;
import reactor.core.scheduler.Scheduler; import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers; import reactor.core.scheduler.Schedulers;
public class AsyncTdMiddleEventBusServer extends AbstractVerticle { public class AsyncTdMiddleEventBusServer {
private static final Logger logger = LoggerFactory.getLogger(AsyncTdMiddleEventBusServer.class); private static final Logger logger = LoggerFactory.getLogger(AsyncTdMiddleEventBusServer.class);
@ -69,7 +68,7 @@ public class AsyncTdMiddleEventBusServer extends AbstractVerticle {
public AsyncTdMiddleEventBusServer(TdClusterManager clusterManager) { public AsyncTdMiddleEventBusServer(TdClusterManager clusterManager) {
this.cluster = clusterManager; this.cluster = clusterManager;
this.tdOptions = new AsyncTdDirectOptions(WAIT_DURATION, 1000); this.tdOptions = new AsyncTdDirectOptions(WAIT_DURATION, 1000);
this.tdSrvPoll = Schedulers.newSingle("TdSrvPoll"); this.tdSrvPoll = Schedulers.single();
if (cluster.registerDefaultCodec(TdResultList.class, new TdResultListMessageCodec())) { if (cluster.registerDefaultCodec(TdResultList.class, new TdResultListMessageCodec())) {
cluster.registerDefaultCodec(ExecuteObject.class, new TdExecuteObjectMessageCodec()); cluster.registerDefaultCodec(ExecuteObject.class, new TdExecuteObjectMessageCodec());
cluster.registerDefaultCodec(TdResultMessage.class, new TdResultMessageCodec()); cluster.registerDefaultCodec(TdResultMessage.class, new TdResultMessageCodec());
@ -79,57 +78,52 @@ public class AsyncTdMiddleEventBusServer extends AbstractVerticle {
} }
} }
@Override public Mono<Void> start(String botAddress, String botAlias, boolean local) {
public void start(Promise<Void> startPromise) { return Mono.<Void>create(sink -> {
var botAddress = config().getString("botAddress"); if (botAddress == null || botAddress.isEmpty()) {
if (botAddress == null || botAddress.isEmpty()) { sink.error(new IllegalArgumentException("botAddress is not set!"));
throw new IllegalArgumentException("botAddress is not set!"); }
} this.botAddress = botAddress;
this.botAddress = botAddress; if (botAlias == null || botAlias.isEmpty()) {
var botAlias = config().getString("botAlias"); sink.error(new IllegalArgumentException("botAlias is not set!"));
if (botAlias == null || botAlias.isEmpty()) { }
throw new IllegalArgumentException("botAlias is not set!"); this.botAlias = botAlias;
} this.local = local;
this.botAlias = botAlias; this.td = new AsyncTdDirectImpl(botAlias);
var local = config().getBoolean("local");
if (local == null) {
throw new IllegalArgumentException("local is not set!");
}
this.local = local;
this.td = new AsyncTdDirectImpl(botAlias);
AtomicBoolean alreadyDeployed = new AtomicBoolean(false); AtomicBoolean alreadyDeployed = new AtomicBoolean(false);
this.startConsumer = cluster.getEventBus().consumer(botAddress + ".start", (Message<byte[]> msg) -> { this.startConsumer = cluster.getEventBus().consumer(botAddress + ".start", (Message<byte[]> msg) -> {
if (alreadyDeployed.compareAndSet(false, true)) { if (alreadyDeployed.compareAndSet(false, true)) {
this.listen().then(this.pipe()).then(Mono.<Void>create(registrationSink -> { this.listen().then(this.pipe()).then(Mono.<Void>create(registrationSink -> {
this.isWorkingConsumer = cluster.getEventBus().consumer(botAddress + ".isWorking", (Message<byte[]> workingMsg) -> { this.isWorkingConsumer = cluster.getEventBus().consumer(botAddress + ".isWorking", (Message<byte[]> workingMsg) -> {
workingMsg.reply(EMPTY, cluster.newDeliveryOpts().setLocalOnly(local)); workingMsg.reply(EMPTY, cluster.newDeliveryOpts().setLocalOnly(local));
});
this.isWorkingConsumer.completionHandler(MonoUtils.toHandler(registrationSink));
})).subscribeOn(this.tdSrvPoll)
.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 { this.isWorkingConsumer.completionHandler(MonoUtils.toHandler(registrationSink));
msg.reply(EMPTY); })).subscribeOn(this.tdSrvPoll)
} .subscribe(v -> {}, ex -> {
}); logger.info(botAddress + " server deployed and started. succeeded: false");
startConsumer.completionHandler(h -> { logger.error(ex.getLocalizedMessage(), ex);
logger.info(botAddress + " server deployed. succeeded: " + h.succeeded()); msg.fail(500, ex.getLocalizedMessage());
if (h.succeeded()) { }, () -> {
logger.debug("Sending " + botAddress + ".readyToStart"); logger.info(botAddress + " server deployed and started. succeeded: true");
cluster.getEventBus().request(botAddress + ".readyToStart", EMPTY, cluster.newDeliveryOpts().setSendTimeout(30000), msg -> { msg.reply(EMPTY);
startPromise.complete(h.result()); });
}); } else {
} else { msg.reply(EMPTY);
startPromise.fail(h.cause()); }
} });
}); 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());
}
});
}).subscribeOn(tdSrvPoll);
} }
public void onBeforeStop(Consumer<Promise<Void>> r) { public void onBeforeStop(Consumer<Promise<Void>> r) {
@ -140,11 +134,6 @@ public class AsyncTdMiddleEventBusServer extends AbstractVerticle {
this.onAfterStopListeners.add(r); this.onAfterStopListeners.add(r);
} }
@Override
public void stop(Promise<Void> stopPromise) {
stopPromise.complete();
}
private void runAll(List<Consumer<Promise<Void>>> actions, Handler<AsyncResult<Void>> resultHandler) { private void runAll(List<Consumer<Promise<Void>>> actions, Handler<AsyncResult<Void>> resultHandler) {
if (actions.isEmpty()) { if (actions.isEmpty()) {
resultHandler.handle(Future.succeededFuture()); resultHandler.handle(Future.succeededFuture());
@ -202,8 +191,7 @@ public class AsyncTdMiddleEventBusServer extends AbstractVerticle {
} }
}); });
executeConsumer.completionHandler(MonoUtils.toHandler(registrationSink)); executeConsumer.completionHandler(MonoUtils.toHandler(registrationSink));
}).subscribeOn(tdSrvPoll);
});
} }
private void undeploy(Runnable whenUndeployed) { private void undeploy(Runnable whenUndeployed) {
@ -237,12 +225,7 @@ public class AsyncTdMiddleEventBusServer extends AbstractVerticle {
logger.error("An afterStop listener failed: " + onAfterStopHandler.cause()); logger.error("An afterStop listener failed: " + onAfterStopHandler.cause());
} }
vertx.undeploy(deploymentID(), undeployed -> { whenUndeployed.run();
if (undeployed.failed()) {
logger.error("Error when undeploying td verticle", undeployed.cause());
}
whenUndeployed.run();
});
}); });
}).subscribeOn(this.tdSrvPoll).subscribe(v -> {}, ex -> { }).subscribeOn(this.tdSrvPoll).subscribe(v -> {}, ex -> {
logger.error("Error when stopping", ex); logger.error("Error when stopping", ex);
@ -260,7 +243,7 @@ public class AsyncTdMiddleEventBusServer extends AbstractVerticle {
.replace(" ", "") .replace(" ", "")
.replace(" = ", "=")); .replace(" = ", "="));
} }
}).bufferTimeout(1000, local ? Duration.ofMillis(1) : Duration.ofMillis(100)) }).bufferTimeout(tdOptions.getEventsSize(), local ? Duration.ofMillis(1) : Duration.ofMillis(100))
.windowTimeout(1, Duration.ofSeconds(5)) .windowTimeout(1, Duration.ofSeconds(5))
.flatMap(w -> w.defaultIfEmpty(Collections.emptyList())) .flatMap(w -> w.defaultIfEmpty(Collections.emptyList()))
.map(TdResultList::new).doFinally(s -> { .map(TdResultList::new).doFinally(s -> {