You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@james.apache.org by bt...@apache.org on 2022/04/08 02:00:29 UTC
[james-project] 02/03: [REACTOR] Fix deprecations: remove UnicastProcessor uses.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit a0cf145d2f7cebea66d0a3ab89f343411f57a3a1
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Tue Apr 5 22:51:22 2022 +0700
[REACTOR] Fix deprecations: remove UnicastProcessor uses.
---
.../java/org/apache/james/GuiceLifecycleHeathCheckTest.java | 12 ++++++------
.../distributed/RabbitMQTerminationSubscriber.java | 11 ++++++-----
.../task/eventsourcing/distributed/RabbitMQWorkQueue.java | 11 ++++++-----
.../src/main/java/org/apache/james/task/MemoryWorkQueue.java | 12 +++++++-----
4 files changed, 25 insertions(+), 21 deletions(-)
diff --git a/server/apps/memory-app/src/test/java/org/apache/james/GuiceLifecycleHeathCheckTest.java b/server/apps/memory-app/src/test/java/org/apache/james/GuiceLifecycleHeathCheckTest.java
index bc36f2c857..6b4613c8ba 100644
--- a/server/apps/memory-app/src/test/java/org/apache/james/GuiceLifecycleHeathCheckTest.java
+++ b/server/apps/memory-app/src/test/java/org/apache/james/GuiceLifecycleHeathCheckTest.java
@@ -23,6 +23,7 @@ import static io.restassured.RestAssured.when;
import static io.restassured.config.EncoderConfig.encoderConfig;
import static io.restassured.config.RestAssuredConfig.newConfig;
import static org.apache.james.data.UsersRepositoryModuleChooser.Implementation.DEFAULT;
+import static reactor.core.publisher.Sinks.EmitFailureHandler.FAIL_FAST;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.CountDownLatch;
@@ -42,7 +43,7 @@ import io.restassured.RestAssured;
import io.restassured.builder.RequestSpecBuilder;
import io.restassured.http.ContentType;
import reactor.core.publisher.Mono;
-import reactor.core.publisher.MonoProcessor;
+import reactor.core.publisher.Sinks;
import reactor.core.scheduler.Schedulers;
class GuiceLifecycleHeathCheckTest {
@@ -117,14 +118,13 @@ class GuiceLifecycleHeathCheckTest {
@Test
void stoppingJamesServerShouldBeUnhealthy(GuiceJamesServer server) {
- Mono<Void> stopMono = Mono.fromRunnable(() -> { });
+ Sinks.Empty<Object> sink = Sinks.empty();
try {
configureRequestSpecification(server);
- stopMono = Mono.fromRunnable(server::stop);
- stopMono
+ Mono.fromRunnable(server::stop)
.publishOn(Schedulers.elastic())
- .subscribeWith(MonoProcessor.create());
+ .subscribe(r -> {}, e -> {}, () -> sink.emitEmpty(FAIL_FAST));
when()
.get("/healthcheck")
@@ -132,7 +132,7 @@ class GuiceLifecycleHeathCheckTest {
.statusCode(HttpStatus.SERVICE_UNAVAILABLE_503);
} finally {
latch.countDown();
- stopMono.block();
+ sink.asMono().block();
}
}
}
diff --git a/server/task/task-distributed/src/main/java/org/apache/james/task/eventsourcing/distributed/RabbitMQTerminationSubscriber.java b/server/task/task-distributed/src/main/java/org/apache/james/task/eventsourcing/distributed/RabbitMQTerminationSubscriber.java
index 9173e83f93..1e9de864de 100644
--- a/server/task/task-distributed/src/main/java/org/apache/james/task/eventsourcing/distributed/RabbitMQTerminationSubscriber.java
+++ b/server/task/task-distributed/src/main/java/org/apache/james/task/eventsourcing/distributed/RabbitMQTerminationSubscriber.java
@@ -23,6 +23,7 @@ package org.apache.james.task.eventsourcing.distributed;
import static org.apache.james.backends.rabbitmq.Constants.AUTO_DELETE;
import static org.apache.james.backends.rabbitmq.Constants.DURABLE;
import static org.apache.james.util.ReactorUtils.publishIfPresent;
+import static reactor.core.publisher.Sinks.EmitFailureHandler.FAIL_FAST;
import java.io.Closeable;
import java.nio.charset.StandardCharsets;
@@ -47,7 +48,7 @@ import com.rabbitmq.client.Delivery;
import reactor.core.Disposable;
import reactor.core.publisher.DirectProcessor;
import reactor.core.publisher.Flux;
-import reactor.core.publisher.UnicastProcessor;
+import reactor.core.publisher.Sinks;
import reactor.core.scheduler.Schedulers;
import reactor.rabbitmq.BindingSpecification;
import reactor.rabbitmq.ExchangeSpecification;
@@ -66,7 +67,7 @@ public class RabbitMQTerminationSubscriber implements TerminationSubscriber, Sta
private final JsonEventSerializer serializer;
private final Sender sender;
private final ReceiverProvider receiverProvider;
- private UnicastProcessor<OutboundMessage> sendQueue;
+ private Sinks.Many<OutboundMessage> sendQueue;
private DirectProcessor<Event> listener;
private Disposable sendQueueHandle;
private Disposable listenQueueHandle;
@@ -83,9 +84,9 @@ public class RabbitMQTerminationSubscriber implements TerminationSubscriber, Sta
sender.declareExchange(ExchangeSpecification.exchange(EXCHANGE_NAME)).block();
sender.declare(QueueSpecification.queue(queueName.asString()).durable(!DURABLE).autoDelete(AUTO_DELETE)).block();
sender.bind(BindingSpecification.binding(EXCHANGE_NAME, ROUTING_KEY, queueName.asString())).block();
- sendQueue = UnicastProcessor.create();
+ sendQueue = Sinks.many().unicast().onBackpressureBuffer();
sendQueueHandle = sender
- .send(sendQueue)
+ .send(sendQueue.asFlux())
.subscribeOn(Schedulers.elastic())
.subscribe();
@@ -116,7 +117,7 @@ public class RabbitMQTerminationSubscriber implements TerminationSubscriber, Sta
byte[] payload = serializer.serialize(event).getBytes(StandardCharsets.UTF_8);
AMQP.BasicProperties basicProperties = new AMQP.BasicProperties.Builder().build();
OutboundMessage message = new OutboundMessage(EXCHANGE_NAME, ROUTING_KEY, basicProperties, payload);
- sendQueue.onNext(message);
+ sendQueue.emitNext(message, FAIL_FAST);
} catch (JsonProcessingException e) {
throw new RuntimeException(e);
}
diff --git a/server/task/task-distributed/src/main/java/org/apache/james/task/eventsourcing/distributed/RabbitMQWorkQueue.java b/server/task/task-distributed/src/main/java/org/apache/james/task/eventsourcing/distributed/RabbitMQWorkQueue.java
index 60c9a3b5ae..c85f5fe823 100644
--- a/server/task/task-distributed/src/main/java/org/apache/james/task/eventsourcing/distributed/RabbitMQWorkQueue.java
+++ b/server/task/task-distributed/src/main/java/org/apache/james/task/eventsourcing/distributed/RabbitMQWorkQueue.java
@@ -24,6 +24,7 @@ import static com.rabbitmq.client.MessageProperties.PERSISTENT_TEXT_PLAIN;
import static org.apache.james.backends.rabbitmq.Constants.AUTO_DELETE;
import static org.apache.james.backends.rabbitmq.Constants.DURABLE;
import static org.apache.james.backends.rabbitmq.Constants.REQUEUE;
+import static reactor.core.publisher.Sinks.EmitFailureHandler.FAIL_FAST;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
@@ -49,7 +50,7 @@ import com.rabbitmq.client.Delivery;
import reactor.core.Disposable;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
-import reactor.core.publisher.UnicastProcessor;
+import reactor.core.publisher.Sinks;
import reactor.core.scheduler.Schedulers;
import reactor.rabbitmq.AcknowledgableDelivery;
import reactor.rabbitmq.BindingSpecification;
@@ -81,7 +82,7 @@ public class RabbitMQWorkQueue implements WorkQueue {
private final Sender sender;
private final ReceiverProvider receiverProvider;
private final CancelRequestQueueName cancelRequestQueueName;
- private UnicastProcessor<TaskId> sendCancelRequestsQueue;
+ private Sinks.Many<TaskId> sendCancelRequestsQueue;
private Disposable sendCancelRequestsQueueHandle;
private Disposable receiverHandle;
private Disposable cancelRequestListenerHandle;
@@ -191,9 +192,9 @@ public class RabbitMQWorkQueue implements WorkQueue {
sender.bind(BindingSpecification.binding(CANCEL_REQUESTS_EXCHANGE_NAME, CANCEL_REQUESTS_ROUTING_KEY, cancelRequestQueueName.asString())).block();
registerCancelRequestsListener(cancelRequestQueueName.asString());
- sendCancelRequestsQueue = UnicastProcessor.create();
+ sendCancelRequestsQueue = Sinks.many().unicast().onBackpressureBuffer();
sendCancelRequestsQueueHandle = sender
- .send(sendCancelRequestsQueue.map(this::makeCancelRequestMessage))
+ .send(sendCancelRequestsQueue.asFlux().map(this::makeCancelRequestMessage))
.subscribeOn(Schedulers.elastic())
.subscribe();
}
@@ -240,7 +241,7 @@ public class RabbitMQWorkQueue implements WorkQueue {
@Override
public void cancel(TaskId taskId) {
- sendCancelRequestsQueue.onNext(taskId);
+ sendCancelRequestsQueue.emitNext(taskId, FAIL_FAST);
}
@Override
diff --git a/server/task/task-memory/src/main/java/org/apache/james/task/MemoryWorkQueue.java b/server/task/task-memory/src/main/java/org/apache/james/task/MemoryWorkQueue.java
index 71275f2a59..b48b54a239 100644
--- a/server/task/task-memory/src/main/java/org/apache/james/task/MemoryWorkQueue.java
+++ b/server/task/task-memory/src/main/java/org/apache/james/task/MemoryWorkQueue.java
@@ -19,22 +19,24 @@
package org.apache.james.task;
+import static reactor.core.publisher.Sinks.EmitFailureHandler.FAIL_FAST;
+
import java.io.IOException;
import reactor.core.Disposable;
import reactor.core.publisher.Mono;
-import reactor.core.publisher.UnicastProcessor;
+import reactor.core.publisher.Sinks;
import reactor.core.scheduler.Schedulers;
public class MemoryWorkQueue implements WorkQueue {
private final TaskManagerWorker worker;
private final Disposable subscription;
- private final UnicastProcessor<TaskWithId> tasks;
+ private final Sinks.Many<TaskWithId> tasks;
public MemoryWorkQueue(TaskManagerWorker worker) {
this.worker = worker;
- this.tasks = UnicastProcessor.create();
- this.subscription = tasks
+ this.tasks = Sinks.many().unicast().onBackpressureBuffer();
+ this.subscription = tasks.asFlux()
.subscribeOn(Schedulers.elastic())
.limitRate(1)
.concatMap(this::dispatchTaskToWorker)
@@ -46,7 +48,7 @@ public class MemoryWorkQueue implements WorkQueue {
}
public void submit(TaskWithId taskWithId) {
- tasks.onNext(taskWithId);
+ tasks.emitNext(taskWithId, FAIL_FAST);
}
public void cancel(TaskId taskId) {
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org