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/19 04:04:27 UTC
[james-project] 36/49: JAMES-3737 IMAP: Reactify SUBSCRIBE / UNSUBSCRIBE
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 9262d07c1238fec59c6de6de451c0de88dc85a40
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Tue Apr 5 09:13:30 2022 +0700
JAMES-3737 IMAP: Reactify SUBSCRIBE / UNSUBSCRIBE
---
.../apache/james/mailbox/SubscriptionManager.java | 3 +++
.../user/CassandraSubscriptionMapper.java | 15 ++++++++++++
.../mailbox/store/StoreSubscriptionManager.java | 24 +++++++++++++++++++
.../mailbox/store/user/SubscriptionMapper.java | 9 ++++++++
.../processor/AbstractSubscriptionProcessor.java | 17 +++++++-------
.../james/imap/processor/SubscribeProcessor.java | 26 ++++++++++-----------
.../james/imap/processor/UnsubscribeProcessor.java | 27 +++++++++++-----------
7 files changed, 85 insertions(+), 36 deletions(-)
diff --git a/mailbox/api/src/main/java/org/apache/james/mailbox/SubscriptionManager.java b/mailbox/api/src/main/java/org/apache/james/mailbox/SubscriptionManager.java
index 1d102059c9..1730c3c94b 100644
--- a/mailbox/api/src/main/java/org/apache/james/mailbox/SubscriptionManager.java
+++ b/mailbox/api/src/main/java/org/apache/james/mailbox/SubscriptionManager.java
@@ -43,6 +43,8 @@ public interface SubscriptionManager extends RequestAware {
*/
void subscribe(MailboxSession session, String mailbox) throws SubscriptionException;
+ Publisher<Void> subscribeReactive(String mailbox, MailboxSession session);
+
/**
* Finds all subscriptions for the user in the session.
*
@@ -68,4 +70,5 @@ public interface SubscriptionManager extends RequestAware {
*/
void unsubscribe(MailboxSession session, String mailbox) throws SubscriptionException;
+ Publisher<Void> unsubscribeReactive(String mailbox, MailboxSession session);
}
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/user/CassandraSubscriptionMapper.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/user/CassandraSubscriptionMapper.java
index 62d7b285f1..c7a176de5d 100644
--- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/user/CassandraSubscriptionMapper.java
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/user/CassandraSubscriptionMapper.java
@@ -42,6 +42,7 @@ import com.datastax.driver.core.Session;
import com.datastax.driver.core.querybuilder.QueryBuilder;
import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
public class CassandraSubscriptionMapper extends NonTransactionalMapper implements SubscriptionMapper {
private final Session session;
@@ -98,4 +99,18 @@ public class CassandraSubscriptionMapper extends NonTransactionalMapper implemen
.setString(USER, subscription.getUser().asString())
.setString(MAILBOX, subscription.getMailbox()));
}
+
+ @Override
+ public Mono<Void> saveReactive(Subscription subscription) {
+ return executor.executeVoid(deleteStatement.bind()
+ .setString(USER, subscription.getUser().asString())
+ .setString(MAILBOX, subscription.getMailbox()));
+ }
+
+ @Override
+ public Mono<Void> deleteReactive(Subscription subscription) {
+ return executor.executeVoid(insertStatement.bind()
+ .setString(USER, subscription.getUser().asString())
+ .setString(MAILBOX, subscription.getMailbox()));
+ }
}
diff --git a/mailbox/store/src/main/java/org/apache/james/mailbox/store/StoreSubscriptionManager.java b/mailbox/store/src/main/java/org/apache/james/mailbox/store/StoreSubscriptionManager.java
index 1d211fcb48..d3e16e1212 100644
--- a/mailbox/store/src/main/java/org/apache/james/mailbox/store/StoreSubscriptionManager.java
+++ b/mailbox/store/src/main/java/org/apache/james/mailbox/store/StoreSubscriptionManager.java
@@ -35,6 +35,8 @@ import org.apache.james.mailbox.store.user.SubscriptionMapperFactory;
import org.apache.james.mailbox.store.user.model.Subscription;
import org.reactivestreams.Publisher;
+import reactor.core.publisher.Mono;
+
/**
* Manages subscriptions for Users and Mailboxes.
*/
@@ -61,6 +63,28 @@ public class StoreSubscriptionManager implements SubscriptionManager {
}
}
+ @Override
+ public Publisher<Void> subscribeReactive(String mailbox, MailboxSession session) {
+ try {
+ SubscriptionMapper mapper = mapperFactory.getSubscriptionMapper(session);
+ Subscription newSubscription = new Subscription(session.getUser(), mailbox);
+ return mapper.executeReactive(mapper.saveReactive(newSubscription));
+ } catch (SubscriptionException e) {
+ return Mono.error(e);
+ }
+ }
+
+ @Override
+ public Publisher<Void> unsubscribeReactive(String mailbox, MailboxSession session) {
+ try {
+ SubscriptionMapper mapper = mapperFactory.getSubscriptionMapper(session);
+ Subscription oldSubscription = new Subscription(session.getUser(), mailbox);
+ return mapper.executeReactive(mapper.deleteReactive(oldSubscription));
+ } catch (SubscriptionException e) {
+ return Mono.error(e);
+ }
+ }
+
@Override
public Collection<String> subscriptions(MailboxSession session) throws SubscriptionException {
return mapperFactory.getSubscriptionMapper(session)
diff --git a/mailbox/store/src/main/java/org/apache/james/mailbox/store/user/SubscriptionMapper.java b/mailbox/store/src/main/java/org/apache/james/mailbox/store/user/SubscriptionMapper.java
index aacd787019..6f0f233d95 100644
--- a/mailbox/store/src/main/java/org/apache/james/mailbox/store/user/SubscriptionMapper.java
+++ b/mailbox/store/src/main/java/org/apache/james/mailbox/store/user/SubscriptionMapper.java
@@ -25,6 +25,7 @@ import org.apache.james.mailbox.exception.SubscriptionException;
import org.apache.james.mailbox.store.transaction.Mapper;
import org.apache.james.mailbox.store.user.model.Subscription;
+import com.github.fge.lambdas.Throwing;
import com.google.common.base.Functions;
import reactor.core.publisher.Flux;
@@ -41,6 +42,10 @@ public interface SubscriptionMapper extends Mapper {
*/
void save(Subscription subscription) throws SubscriptionException;
+ default Mono<Void> saveReactive(Subscription subscription) {
+ return Mono.fromRunnable(Throwing.runnable(() -> save(subscription)));
+ }
+
/**
* Finds subscriptions for the given user.
* @param user not null
@@ -58,4 +63,8 @@ public interface SubscriptionMapper extends Mapper {
* @param subscription not null
*/
void delete(Subscription subscription) throws SubscriptionException;
+
+ default Mono<Void> deleteReactive(Subscription subscription) {
+ return Mono.fromRunnable(Throwing.runnable(() -> delete(subscription)));
+ }
}
\ No newline at end of file
diff --git a/protocols/imap/src/main/java/org/apache/james/imap/processor/AbstractSubscriptionProcessor.java b/protocols/imap/src/main/java/org/apache/james/imap/processor/AbstractSubscriptionProcessor.java
index 901b3c9968..06b863359c 100644
--- a/protocols/imap/src/main/java/org/apache/james/imap/processor/AbstractSubscriptionProcessor.java
+++ b/protocols/imap/src/main/java/org/apache/james/imap/processor/AbstractSubscriptionProcessor.java
@@ -26,6 +26,8 @@ import org.apache.james.mailbox.MailboxSession;
import org.apache.james.mailbox.SubscriptionManager;
import org.apache.james.metrics.api.MetricFactory;
+import reactor.core.publisher.Mono;
+
/**
* Abstract base class which should be used by implementations which need to
* access the {@link SubscriptionManager}
@@ -50,22 +52,19 @@ public abstract class AbstractSubscriptionProcessor<R extends ImapRequest> exten
}
@Override
- protected final void processRequest(R request, ImapSession session, Responder responder) {
+ protected final Mono<Void> processRequestReactive(R request, ImapSession session, Responder responder) {
// take care of calling the start/end processing
MailboxSession mSession = session.getMailboxSession();
- getSubscriptionManager().startProcessingRequest(mSession);
- doProcessRequest(request, session, responder);
- getSubscriptionManager().endProcessingRequest(mSession);
+
+ return Mono.fromRunnable(() -> getSubscriptionManager().startProcessingRequest(mSession))
+ .then(doProcessRequest(request, session, responder))
+ .then(Mono.fromRunnable(() -> getSubscriptionManager().endProcessingRequest(mSession)));
}
/**
* Process the request
- *
- * @param request
- * @param session
- * @param responder
*/
- protected abstract void doProcessRequest(R request, ImapSession session, Responder responder);
+ protected abstract Mono<Void> doProcessRequest(R request, ImapSession session, Responder responder);
}
diff --git a/protocols/imap/src/main/java/org/apache/james/imap/processor/SubscribeProcessor.java b/protocols/imap/src/main/java/org/apache/james/imap/processor/SubscribeProcessor.java
index c13175018f..12039473a9 100644
--- a/protocols/imap/src/main/java/org/apache/james/imap/processor/SubscribeProcessor.java
+++ b/protocols/imap/src/main/java/org/apache/james/imap/processor/SubscribeProcessor.java
@@ -34,6 +34,8 @@ import org.apache.james.util.MDCBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import reactor.core.publisher.Mono;
+
public class SubscribeProcessor extends AbstractSubscriptionProcessor<SubscribeRequest> {
private static final Logger LOGGER = LoggerFactory.getLogger(SubscribeProcessor.class);
@@ -43,20 +45,18 @@ public class SubscribeProcessor extends AbstractSubscriptionProcessor<SubscribeR
}
@Override
- protected void doProcessRequest(SubscribeRequest request, ImapSession session, Responder responder) {
- final String mailboxName = request.getMailboxName();
- final MailboxSession mailboxSession = session.getMailboxSession();
- try {
- getSubscriptionManager().subscribe(mailboxSession, mailboxName);
-
- unsolicitedResponses(session, responder, false).block();
- okComplete(request, responder);
+ protected Mono<Void> doProcessRequest(SubscribeRequest request, ImapSession session, Responder responder) {
+ String mailboxName = request.getMailboxName();
+ MailboxSession mailboxSession = session.getMailboxSession();
- } catch (SubscriptionException e) {
- LOGGER.info("Subscribe failed for mailbox {}", mailboxName, e);
- unsolicitedResponses(session, responder, false).block();
- no(request, responder, HumanReadableText.GENERIC_SUBSCRIPTION_FAILURE);
- }
+ return Mono.from(getSubscriptionManager().subscribeReactive(mailboxName, mailboxSession))
+ .then(unsolicitedResponses(session, responder, false))
+ .then(Mono.fromRunnable(() -> okComplete(request, responder)))
+ .onErrorResume(SubscriptionException.class, e -> {
+ LOGGER.info("Subscribe failed for mailbox {}", mailboxName, e);
+ return unsolicitedResponses(session, responder, false)
+ .then(Mono.fromRunnable(() -> no(request, responder, HumanReadableText.GENERIC_SUBSCRIPTION_FAILURE)));
+ }).then();
}
@Override
diff --git a/protocols/imap/src/main/java/org/apache/james/imap/processor/UnsubscribeProcessor.java b/protocols/imap/src/main/java/org/apache/james/imap/processor/UnsubscribeProcessor.java
index 8d375ba899..5b4f7ebcd4 100644
--- a/protocols/imap/src/main/java/org/apache/james/imap/processor/UnsubscribeProcessor.java
+++ b/protocols/imap/src/main/java/org/apache/james/imap/processor/UnsubscribeProcessor.java
@@ -34,6 +34,8 @@ import org.apache.james.util.MDCBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import reactor.core.publisher.Mono;
+
public class UnsubscribeProcessor extends AbstractSubscriptionProcessor<UnsubscribeRequest> {
private static final Logger LOGGER = LoggerFactory.getLogger(UnsubscribeProcessor.class);
@@ -43,21 +45,18 @@ public class UnsubscribeProcessor extends AbstractSubscriptionProcessor<Unsubscr
}
@Override
- protected void doProcessRequest(UnsubscribeRequest request, ImapSession session, Responder responder) {
- final String mailboxName = request.getMailboxName();
- final MailboxSession mailboxSession = session.getMailboxSession();
- try {
- getSubscriptionManager().unsubscribe(mailboxSession, mailboxName);
-
- unsolicitedResponses(session, responder, false).block();
- okComplete(request, responder);
-
- } catch (SubscriptionException e) {
- LOGGER.info("Unsubscribe failed for mailbox {}", mailboxName, e);
- unsolicitedResponses(session, responder, false).block();
+ protected Mono<Void> doProcessRequest(UnsubscribeRequest request, ImapSession session, Responder responder) {
+ String mailboxName = request.getMailboxName();
+ MailboxSession mailboxSession = session.getMailboxSession();
- no(request, responder, HumanReadableText.GENERIC_SUBSCRIPTION_FAILURE);
- }
+ return Mono.from(getSubscriptionManager().unsubscribeReactive(mailboxName, mailboxSession))
+ .then(unsolicitedResponses(session, responder, false))
+ .then(Mono.fromRunnable(() -> okComplete(request, responder)))
+ .onErrorResume(SubscriptionException.class, e -> {
+ LOGGER.info("Unsubscribe failed for mailbox {}", mailboxName, e);
+ return unsolicitedResponses(session, responder, false)
+ .then(Mono.fromRunnable(() -> no(request, responder, HumanReadableText.GENERIC_SUBSCRIPTION_FAILURE)));
+ }).then();
}
@Override
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org