Bugfixes

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

@@ -1,6 +1,8 @@
 package it.tdlight.tdlibsession.td.middle.client;
 
 import io.vertx.core.eventbus.DeliveryOptions;
+import io.vertx.core.eventbus.ReplyException;
+import io.vertx.core.eventbus.ReplyFailure;
 import io.vertx.core.json.JsonObject;
 import io.vertx.reactivex.core.Vertx;
 import io.vertx.reactivex.core.buffer.Buffer;
@@ -32,6 +34,8 @@ import java.util.concurrent.locks.LockSupport;
 import org.warp.commonutils.locks.LockUtils;
 import org.warp.commonutils.log.Logger;
 import org.warp.commonutils.log.LoggerFactory;
+import reactor.adapter.rxjava.RxJava2Adapter;
+import reactor.core.Disposable;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.core.publisher.Sinks;
@@ -51,13 +55,15 @@ public class AsyncTdMiddleEventBusClient implements AsyncTdMiddle {
 	private final DeliveryOptions deliveryOptionsWithTimeout;
 	private final DeliveryOptions pingDeliveryOptions;
 
-	private final One<BinlogAsyncFile> binlog = Sinks.one();
+	private final AtomicReference<BinlogAsyncFile> binlog = new AtomicReference<>();
+
+	private final AtomicReference<Disposable> pinger = new AtomicReference<>();
 
 	private final AtomicReference<MessageConsumer<TdResultList>> updates = new AtomicReference<>();
 	// This will only result in a successful completion, never completes in other ways
 	private final Empty<Void> updatesStreamEnd = Sinks.empty();
 	// This will only result in a crash, never completes in other ways
-	private final Empty<Void> crash = Sinks.empty();
+	private final AtomicReference<Throwable> crash = new AtomicReference<>();
 	// This will only result in a successful completion, never completes in other ways
 	private final Empty<Void> pingFail = Sinks.empty();
 	// This will only result in a successful completion, never completes in other ways.
@@ -121,7 +127,7 @@ public class AsyncTdMiddleEventBusClient implements AsyncTdMiddle {
 	}
 
 	private Mono<Void> saveBinlog(Buffer data) {
-		return this.binlog.asMono().flatMap(binlog -> BinlogUtils.saveBinlog(binlog, data));
+		return Mono.fromSupplier(this.binlog::get).flatMap(binlog -> BinlogUtils.saveBinlog(binlog, data));
 	}
 
 	public Mono<Void> start(long botId,
@@ -134,15 +140,9 @@ public class AsyncTdMiddleEventBusClient implements AsyncTdMiddle {
 		this.botAddress = "bots.bot." + this.botId;
 		this.local = local;
 		this.logger = LoggerFactory.getLogger(this.botId + " " + botAlias);
-		return MonoUtils
+
-				.fromBlockingEmpty(() -> {
+		return Mono
-					EmitResult result;
+				.fromRunnable(() -> this.binlog.set(binlog))
-					while ((result = this.binlog.tryEmitValue(binlog)) == EmitResult.FAIL_NON_SERIALIZED) {
-						// 10ms
-						LockSupport.parkNanos(10000000);
-					}
-					result.orThrow();
-				})
 				.then(binlog.getLastModifiedTime())
 				.zipWith(binlog.readFully().map(Buffer::getDelegate))
 				.single()
@@ -156,69 +156,76 @@ public class AsyncTdMiddleEventBusClient implements AsyncTdMiddle {
 							binlogLastModifiedTime,
 							implementationDetails
 					);
-					return setupUpdatesListener()
+
-							.then(Mono.defer(() -> {
+					Mono<Void> startBotRequest;
-								if (local) {
+
-									return Mono.empty();
+					if (local) {
-								}
+						startBotRequest = Mono.empty();
-								logger.trace("Requesting bots.start-bot");
+					} else {
-								return cluster.getEventBus()
+						startBotRequest = cluster
-										.<byte[]>rxRequest("bots.start-bot", msg).as(MonoUtils::toMono)
+								.getEventBus()
-										.doOnSuccess(s -> logger.trace("bots.start-bot returned successfully"))
+								.<byte[]>rxRequest("bots.start-bot", msg)
-										.subscribeOn(Schedulers.boundedElastic());
+								.to(RxJava2Adapter::singleToMono)
-							}))
+								.doOnSuccess(s -> logger.trace("bots.start-bot returned successfully"))
-							.then(setupPing());
+								.doFirst(() -> logger.trace("Requesting bots.start-bot"))
+								.onErrorMap(ex -> {
+									if (ex instanceof ReplyException) {
+										if (((ReplyException) ex).failureType() == ReplyFailure.NO_HANDLERS) {
+											return new NoClustersAvailableException("Can't start bot "
+													+ botId + " " + botAlias);
+										}
+									}
+									return ex;
+								})
+								.then()
+								.subscribeOn(Schedulers.boundedElastic());
+					}
+
+					return setupUpdatesListener().then(startBotRequest).then(setupPing());
 				});
 	}
 
 	private Mono<Void> setupPing() {
-		return Mono.<Void>fromCallable(() -> {
+		// Disable ping on local servers
-			logger.trace("Setting up ping");
+		if (local) {
-			// Disable ping on local servers
+			return Mono.empty();
-			if (!local) {
+		}
-				Mono
+
-						.defer(() -> {
+		var pingRequest = cluster.getEventBus()
-							logger.trace("Requesting ping...");
+				.<byte[]>rxRequest(botAddress + ".ping", EMPTY, pingDeliveryOptions)
-							return cluster.getEventBus()
+				.to(RxJava2Adapter::singleToMono)
-									.<byte[]>rxRequest(botAddress + ".ping", EMPTY, pingDeliveryOptions)
+				.doFirst(() -> logger.trace("Requesting ping..."));
-									.as(MonoUtils::toMono);
+
-						})
+		return Mono
-						.flatMap(msg -> Mono.fromCallable(msg::body).subscribeOn(Schedulers.boundedElastic()))
+				.fromRunnable(() -> pinger.set(pingRequest
-						.repeatWhen(l -> l.delayElements(Duration.ofSeconds(10)).takeWhile(x -> true))
+					.flatMap(msg -> Mono.fromCallable(msg::body).subscribeOn(Schedulers.boundedElastic()))
-						.takeUntilOther(Mono.firstWithSignal(
+					.repeatWhen(l -> l.delayElements(Duration.ofSeconds(10)).takeWhile(x -> true))
-								this.updatesStreamEnd
+					.doOnNext(s -> logger.trace("PING"))
-										.asMono()
+					.then()
-										.doOnTerminate(() -> logger.trace("About to kill pinger because updates stream ended")),
+					.onErrorResume(ex -> {
-								this.crash
+						logger.warn("Ping failed: {}", ex.getMessage());
-										.asMono()
+						return Mono.empty();
-										.onErrorResume(ex -> Mono.empty())
+					})
-										.doOnTerminate(() -> logger.trace("About to kill pinger because it has seen a crash signal"))
+					.doOnNext(s -> logger.debug("END PING"))
-						))
+					.then(Mono.fromRunnable(() -> {
-						.doOnNext(s -> logger.trace("PING"))
+						while (this.pingFail.tryEmitEmpty() == EmitResult.FAIL_NON_SERIALIZED) {
-						.then()
+							// 10ms
-						.onErrorResume(ex -> {
+							LockSupport.parkNanos(10000000);
-							logger.warn("Ping failed: {}", ex.getMessage());
+						}
-							return Mono.empty();
+					}).subscribeOn(Schedulers.boundedElastic()))
-						})
+					.subscribeOn(Schedulers.parallel())
-						.doOnNext(s -> logger.debug("END PING"))
+					.subscribe())
-						.then(MonoUtils.fromBlockingEmpty(() -> {
+				)
-							while (this.pingFail.tryEmitEmpty() == EmitResult.FAIL_NON_SERIALIZED) {
+				.then()
-								// 10ms
+				.doFirst(() -> logger.trace("Setting up ping"))
-								LockSupport.parkNanos(10000000);
+				.doOnSuccess(s -> logger.trace("Ping setup success"))
-							}
+				.subscribeOn(Schedulers.boundedElastic());
-						}))
-						.subscribeOn(Schedulers.parallel())
-						.subscribe();
-			}
-			logger.trace("Ping setup success");
-			return null;
-		}).subscribeOn(Schedulers.boundedElastic());
 	}
 
 	private Mono<Void> setupUpdatesListener() {
 		return Mono
 				.fromRunnable(() -> logger.trace("Setting up updates listener..."))
-				.then(MonoUtils.<MessageConsumer<TdResultList>>fromBlockingSingle(() -> MessageConsumer
+				.then(Mono.<MessageConsumer<TdResultList>>fromSupplier(() -> MessageConsumer
 						.newInstance(cluster.getEventBus().<TdResultList>consumer(botAddress + ".updates")
 								.setMaxBufferedMessages(5000)
 								.getDelegate()
@@ -228,7 +235,7 @@ public class AsyncTdMiddleEventBusClient implements AsyncTdMiddle {
 					// Return when the registration of all the consumers has been done across the cluster
 					return Mono
 							.fromRunnable(() -> logger.trace("Emitting updates flux to sink"))
-							.then(MonoUtils.fromBlockingEmpty(() -> {
+							.then(Mono.fromRunnable(() -> {
 								var previous = this.updates.getAndSet(updateConsumer);
 								if (previous != null) {
 									logger.error("Already subscribed a consumer to the updates");
@@ -257,38 +264,12 @@ public class AsyncTdMiddleEventBusClient implements AsyncTdMiddle {
 						.then(cluster.getEventBus().<byte[]>rxRequest(botAddress + ".ready-to-receive",
 								EMPTY,
 								deliveryOptionsWithTimeout
-						).as(MonoUtils::toMono))
+						).to(RxJava2Adapter::singleToMono))
 						.doOnSuccess(s -> logger.trace("Sent ready-to-receive, received reply"))
 						.doOnSuccess(s -> logger.trace("About to read updates flux"))
 						.then(), updatesMessageConsumer)
 				)
-				.takeUntilOther(Flux
+				.takeUntilOther(pingFail.asMono())
-						.merge(
-								crash.asMono()
-										.onErrorResume(ex -> {
-											logger.error("TDLib crashed", ex);
-											return Mono.empty();
-										}),
-								pingFail.asMono()
-										.then(Mono.fromCallable(() -> {
-											var ex = new ConnectException("Server did not respond to ping");
-											ex.setStackTrace(new StackTraceElement[0]);
-											throw ex;
-										})).onErrorResume(ex -> MonoUtils.fromBlockingSingle(() -> {
-											EmitResult result;
-											while ((result = this.crash.tryEmitError(ex)) == EmitResult.FAIL_NON_SERIALIZED) {
-												// 10ms
-												LockSupport.parkNanos(10000000);
-											}
-											return result;
-										}))
-										.takeUntilOther(Mono
-												.firstWithSignal(crash.asMono(), authStateClosing.asMono())
-												.onErrorResume(e -> Mono.empty())
-										)
-						)
-						.doOnTerminate(() -> logger.trace("TakeUntilOther has been trigghered, the receive() flux will end"))
-				)
 				.takeUntil(a -> a.succeeded() && a.value().stream().anyMatch(item -> {
 					if (item.getConstructor() == UpdateAuthorizationState.CONSTRUCTOR) {
 						return ((UpdateAuthorizationState) item).authorizationState.getConstructor()
@@ -305,13 +286,20 @@ public class AsyncTdMiddleEventBusClient implements AsyncTdMiddle {
 				})
 				.concatMap(update -> interceptUpdate(update))
 				// Redirect errors to crash sink
-				.doOnError(error -> crash.tryEmitError(error))
+				.doOnError(error -> crash.compareAndSet(null, error))
 				.onErrorResume(ex -> {
 					logger.trace("Absorbing the error, the error has been published using the crash sink", ex);
 					return Mono.empty();
 				})
-
+				.doOnCancel(() -> {
-				.doOnTerminate(updatesStreamEnd::tryEmitEmpty);
+				})
+				.doFinally(s -> {
+					var pinger = this.pinger.get();
+					if (pinger != null) {
+						pinger.dispose();
+					}
+					updatesStreamEnd.tryEmitEmpty();
+				});
 	}
 
 	private Mono<TdApi.Object> interceptUpdate(Object update) {
@@ -326,7 +314,7 @@ public class AsyncTdMiddleEventBusClient implements AsyncTdMiddle {
 					logger.info("Received AuthorizationStateClosed from tdlib");
 					return cluster.getEventBus()
 							.<EndSessionMessage>rxRequest(this.botAddress + ".read-binlog", EMPTY)
-							.as(MonoUtils::toMono)
+							.to(RxJava2Adapter::singleToMono)
 							.mapNotNull(Message::body)
 							.doOnNext(latestBinlog -> logger.info("Received binlog from server. Size: {}",
 									BinlogUtils.humanReadableByteCountBin(latestBinlog.binlog().length())))
@@ -346,13 +334,9 @@ public class AsyncTdMiddleEventBusClient implements AsyncTdMiddle {
 				// Timeout + 5s (5 seconds extra are used to wait the graceful server-side timeout response)
 				.setSendTimeout(timeout.toMillis() + 5000);
 
-		var crashMono = crash.asMono()
-				.doOnSuccess(s -> logger.debug("Failed request {} because the TDLib session was already crashed", request))
-				.then(Mono.<TdResult<T>>empty());
-
 		var executionMono = cluster.getEventBus()
 				.<TdResultMessage>rxRequest(botAddress + ".execute", req, deliveryOptions)
-				.as(MonoUtils::toMono)
+				.to(RxJava2Adapter::singleToMono)
 				.onErrorMap(ex -> ResponseError.newResponseError(request, botAlias, ex))
 				.<TdResult<T>>handle((resp, sink) -> {
 					if (resp.body() == null) {
@@ -366,9 +350,17 @@ public class AsyncTdMiddleEventBusClient implements AsyncTdMiddle {
 				.doOnSuccess(s -> logger.trace("Executed request {}", request))
 				.doOnError(ex -> logger.debug("Failed request {}: {}", req, ex));
 
-		return Mono
+		return executionMono
-				.firstWithSignal(crashMono, executionMono)
+				.transformDeferred(mono -> {
+					var crash = this.crash.get();
+					if (crash != null) {
+						logger.debug("Failed request {} because the TDLib session was already crashed", request);
+						return Mono.empty();
+					} else {
+						return mono;
+					}
+				})
 				.switchIfEmpty(Mono.error(() -> ResponseError.newResponseError(request, botAlias,
-						new TdError(500, "Client is closed or response is empty"))));
+						new TdError(500, "The client is closed or the response is empty"))));
 	}
 }

src/main/java/it/tdlight/tdlibsession/td/middle/client/NoClustersAvailableException.java

@@ -0,0 +1,13 @@
+package it.tdlight.tdlibsession.td.middle.client;
+
+public class NoClustersAvailableException extends Throwable {
+
+	public NoClustersAvailableException(String error) {
+		super(error);
+	}
+
+	@Override
+	public String toString() {
+		return "No clusters are available. " + this.getMessage();
+	}
+}

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

@@ -1,13 +1,21 @@
 package it.tdlight.tdlibsession.td.middle.server;
 
 import io.reactivex.Completable;
+import io.reactivex.processors.BehaviorProcessor;
 import io.vertx.core.eventbus.DeliveryOptions;
 import io.vertx.core.eventbus.ReplyException;
 import io.vertx.core.eventbus.ReplyFailure;
+import io.vertx.core.streams.Pipe;
+import io.vertx.core.streams.Pump;
+import io.vertx.ext.reactivestreams.impl.ReactiveWriteStreamImpl;
+import io.vertx.reactivex.RxHelper;
+import io.vertx.reactivex.WriteStreamObserver;
 import io.vertx.reactivex.core.AbstractVerticle;
 import io.vertx.reactivex.core.eventbus.Message;
 import io.vertx.reactivex.core.eventbus.MessageConsumer;
 import io.vertx.reactivex.core.eventbus.MessageProducer;
+import io.vertx.reactivex.core.streams.WriteStream;
+import io.vertx.reactivex.impl.FlowableReadStream;
 import it.tdlight.jni.TdApi;
 import it.tdlight.jni.TdApi.AuthorizationStateClosed;
 import it.tdlight.jni.TdApi.Error;
@@ -25,7 +33,6 @@ import it.tdlight.tdlibsession.td.middle.TdResultList;
 import it.tdlight.tdlibsession.td.middle.TdResultListMessageCodec;
 import it.tdlight.tdlibsession.td.middle.TdResultMessage;
 import it.tdlight.utils.BinlogUtils;
-import it.tdlight.utils.MonoUtils;
 import java.net.ConnectException;
 import java.time.Duration;
 import java.util.List;
@@ -33,6 +40,8 @@ import java.util.Objects;
 import java.util.concurrent.atomic.AtomicReference;
 import org.warp.commonutils.log.Logger;
 import org.warp.commonutils.log.LoggerFactory;
+import reactor.adapter.rxjava.RxJava2Adapter;
+import reactor.core.Disposable;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.core.scheduler.Schedulers;
@@ -56,10 +65,11 @@ public class AsyncTdMiddleEventBusServer extends AbstractVerticle {
 
 	// Variables configured at startup
 	private final AtomicReference<AsyncTdDirectImpl> td = new AtomicReference<>();
-	private final AtomicReference<MessageConsumer<ExecuteObject<?>>> executeConsumer = new AtomicReference<>();
+	private final AtomicReference<Disposable> executeConsumer = new AtomicReference<>();
-	private final AtomicReference<MessageConsumer<byte[]>> readBinlogConsumer = new AtomicReference<>();
+	private final AtomicReference<Disposable> readBinlogConsumer = new AtomicReference<>();
-	private final AtomicReference<MessageConsumer<byte[]>> readyToReceiveConsumer = new AtomicReference<>();
+	private final AtomicReference<Disposable> readyToReceiveConsumer = new AtomicReference<>();
-	private final AtomicReference<MessageConsumer<byte[]>> pingConsumer = new AtomicReference<>();
+	private final AtomicReference<Disposable> pingConsumer = new AtomicReference<>();
+	private final AtomicReference<Disposable> clusterPropagationWaiter = new AtomicReference<>();
 	private final AtomicReference<Flux<Void>> pipeFlux = new AtomicReference<>();
 
 	public AsyncTdMiddleEventBusServer() {
@@ -69,38 +79,37 @@ public class AsyncTdMiddleEventBusServer extends AbstractVerticle {
 
 	@Override
 	public Completable rxStart() {
-		return MonoUtils
+		return Mono
-				.toCompletable(MonoUtils
+				.fromCallable(() -> {
-						.fromBlockingMaybe(() -> {
+					logger.trace("Stating verticle");
-							logger.trace("Stating verticle");
+					var botId = config().getInteger("botId");
-							var botId = config().getInteger("botId");
+					if (botId == null || botId <= 0) {
-							if (botId == null || botId <= 0) {
+						throw new IllegalArgumentException("botId is not set!");
-								throw new IllegalArgumentException("botId is not set!");
+					}
-							}
+					this.botId.set(botId);
-							this.botId.set(botId);
+					var botAddress = "bots.bot." + botId;
-							var botAddress = "bots.bot." + botId;
+					this.botAddress.set(botAddress);
-							this.botAddress.set(botAddress);
+					var botAlias = config().getString("botAlias");
-							var botAlias = config().getString("botAlias");
+					if (botAlias == null || botAlias.isEmpty()) {
-							if (botAlias == null || botAlias.isEmpty()) {
+						throw new IllegalArgumentException("botAlias is not set!");
-								throw new IllegalArgumentException("botAlias is not set!");
+					}
-							}
+					this.botAlias.set(botAlias);
-							this.botAlias.set(botAlias);
+					var local = config().getBoolean("local");
-							var local = config().getBoolean("local");
+					if (local == null) {
-							if (local == null) {
+						throw new IllegalArgumentException("local is not set!");
-								throw new IllegalArgumentException("local is not set!");
+					}
-							}
+					var implementationDetails = config().getJsonObject("implementationDetails");
-							var implementationDetails = config().getJsonObject("implementationDetails");
+					if (implementationDetails == null) {
-							if (implementationDetails == null) {
+						throw new IllegalArgumentException("implementationDetails is not set!");
-								throw new IllegalArgumentException("implementationDetails is not set!");
+					}
-							}
 
-							var td = new AsyncTdDirectImpl(clientFactory, implementationDetails, botAlias);
+					var td = new AsyncTdDirectImpl(clientFactory, implementationDetails, botAlias);
-							this.td.set(td);
+					this.td.set(td);
-							return new OnSuccessfulStartRequestInfo(td, botAddress, botAlias, botId, local);
+					return new OnSuccessfulStartRequestInfo(td, botAddress, botAlias, botId, local);
-						})
+				})
-						.flatMap(r -> onSuccessfulStartRequest(r.td, r.botAddress, r.botAlias, r.botId, r.local))
+				.flatMap(r -> onSuccessfulStartRequest(r.td, r.botAddress, r.botAlias, r.botId, r.local))
-						.doOnSuccess(s -> logger.trace("Stated verticle"))
+				.doOnSuccess(s -> logger.trace("Started verticle"))
-				);
+				.as(RxJava2Adapter::monoToCompletable);
 	}
 
 	private static class OnSuccessfulStartRequestInfo {
@@ -132,16 +141,13 @@ public class AsyncTdMiddleEventBusServer extends AbstractVerticle {
 	}
 
 	private Mono<Void> listen(AsyncTdDirectImpl td, String botAddress, int botId, boolean local) {
-		return Mono.<Void>create(registrationSink -> {
+		return Mono.create(registrationSink -> {
 			logger.trace("Preparing listeners");
 
 			MessageConsumer<ExecuteObject<?>> executeConsumer = vertx.eventBus().consumer(botAddress + ".execute");
-			this.executeConsumer.set(executeConsumer);
+			this.executeConsumer.set(executeConsumer
-			Flux
+					.toFlowable()
-					.<Message<ExecuteObject<?>>>create(sink -> {
+					.to(RxJava2Adapter::flowableToFlux)
-						executeConsumer.handler(sink::next);
-						executeConsumer.endHandler(h -> sink.complete());
-					})
 					.flatMap(msg -> {
 						var body = msg.body();
 						var request = overrideRequest(body.getRequest(), botId);
@@ -154,7 +160,7 @@ public class AsyncTdMiddleEventBusServer extends AbstractVerticle {
 								.single()
 								.doOnSuccess(s -> logger.trace("Executed successfully. Request was {}", request))
 								.onErrorResume(ex -> Mono.fromRunnable(() -> msg.fail(500, ex.getLocalizedMessage())))
-								.flatMap(response -> Mono.fromCallable(() -> {
+								.map(response -> {
 									var replyOpts = new DeliveryOptions().setLocalOnly(local);
 									var replyValue = new TdResultMessage(response.result(), response.cause());
 									try {
@@ -162,38 +168,32 @@ public class AsyncTdMiddleEventBusServer extends AbstractVerticle {
 										msg.reply(replyValue, replyOpts);
 										return response;
 									} catch (Exception ex) {
-										logger.debug("Replying with error response: {}. Request was {}", ex.getLocalizedMessage(),
+										logger.debug("Replying with error response: {}. Request was {}", ex.getLocalizedMessage(), request);
-												request);
 										msg.fail(500, ex.getLocalizedMessage());
 										throw ex;
 									}
-								}).subscribeOn(Schedulers.boundedElastic()));
+								});
 					})
 					.then()
-					.subscribeOn(Schedulers.parallel())
+					.subscribeOn(Schedulers.boundedElastic())
 					.subscribe(v -> {},
 							ex -> logger.error("Fatal error when processing an execute request."
 									+ " Can't process further requests since the subscription has been broken", ex),
 							() -> logger.trace("Finished handling execute requests")
-					);
+					));
 
 			MessageConsumer<byte[]> readBinlogConsumer = vertx.eventBus().consumer(botAddress + ".read-binlog");
-			this.readBinlogConsumer.set(readBinlogConsumer);
+			this.readBinlogConsumer.set(BinlogUtils
-			BinlogUtils
 					.readBinlogConsumer(vertx, readBinlogConsumer, botId, local)
-					.subscribeOn(Schedulers.parallel())
+					.subscribeOn(Schedulers.boundedElastic())
-					.subscribe(v -> {}, ex -> logger.error("Error when processing a read-binlog request", ex));
+					.subscribe(v -> {}, ex -> logger.error("Error when processing a read-binlog request", ex)));
 
-			MessageConsumer<byte[]> readyToReceiveConsumer = vertx.eventBus().consumer(botAddress
+			MessageConsumer<byte[]> readyToReceiveConsumer = vertx.eventBus().consumer(botAddress + ".ready-to-receive");
-					+ ".ready-to-receive");
-			this.readyToReceiveConsumer.set(readyToReceiveConsumer);
 
 			// Pipe the data
-			Flux
+			this.readyToReceiveConsumer.set(readyToReceiveConsumer
-					.<Message<byte[]>>create(sink -> {
+					.toFlowable()
-						readyToReceiveConsumer.handler(sink::next);
+					.to(RxJava2Adapter::flowableToFlux)
-						readyToReceiveConsumer.endHandler(h -> sink.complete());
-					})
 					.take(1, true)
 					.single()
 					.doOnNext(s -> logger.trace("Received ready-to-receive request from client"))
@@ -201,8 +201,7 @@ public class AsyncTdMiddleEventBusServer extends AbstractVerticle {
 					.doOnError(ex -> logger.error("Error when processing a ready-to-receive request", ex))
 					.doOnNext(s -> logger.trace("Replying to ready-to-receive request"))
 					.flatMapMany(tuple -> {
-						var opts = new DeliveryOptions().setLocalOnly(local)
+						var opts = new DeliveryOptions().setLocalOnly(local).setSendTimeout(Duration.ofSeconds(10).toMillis());
-								.setSendTimeout(Duration.ofSeconds(10).toMillis());
 
 						tuple.getT1().reply(EMPTY, opts);
 						logger.trace("Replied to ready-to-receive");
@@ -216,39 +215,33 @@ public class AsyncTdMiddleEventBusServer extends AbstractVerticle {
 					.doOnSuccess(s -> logger.trace("Finished handling ready-to-receive requests (updates pipe ended)"))
 					.subscribeOn(Schedulers.boundedElastic())
 					// Don't handle errors here. Handle them in pipeFlux
-					.subscribe(v -> {});
+					.subscribe(v -> {}));
 
 			MessageConsumer<byte[]> pingConsumer = vertx.eventBus().consumer(botAddress + ".ping");
-			this.pingConsumer.set(pingConsumer);
-			Flux
-					.<Message<byte[]>>create(sink -> {
-						pingConsumer.handler(sink::next);
-						pingConsumer.endHandler(h -> sink.complete());
-					})
-					.concatMap(msg -> Mono.fromCallable(() -> {
-						var opts = new DeliveryOptions().setLocalOnly(local)
-								.setSendTimeout(Duration.ofSeconds(10).toMillis());
-						msg.reply(EMPTY, opts);
-						return null;
-					}))
-					.then()
-					.subscribeOn(Schedulers.boundedElastic())
-					.subscribe(v -> {},
-							ex -> logger.error("Error when processing a ping request", ex),
-							() -> logger.trace("Finished handling ping requests")
-					);
 
-			executeConsumer
+			this.pingConsumer.set(pingConsumer
-					.rxCompletionHandler()
+					.toFlowable()
-					.andThen(readBinlogConsumer.rxCompletionHandler())
+					.to(RxJava2Adapter::flowableToFlux)
-					.andThen(readyToReceiveConsumer.rxCompletionHandler())
+					.doOnNext(msg -> {
-					.andThen(pingConsumer.rxCompletionHandler())
+						var opts = new DeliveryOptions().setLocalOnly(local).setSendTimeout(Duration.ofSeconds(10).toMillis());
-					.as(MonoUtils::toMono)
+						msg.reply(EMPTY, opts);
+					})
+					.subscribeOn(Schedulers.boundedElastic())
+					.subscribe(unused -> logger.trace("Finished handling ping requests"),
+							ex -> logger.error("Error when processing a ping request", ex)
+					));
+
+			var executorPropagated = executeConsumer.rxCompletionHandler().to(RxJava2Adapter::completableToMono);
+			var readyToReceivePropagated = executeConsumer.rxCompletionHandler().to(RxJava2Adapter::completableToMono);
+			var readBinLogPropagated = executeConsumer.rxCompletionHandler().to(RxJava2Adapter::completableToMono);
+			var pingPropagated = executeConsumer.rxCompletionHandler().to(RxJava2Adapter::completableToMono);
+
+			var allPropagated = Mono.when(executorPropagated, readyToReceivePropagated, readBinLogPropagated, pingPropagated);
+			this.clusterPropagationWaiter.set(allPropagated
 					.doOnSuccess(s -> logger.trace("Finished preparing listeners"))
-					.subscribeOn(Schedulers.parallel())
+					.subscribeOn(Schedulers.boundedElastic())
-					.subscribe(v -> {}, registrationSink::error, registrationSink::success);
+					.subscribe(v -> {}, registrationSink::error, registrationSink::success));
-		})
+		});
-				.subscribeOn(Schedulers.boundedElastic());
 	}
 
 	/**
@@ -266,35 +259,34 @@ public class AsyncTdMiddleEventBusServer extends AbstractVerticle {
 
 	@Override
 	public Completable rxStop() {
-		return MonoUtils.toCompletable(Mono
+		return Mono
-				.fromRunnable(() -> logger.info("Undeploy of bot \"" + botAlias.get() + "\": stopping"))
+				.fromRunnable(() -> {
-				.then(Mono
+					logger.info("Undeploy of bot \"" + botAlias.get() + "\": stopping");
-						.fromCallable(executeConsumer::get)
+					var executeConsumer = this.executeConsumer.get();
-						.flatMap(executeConsumer -> executeConsumer.rxUnregister().as(MonoUtils::toMono))
+					if (executeConsumer != null) {
-						.doOnSuccess(s -> logger.trace("Unregistered execute consumer"))
+						executeConsumer.dispose();
-				)
+						logger.trace("Unregistered execute consumer");
-				.then(MonoUtils.fromBlockingEmpty(() -> {
+					}
+					var pingConsumer = this.pingConsumer.get();
+					if (pingConsumer != null) {
+						pingConsumer.dispose();
+					}
 					var readBinlogConsumer = this.readBinlogConsumer.get();
 					if (readBinlogConsumer != null) {
-						Mono
+						readBinlogConsumer.dispose();
-								// ReadBinLog will live for another 10 minutes.
-								// Since every consumer of ReadBinLog is identical, this should not pose a problem.
-								.delay(Duration.ofMinutes(10))
-								.then(readBinlogConsumer.rxUnregister().as(MonoUtils::toMono))
-								.subscribe();
 					}
-				}))
+					var readyToReceiveConsumer = this.readyToReceiveConsumer.get();
-				.then(Mono
+					if (readyToReceiveConsumer != null) {
-						.fromCallable(readyToReceiveConsumer::get)
+						readyToReceiveConsumer.dispose();
-						.flatMap(ec -> ec.rxUnregister().as(MonoUtils::toMono))
+					}
-				)
+					var clusterPropagationWaiter = this.clusterPropagationWaiter.get();
-				.then(Mono
+					if (clusterPropagationWaiter != null) {
-						.fromCallable(pingConsumer::get)
+						clusterPropagationWaiter.dispose();
-						.flatMap(ec -> ec.rxUnregister().as(MonoUtils::toMono))
+					}
-				)
+				})
 				.doOnError(ex -> logger.error("Undeploy of bot \"" + botAlias.get() + "\": stop failed", ex))
 				.doOnTerminate(() -> logger.info("Undeploy of bot \"" + botAlias.get() + "\": stopped"))
-		);
+				.as(RxJava2Adapter::monoToCompletable);
 	}
 
 	private Mono<Void> pipe(AsyncTdDirectImpl td, String botAddress, boolean local) {
@@ -342,7 +334,7 @@ public class AsyncTdMiddleEventBusServer extends AbstractVerticle {
 		var pipeFlux = updatesFlux
 				.concatMap(updatesList -> updatesSender
 						.rxWrite(updatesList)
-						.as(MonoUtils::toMono)
+						.to(RxJava2Adapter::completableToMono)
 						.thenReturn(updatesList)
 				)
 				.concatMap(updatesList -> Flux
@@ -355,13 +347,13 @@ public class AsyncTdMiddleEventBusServer extends AbstractVerticle {
 									if (tdUpdateAuthorizationState.authorizationState.getConstructor()
 											== AuthorizationStateClosed.CONSTRUCTOR) {
 										logger.info("Undeploying after receiving AuthorizationStateClosed");
-										return rxStop().as(MonoUtils::toMono).thenReturn(item);
+										return rxStop().to(RxJava2Adapter::completableToMono).thenReturn(item);
 									}
 								}
 							} else if (item instanceof Error) {
 								// An error in updates means that a fatal error occurred
 								logger.info("Undeploying after receiving a fatal error");
-								return rxStop().as(MonoUtils::toMono).thenReturn(item);
+								return rxStop().to(RxJava2Adapter::completableToMono).thenReturn(item);
 							}
 							return Mono.just(item);
 						})
@@ -393,10 +385,10 @@ public class AsyncTdMiddleEventBusServer extends AbstractVerticle {
 							.execute(new TdApi.Close(), Duration.ofDays(1), false)
 							.doOnError(ex2 -> logger.error("Unexpected error", ex2))
 							.doOnSuccess(s -> logger.debug("Emergency Close() signal has been sent successfully"))
-							.then(rxStop().as(MonoUtils::toMono));
+							.then(rxStop().to(RxJava2Adapter::completableToMono));
 				});
 
-		return MonoUtils.fromBlockingEmpty(() -> {
+		return Mono.fromRunnable(() -> {
 			this.pipeFlux.set(pipeFlux);
 			logger.trace("Prepared piping requests successfully");
 		});

src/main/java/it/tdlight/utils/MonoUtils.java

@@ -35,6 +35,7 @@ import org.warp.commonutils.concurrency.future.CompletableFutureUtils;
 import org.warp.commonutils.functional.IOConsumer;
 import org.warp.commonutils.log.Logger;
 import org.warp.commonutils.log.LoggerFactory;
+import reactor.adapter.rxjava.RxJava2Adapter;
 import reactor.core.CoreSubscriber;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.FluxSink.OverflowStrategy;
@@ -171,38 +172,12 @@ public class MonoUtils {
 
 	public static <T> Flux<Message<T>> fromReplyableMessageConsumer(Mono<Void> onRegistered,
 			MessageConsumer<T> messageConsumer) {
-		Mono<Void> endMono = Mono.create(sink -> {
+		var registration = messageConsumer
-			AtomicBoolean alreadyRequested = new AtomicBoolean();
+				.rxCompletionHandler().to(RxJava2Adapter::completableToMono)
-			sink.onRequest(n -> {
+				.doFirst(() -> logger.trace("Waiting for consumer registration completion..."))
-				if (n > 0 && alreadyRequested.compareAndSet(false, true)) {
-					messageConsumer.endHandler(e -> sink.success());
-				}
-			});
-		});
-
-		Mono<MessageConsumer<T>> registrationCompletionMono = Mono
-				.fromRunnable(() -> logger.trace("Waiting for consumer registration completion..."))
-				.<Void>then(messageConsumer.rxCompletionHandler().as(MonoUtils::toMono))
 				.doOnSuccess(s -> logger.trace("Consumer registered"))
-				.then(onRegistered)
+				.then(onRegistered);
-				.thenReturn(messageConsumer);
+		return messageConsumer.toFlowable().to(RxJava2Adapter::flowableToFlux).mergeWith(registration.then(Mono.empty()));
-
-		messageConsumer.handler(s -> {
-			throw new IllegalStateException("Subscriber still didn't request any value!");
-		});
-
-		Flux<Message<T>> dataFlux = Flux
-				.push(sink -> sink.onRequest(n -> messageConsumer.handler(sink::next)), OverflowStrategy.ERROR);
-
-		Mono<Void> disposeMono = messageConsumer
-				.rxUnregister()
-				.as(MonoUtils::<Message<T>>toMono)
-				.doOnSuccess(s -> logger.trace("Unregistered message consumer"))
-				.then();
-
-		return Flux
-				.usingWhen(registrationCompletionMono, msgCons -> dataFlux, msgCons -> disposeMono)
-				.takeUntilOther(endMono);
 	}
 
 	public static Scheduler newBoundedSingle(String name) {