[ENHANCEMENT] Better reactify Identity methods
diff --git a/server/data/data-api/src/main/java/org/apache/james/rrt/api/AliasReverseResolver.java b/server/data/data-api/src/main/java/org/apache/james/rrt/api/AliasReverseResolver.java
index 05117c5..0967209 100644
--- a/server/data/data-api/src/main/java/org/apache/james/rrt/api/AliasReverseResolver.java
+++ b/server/data/data-api/src/main/java/org/apache/james/rrt/api/AliasReverseResolver.java
@@ -19,11 +19,10 @@
package org.apache.james.rrt.api;
-import java.util.stream.Stream;
-
import org.apache.james.core.MailAddress;
import org.apache.james.core.Username;
+import org.reactivestreams.Publisher;
public interface AliasReverseResolver {
- Stream<MailAddress> listAddresses(Username user) throws RecipientRewriteTable.ErrorMappingException, RecipientRewriteTableException;
+ Publisher<MailAddress> listAddresses(Username user) throws RecipientRewriteTable.ErrorMappingException, RecipientRewriteTableException;
}
diff --git a/server/data/data-api/src/main/java/org/apache/james/rrt/api/CanSendFrom.java b/server/data/data-api/src/main/java/org/apache/james/rrt/api/CanSendFrom.java
index 9fee906..582376e 100644
--- a/server/data/data-api/src/main/java/org/apache/james/rrt/api/CanSendFrom.java
+++ b/server/data/data-api/src/main/java/org/apache/james/rrt/api/CanSendFrom.java
@@ -18,8 +18,6 @@
****************************************************************/
package org.apache.james.rrt.api;
-import java.util.stream.Stream;
-
import org.apache.james.core.MailAddress;
import org.apache.james.core.Username;
import org.reactivestreams.Publisher;
@@ -36,6 +34,6 @@
/**
* For a given user, return all the addresses he can use in the from clause of an email.
*/
- Stream<MailAddress> allValidFromAddressesForUser(Username user) throws RecipientRewriteTable.ErrorMappingException, RecipientRewriteTableException;
+ Publisher<MailAddress> allValidFromAddressesForUser(Username user) throws RecipientRewriteTable.ErrorMappingException, RecipientRewriteTableException;
}
diff --git a/server/data/data-api/src/main/java/org/apache/james/rrt/api/RecipientRewriteTable.java b/server/data/data-api/src/main/java/org/apache/james/rrt/api/RecipientRewriteTable.java
index e7506b2..01078df 100644
--- a/server/data/data-api/src/main/java/org/apache/james/rrt/api/RecipientRewriteTable.java
+++ b/server/data/data-api/src/main/java/org/apache/james/rrt/api/RecipientRewriteTable.java
@@ -31,6 +31,8 @@
import com.google.common.base.Preconditions;
+import reactor.core.publisher.Flux;
+
/**
* Interface which should be implemented of classes which map recipients.
*/
@@ -138,6 +140,14 @@
.map(Map.Entry::getKey);
}
+ default Flux<MappingSource> listSourcesReactive(Mapping mapping) {
+ try {
+ return Flux.fromStream(listSources(mapping));
+ } catch (RecipientRewriteTableException e) {
+ return Flux.error(e);
+ }
+ }
+
default Stream<MappingSource> getSourcesForType(Mapping.Type type) throws RecipientRewriteTableException {
return getAllMappings()
.entrySet().stream()
diff --git a/server/data/data-api/src/test/java/org/apache/james/rrt/lib/AliasReverseResolverContract.java b/server/data/data-api/src/test/java/org/apache/james/rrt/lib/AliasReverseResolverContract.java
index 891fe46..3504256 100644
--- a/server/data/data-api/src/test/java/org/apache/james/rrt/lib/AliasReverseResolverContract.java
+++ b/server/data/data-api/src/test/java/org/apache/james/rrt/lib/AliasReverseResolverContract.java
@@ -30,6 +30,8 @@
import com.github.fge.lambdas.Throwing;
+import reactor.core.publisher.Flux;
+
public interface AliasReverseResolverContract {
Domain DOMAIN = Domain.of("example.com");
@@ -66,14 +68,14 @@
@Test
default void listAddressesShouldContainOnlyUserAddressWhenUserHasNoAlias() throws Exception {
- assertThat(aliasReverseResolver().listAddresses(USER))
+ assertThat(Flux.from(aliasReverseResolver().listAddresses(USER)).toStream())
.containsExactly(USER.asMailAddress());
}
@Test
default void listAddressesShouldContainOnlyUserAddressWhenUserHasNoAliasAndAnotherUserHasOne() throws Exception {
redirectUser(USER_ALIAS).to(OTHER_USER);
- assertThat(aliasReverseResolver().listAddresses(USER))
+ assertThat(Flux.from(aliasReverseResolver().listAddresses(USER)).toStream())
.containsExactly(USER.asMailAddress());
}
@@ -81,7 +83,7 @@
default void listAddressesShouldContainUserAddressAndAnAliasOfTheUser() throws Exception {
redirectUser(USER_ALIAS).to(USER);
- assertThat(aliasReverseResolver().listAddresses(USER))
+ assertThat(Flux.from(aliasReverseResolver().listAddresses(USER)).toStream())
.containsExactlyInAnyOrder(USER.asMailAddress(), USER_ALIAS.asMailAddress());
}
@@ -91,7 +93,7 @@
redirectUser(userAliasBis).to(USER_ALIAS);
redirectUser(USER_ALIAS).to(USER);
- assertThat(aliasReverseResolver().listAddresses(USER))
+ assertThat(Flux.from(aliasReverseResolver().listAddresses(USER)).toStream())
.containsExactlyInAnyOrder(USER.asMailAddress(), USER_ALIAS.asMailAddress(), userAliasBis.asMailAddress());
}
@@ -101,7 +103,7 @@
redirectDomain(OTHER_DOMAIN).to(DOMAIN);
- assertThat(aliasReverseResolver().listAddresses(USER))
+ assertThat(Flux.from(aliasReverseResolver().listAddresses(USER)).toStream())
.containsExactlyInAnyOrder(USER.asMailAddress(), fromUser.asMailAddress());
}
@@ -114,7 +116,7 @@
Username userAliasMainDomain = USER_ALIAS.withOtherDomain(DOMAIN);
Username userOtherDomain = USER.withOtherDomain(OTHER_DOMAIN);
- assertThat(aliasReverseResolver().listAddresses(USER))
+ assertThat(Flux.from(aliasReverseResolver().listAddresses(USER)).toStream())
.containsExactlyInAnyOrder(USER.asMailAddress(), userAliasOtherDomain.asMailAddress(), userAliasMainDomain.asMailAddress(), userOtherDomain.asMailAddress());
}
@@ -136,7 +138,7 @@
Username userAliasExcluded = Username.of("alias" + (recursionLevel - 1) + "@" + DOMAIN.asString());
redirectUser(USER_ALIAS).to(USER);
- assertThat(aliasReverseResolver().listAddresses(USER))
+ assertThat(Flux.from(aliasReverseResolver().listAddresses(USER)).toStream())
.doesNotContain(userAliasExcluded.asMailAddress());
}
@@ -147,7 +149,7 @@
redirectUser(userAliasBis).to(userAlias);
redirectUser(userAlias).to(USER);
- assertThat(aliasReverseResolver().listAddresses(USER))
+ assertThat(Flux.from(aliasReverseResolver().listAddresses(USER)).toStream())
.contains(userAliasBis.asMailAddress());
}
@@ -157,7 +159,7 @@
redirectUser(userAliasBis).to(USER_ALIAS);
redirectUser(USER_ALIAS).to(USER);
- assertThat(aliasReverseResolver().listAddresses(USER))
+ assertThat(Flux.from(aliasReverseResolver().listAddresses(USER)).toStream())
.contains(userAliasBis.asMailAddress());
}
}
diff --git a/server/data/data-api/src/test/java/org/apache/james/rrt/lib/CanSendFromContract.java b/server/data/data-api/src/test/java/org/apache/james/rrt/lib/CanSendFromContract.java
index 3262131..1127efd 100644
--- a/server/data/data-api/src/test/java/org/apache/james/rrt/lib/CanSendFromContract.java
+++ b/server/data/data-api/src/test/java/org/apache/james/rrt/lib/CanSendFromContract.java
@@ -29,6 +29,8 @@
import com.github.fge.lambdas.Throwing;
+import reactor.core.publisher.Flux;
+
public interface CanSendFromContract {
Domain DOMAIN = Domain.of("example.com");
@@ -173,14 +175,14 @@
@Test
default void allValidFromAddressesShouldContainOnlyUserAddressWhenUserHasNoAlias() throws Exception {
- assertThat(canSendFrom().allValidFromAddressesForUser(USER))
+ assertThat(Flux.from(canSendFrom().allValidFromAddressesForUser(USER)).toStream())
.containsExactly(USER.asMailAddress());
}
@Test
default void allValidFromAddressesShouldContainOnlyUserAddressWhenUserHasNoAliasAndAnotherUserHasOne() throws Exception {
redirectUser(USER_ALIAS).to(OTHER_USER);
- assertThat(canSendFrom().allValidFromAddressesForUser(USER))
+ assertThat(Flux.from(canSendFrom().allValidFromAddressesForUser(USER)).toStream())
.containsExactly(USER.asMailAddress());
}
@@ -188,7 +190,7 @@
default void allValidFromAddressesShouldContainUserAddressAndAnAliasOfTheUser() throws Exception {
redirectUser(USER_ALIAS).to(USER);
- assertThat(canSendFrom().allValidFromAddressesForUser(USER))
+ assertThat(Flux.from(canSendFrom().allValidFromAddressesForUser(USER)).toStream())
.containsExactlyInAnyOrder(USER.asMailAddress(), USER_ALIAS.asMailAddress());
}
@@ -198,7 +200,7 @@
redirectUser(userAliasBis).to(USER_ALIAS);
redirectUser(USER_ALIAS).to(USER);
- assertThat(canSendFrom().allValidFromAddressesForUser(USER))
+ assertThat(Flux.from(canSendFrom().allValidFromAddressesForUser(USER)).toStream())
.containsExactlyInAnyOrder(USER.asMailAddress(), USER_ALIAS.asMailAddress(), userAliasBis.asMailAddress());
}
@@ -208,7 +210,7 @@
redirectDomain(OTHER_DOMAIN).to(DOMAIN).asAlias();
- assertThat(canSendFrom().allValidFromAddressesForUser(USER))
+ assertThat(Flux.from(canSendFrom().allValidFromAddressesForUser(USER)).toStream())
.containsExactlyInAnyOrder(USER.asMailAddress(), fromUser.asMailAddress());
}
@@ -221,7 +223,7 @@
Username userAliasMainDomain = USER_ALIAS.withOtherDomain(DOMAIN);
Username userOtherDomain = USER.withOtherDomain(OTHER_DOMAIN);
- assertThat(canSendFrom().allValidFromAddressesForUser(USER))
+ assertThat(Flux.from(canSendFrom().allValidFromAddressesForUser(USER)).toStream())
.containsExactlyInAnyOrder(USER.asMailAddress(), userAliasOtherDomain.asMailAddress(), userAliasMainDomain.asMailAddress(), userOtherDomain.asMailAddress());
}
@@ -252,7 +254,7 @@
Username userAliasExcluded = Username.of("alias" + (recursionLevel - 1) + "@" + DOMAIN.asString());
redirectUser(USER_ALIAS).to(USER);
- assertThat(canSendFrom().allValidFromAddressesForUser(USER))
+ assertThat(Flux.from(canSendFrom().allValidFromAddressesForUser(USER)).toStream())
.doesNotContain(userAliasExcluded.asMailAddress());
}
@@ -263,7 +265,7 @@
redirectUser(userAliasBis).to(userAlias);
redirectUser(userAlias).to(USER);
- assertThat(canSendFrom().allValidFromAddressesForUser(USER))
+ assertThat(Flux.from(canSendFrom().allValidFromAddressesForUser(USER)).toStream())
.contains(userAliasBis.asMailAddress());
}
@@ -273,7 +275,7 @@
redirectUser(userAliasBis).to(USER_ALIAS);
redirectUser(USER_ALIAS).to(USER);
- assertThat(canSendFrom().allValidFromAddressesForUser(USER))
+ assertThat(Flux.from(canSendFrom().allValidFromAddressesForUser(USER)).toStream())
.contains(userAliasBis.asMailAddress());
}
}
diff --git a/server/data/data-cassandra/src/main/java/org/apache/james/rrt/cassandra/CassandraRecipientRewriteTable.java b/server/data/data-cassandra/src/main/java/org/apache/james/rrt/cassandra/CassandraRecipientRewriteTable.java
index deacb92..6eda5a3 100644
--- a/server/data/data-cassandra/src/main/java/org/apache/james/rrt/cassandra/CassandraRecipientRewriteTable.java
+++ b/server/data/data-cassandra/src/main/java/org/apache/james/rrt/cassandra/CassandraRecipientRewriteTable.java
@@ -35,6 +35,8 @@
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableMap;
+import reactor.core.publisher.Flux;
+
public class CassandraRecipientRewriteTable extends AbstractRecipientRewriteTable {
private final CassandraRecipientRewriteTableDAO cassandraRecipientRewriteTableDAO;
private final CassandraMappingsSourcesDAO cassandraMappingsSourcesDAO;
@@ -91,4 +93,11 @@
return cassandraMappingsSourcesDAO.retrieveSources(mapping).toStream();
}
+
+ @Override
+ public Flux<MappingSource> listSourcesReactive(Mapping mapping) {
+ Preconditions.checkArgument(listSourcesSupportedType.contains(mapping.getType()),
+ "Not supported mapping of type %s", mapping.getType());
+ return cassandraMappingsSourcesDAO.retrieveSources(mapping);
+ }
}
diff --git a/server/data/data-jmap/src/main/scala/org/apache/james/jmap/api/identity/CustomIdentityDAO.scala b/server/data/data-jmap/src/main/scala/org/apache/james/jmap/api/identity/CustomIdentityDAO.scala
index a2ee0d5..f746c2c 100644
--- a/server/data/data-jmap/src/main/scala/org/apache/james/jmap/api/identity/CustomIdentityDAO.scala
+++ b/server/data/data-jmap/src/main/scala/org/apache/james/jmap/api/identity/CustomIdentityDAO.scala
@@ -26,11 +26,9 @@
import org.apache.james.jmap.api.model.{EmailAddress, ForbiddenSendFromException, HtmlSignature, Identity, IdentityId, IdentityName, MayDeleteIdentity, TextSignature}
import org.apache.james.rrt.api.CanSendFrom
import org.apache.james.user.api.UsersRepository
-import org.apache.james.util.ReactorUtils
import org.reactivestreams.Publisher
import reactor.core.scala.publisher.{SFlux, SMono}
-import scala.jdk.StreamConverters._
import scala.util.Try
import scala.jdk.OptionConverters._
import scala.jdk.CollectionConverters._
@@ -171,9 +169,8 @@
}
class DefaultIdentitySupplier @Inject()(canSendFrom: CanSendFrom, usersRepository: UsersRepository) {
- def listIdentities(username: Username): List[Identity] = canSendFrom.allValidFromAddressesForUser(username)
- .toScala(LazyList).toList
- .flatMap(address =>
+ def listIdentities(username: Username): Publisher[Identity] = SFlux(canSendFrom.allValidFromAddressesForUser(username))
+ .map(address =>
from(address).map(id =>
Identity(
id = id,
@@ -185,14 +182,12 @@
htmlSignature = HtmlSignature.DEFAULT,
mayDelete = MayDeleteIdentity(false),
sortOrder = Identity.DEFAULT_SORTORDER)))
+ .flatMap(option => option.map(SMono.just).getOrElse(SMono.empty))
def userCanSendFrom(username: Username, mailAddress: MailAddress): SMono[Boolean] =
SMono.fromPublisher(canSendFrom.userCanSendFromReactive(username, usersRepository.getUsername(mailAddress)))
.map(boolean2Boolean(_))
- def isServerSetIdentity(username: Username, id: IdentityId): Boolean =
- listIdentities(username).map(_.id).contains(id)
-
private def from(address: MailAddress): Option[IdentityId] =
Try(UUID.nameUUIDFromBytes(address.asString().getBytes(StandardCharsets.UTF_8)))
.toEither
@@ -221,8 +216,8 @@
.map(_.identity)
private def listServerSetIdentity(user: Username): SMono[(Set[MailAddress], List[Identity])] =
- SMono.fromCallable(() => identityFactory.listIdentities(user))
- .subscribeOn(ReactorUtils.BLOCKING_CALL_WRAPPER)
+ SFlux(identityFactory.listIdentities(user))
+ .collectSeq().map(s => s.toList)
.map(list => (list.map(_.email).toSet, list))
private def listCustomIdentity(user: Username, availableMailAddresses: Set[MailAddress]): SFlux[Identity] =
@@ -230,9 +225,9 @@
.filter(identity => availableMailAddresses.contains(identity.email))
def update(user: Username, identityId: IdentityId, identityUpdateRequest: IdentityUpdateRequest): Publisher[Unit] = {
- val findServerSetIdentity: SMono[Option[Identity]] = SMono.fromCallable(() => identityFactory.listIdentities(user)
- .find(identity => identity.id.equals(identityId)))
- .subscribeOn(ReactorUtils.BLOCKING_CALL_WRAPPER)
+ val findServerSetIdentity: SMono[Option[Identity]] = SFlux(identityFactory.listIdentities(user))
+ .collectSeq().map(s => s.toList)
+ .map(list => list.find(identity => identity.id.equals(identityId)))
val findCustomIdentity: SMono[Option[Identity]] = SMono(customIdentityDao.findByIdentityId(user, identityId))
.map(Some(_))
.switchIfEmpty(SMono.just(None))
@@ -252,17 +247,20 @@
.`then`()
}
- def delete(username: Username, ids: Set[IdentityId]): Publisher[Unit] =
- SMono.just(ids)
- .handle[Set[IdentityId]]{
- case (ids, sink) => if (identityFactory.isServerSetIdentity(username, ids.head)) {
- sink.error(IdentityForbiddenDeleteException(ids.head))
+ def delete(username: Username, ids: Set[IdentityId]): Publisher[Unit] = {
+ SFlux(identityFactory.listIdentities(username))
+ .map(_.id)
+ .collectSeq()
+ .flatMapMany(serverSetIdentities => SFlux.fromIterable(ids)
+ .handle[IdentityId] {
+ case (id, sink) => if (serverSetIdentities.contains(id)) {
+ sink.error(IdentityForbiddenDeleteException(id))
} else {
- sink.next(ids)
+ sink.next(id)
}
- }
- .flatMap(ids => SMono.fromPublisher(customIdentityDao.delete(username, ids)))
- .subscribeOn(ReactorUtils.BLOCKING_CALL_WRAPPER)
+ }).collectSeq()
+ .flatMap(ids => SMono.fromPublisher(customIdentityDao.delete(username, ids.toSet)))
+ }
}
case class IdentityNotFoundException(id: IdentityId) extends RuntimeException(s"$id could not be found")
diff --git a/server/data/data-jmap/src/test/scala/org/apache/james/jmap/api/identity/IdentityRepositoryTest.scala b/server/data/data-jmap/src/test/scala/org/apache/james/jmap/api/identity/IdentityRepositoryTest.scala
index a5d28c2..1bc9785 100644
--- a/server/data/data-jmap/src/test/scala/org/apache/james/jmap/api/identity/IdentityRepositoryTest.scala
+++ b/server/data/data-jmap/src/test/scala/org/apache/james/jmap/api/identity/IdentityRepositoryTest.scala
@@ -99,7 +99,7 @@
@Test
def listShouldReturnCustomAndServerSetEntries(): Unit = {
val customIdentity: Identity = SMono.fromPublisher(customIdentityDAO.save(BOB, CREATION_REQUEST)).block()
- when(identityFactory.listIdentities(BOB)).thenReturn(List(IDENTITY1))
+ when(identityFactory.listIdentities(BOB)).thenReturn(SFlux.just(IDENTITY1))
assertThat(SFlux.fromPublisher(testee.list(BOB)).collectSeq().block().asJava)
.containsExactlyInAnyOrder(IDENTITY1, customIdentity)
@@ -109,7 +109,7 @@
def listShouldReturnCustomEntryWhenIdExistsInBothCustomAndServerSet(): Unit = {
val customIdentity: Identity = SMono.fromPublisher(customIdentityDAO.save(BOB, CREATION_REQUEST)).block()
val differentIdentityWithSameId: Identity = customIdentity.copy(name = IdentityName("different name"))
- when(identityFactory.listIdentities(BOB)).thenReturn(List(differentIdentityWithSameId))
+ when(identityFactory.listIdentities(BOB)).thenReturn(SFlux.just(differentIdentityWithSameId))
assertThat(SFlux.fromPublisher(testee.list(BOB)).collectSeq().block().asJava)
.containsExactlyInAnyOrder(customIdentity.copy(mayDelete = MayDeleteIdentity(false)))
@@ -118,7 +118,7 @@
@Test
def listShouldNotReturnTheIdentityHasAliasNotExists(): Unit = {
SMono(customIdentityDAO.save(BOB, CREATION_REQUEST)).block()
- when(identityFactory.listIdentities(BOB)).thenReturn(List())
+ when(identityFactory.listIdentities(BOB)).thenReturn(SFlux.empty)
assertThat(SFlux(testee.list(BOB)).collectSeq().block().asJava)
.isEmpty()
}
@@ -127,7 +127,7 @@
def listShouldWorkWhenMixCases(): Unit = {
val customIdentity: Identity = SMono(customIdentityDAO.save(BOB, CREATION_REQUEST)).block()
val customIdentityHasEmailNotExist: Identity = SMono(customIdentityDAO.save(BOB, CREATION_REQUEST.copy(email = new MailAddress("bob2@domain.tld")))).block()
- when(identityFactory.listIdentities(BOB)).thenReturn(List(IDENTITY1))
+ when(identityFactory.listIdentities(BOB)).thenReturn(SFlux.just(IDENTITY1))
assertThat(SFlux(testee.list(BOB)).collectSeq().block().asJava)
.containsExactlyInAnyOrder(customIdentity, IDENTITY1)
@@ -136,7 +136,7 @@
@Test
def updateShouldFailWhenIdNotFoundInBothCustomAndServerSetDAO(): Unit = {
- when(identityFactory.listIdentities(BOB)).thenReturn(List())
+ when(identityFactory.listIdentities(BOB)).thenReturn(SFlux.empty)
assertThatThrownBy(() => SMono.fromPublisher(testee.update(BOB, IdentityId.generate, UPDATE_REQUEST)).block())
.isInstanceOf(classOf[IdentityNotFoundException])
@@ -144,7 +144,7 @@
@Test
def updateShouldSuccessWhenCustomNotFoundAndServerSetExists(): Unit = {
- when(identityFactory.listIdentities(BOB)).thenReturn(List(IDENTITY1))
+ when(identityFactory.listIdentities(BOB)).thenReturn(SFlux.just(IDENTITY1))
assertThatCode(() => SMono.fromPublisher(testee.update(BOB, IDENTITY1.id, UPDATE_REQUEST)).block())
.doesNotThrowAnyException()
@@ -152,7 +152,7 @@
@Test
def updateShouldSuccessWhenCustomExistsAndAliasExists(): Unit = {
- when(identityFactory.listIdentities(BOB)).thenReturn(List(IDENTITY1))
+ when(identityFactory.listIdentities(BOB)).thenReturn(SFlux.just(IDENTITY1))
when(identityFactory.userCanSendFrom(BOB, BOB.asMailAddress())).thenReturn(SMono.just(true))
val customIdentity: Identity = SMono(customIdentityDAO.save(BOB, CREATION_REQUEST)).block()
@@ -162,7 +162,7 @@
@Test
def updateShouldModifiedEntry(): Unit = {
- when(identityFactory.listIdentities(BOB)).thenReturn(List(IDENTITY1))
+ when(identityFactory.listIdentities(BOB)).thenReturn(SFlux.just(IDENTITY1))
when(identityFactory.userCanSendFrom(BOB, BOB.asMailAddress())).thenReturn(SMono.just(true))
val customIdentity: Identity = SMono(customIdentityDAO.save(BOB, CREATION_REQUEST)).block()
@@ -176,7 +176,7 @@
@Test
def updateShouldSuccessWhenMultiUpdateServerSetId(): Unit = {
- when(identityFactory.listIdentities(BOB)).thenReturn(List(IDENTITY1))
+ when(identityFactory.listIdentities(BOB)).thenReturn(SFlux.just(IDENTITY1))
when(identityFactory.userCanSendFrom(BOB, BOB.asMailAddress())).thenReturn(SMono.just(true))
SMono(testee.update(BOB, IDENTITY1.id, UPDATE_REQUEST)).block()
@@ -196,7 +196,7 @@
@Test
def updateShouldSuccessWhenSecondPartialUpdateServerSetId(): Unit = {
- when(identityFactory.listIdentities(BOB)).thenReturn(List(IDENTITY1))
+ when(identityFactory.listIdentities(BOB)).thenReturn(SFlux.just(IDENTITY1))
when(identityFactory.userCanSendFrom(BOB, BOB.asMailAddress())).thenReturn(SMono.just(true))
SMono(testee.update(BOB, IDENTITY1.id, UPDATE_REQUEST)).block()
@@ -223,7 +223,7 @@
val identity: Identity = SMono.fromPublisher(testee.save(BOB, CREATION_REQUEST)).block()
when(identityFactory.userCanSendFrom(BOB, BOB.asMailAddress())).thenReturn(SMono.just(false))
- when(identityFactory.listIdentities(BOB)).thenReturn(List())
+ when(identityFactory.listIdentities(BOB)).thenReturn(SFlux.empty)
assertThatThrownBy(() => SMono.fromPublisher(testee.update(BOB, identity.id, UPDATE_REQUEST)).block())
.isInstanceOf(classOf[IdentityNotFoundException])
diff --git a/server/data/data-library/src/main/java/org/apache/james/rrt/lib/AliasReverseResolverImpl.java b/server/data/data-library/src/main/java/org/apache/james/rrt/lib/AliasReverseResolverImpl.java
index 273ac3e..6829097 100644
--- a/server/data/data-library/src/main/java/org/apache/james/rrt/lib/AliasReverseResolverImpl.java
+++ b/server/data/data-library/src/main/java/org/apache/james/rrt/lib/AliasReverseResolverImpl.java
@@ -19,25 +19,25 @@
package org.apache.james.rrt.lib;
-import java.util.Arrays;
-import java.util.HashMap;
import java.util.List;
-import java.util.Optional;
-import java.util.function.Function;
-import java.util.stream.Stream;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
import jakarta.inject.Inject;
+import org.apache.commons.lang3.tuple.Pair;
import org.apache.james.core.Domain;
import org.apache.james.core.MailAddress;
import org.apache.james.core.Username;
import org.apache.james.rrt.api.AliasReverseResolver;
import org.apache.james.rrt.api.RecipientRewriteTable;
-import org.apache.james.rrt.api.RecipientRewriteTableException;
-import org.apache.james.util.StreamUtils;
+import org.apache.james.util.ReactorUtils;
import com.github.fge.lambdas.Throwing;
-import com.google.common.collect.ImmutableList;
+import com.google.common.base.Preconditions;
+
+import reactor.core.publisher.Flux;
public class AliasReverseResolverImpl implements AliasReverseResolver {
private final RecipientRewriteTable recipientRewriteTable;
@@ -48,58 +48,63 @@
}
@Override
- public Stream<MailAddress> listAddresses(Username user) throws RecipientRewriteTable.ErrorMappingException, RecipientRewriteTableException {
- CanSendFromImpl.DomainFetcher domains = domainFetcher(user);
+ public Flux<MailAddress> listAddresses(Username user) {
+ CanSendFromImpl.DomainFetcher domains = domainFetcher();
return relatedAliases(user)
- .flatMap(allowedUser -> domains.fetch(allowedUser)
- .stream()
- .map(Optional::of)
- .map(allowedUser::withOtherDomain)
- .map(Throwing.function(Username::asMailAddress).sneakyThrow()))
+ .flatMap(allowedUser -> user.getDomainPart()
+ .map(domain -> Flux.concat(
+ Flux.just(allowedUser),
+ domains.fetch(domain).map(allowedUser::withOtherDomain)))
+ .orElseGet(() -> Flux.just(allowedUser)))
+ .map(Throwing.function(Username::asMailAddress).sneakyThrow())
.distinct();
}
- private Stream<Username> relatedAliases(Username user) {
- return StreamUtils.iterate(
- user,
- getMappingLimit(),
- Throwing.<Username, Stream<Username>>function(targetUser ->
- recipientRewriteTable
- .listSources(Mapping.alias(targetUser.asString()))
+ private Flux<Username> relatedAliases(Username user) {
+ Pair<Username, Integer> userWithDepth = Pair.of(user, 0);
+ return Flux.just(userWithDepth)
+ .expand(value -> {
+ if (value.getRight() >= getMappingLimit()) {
+ return Flux.empty();
+ }
+ return recipientRewriteTable.listSourcesReactive(Mapping.alias(value.getLeft().asString()))
.map(MappingSource::asUsername)
- .flatMap(Optional::stream)).sneakyThrow()
- );
+ .handle(ReactorUtils.publishIfPresent())
+ .map(u -> Pair.of(u, value.getRight() + 1));
+ }).map(Pair::getLeft);
}
- private CanSendFromImpl.DomainFetcher domainFetcher(Username user) {
- HashMap<Domain, List<Domain>> fetchedDomains = new HashMap<>();
- List<Domain> userDomains = relatedDomains(user).collect(ImmutableList.toImmutableList());
- user.getDomainPart().ifPresent(domain -> fetchedDomains.put(domain, userDomains));
- Function<Domain, List<Domain>> computeDomain = givenDomain -> Stream.concat(userDomains.stream(), fetchDomains(givenDomain)).collect(ImmutableList.toImmutableList());
- return givenUsername ->
- givenUsername
- .getDomainPart()
- .map(domain -> fetchedDomains.computeIfAbsent(domain, computeDomain))
- .orElseGet(Arrays::asList);
+ private CanSendFromImpl.DomainFetcher domainFetcher() {
+ return new CanSendFromImpl.DomainFetcher() {
+ private final Map<Domain, List<Domain>> memoized = new ConcurrentHashMap<>();
+ @Override
+ public Flux<Domain> fetch(Domain domain) {
+ if (memoized.containsKey(domain)) {
+ return Flux.fromIterable(memoized.get(domain));
+ }
+ return fetchDomains(domain)
+ .collect(Collectors.toList())
+ .doOnNext(next -> memoized.put(domain, next))
+ .flatMapIterable(i -> i);
+ }
+ };
}
- private Stream<Domain> relatedDomains(Username user) {
- return user.getDomainPart()
- .map(this::fetchDomains)
- .orElseGet(Stream::empty);
+ private Flux<Domain> fetchDomains(Domain domain) {
+ Pair<Domain, Integer> domainWithDepth = Pair.of(domain, 0);
+ Flux<Pair<Domain, Integer>> flux = Flux.just(domainWithDepth);
+ return expandDomains(flux);
}
- private Stream<Domain> fetchDomains(Domain domain) {
- return StreamUtils.iterate(
- domain,
- getMappingLimit(),
- Throwing.<Domain, Stream<Domain>>function(targetDomain ->
- recipientRewriteTable
- .listSources(Mapping.domainAlias(targetDomain))
+ private Flux<Domain> expandDomains(Flux<Pair<Domain, Integer>> flux) {
+ return flux.expand(value -> {
+ Preconditions.checkArgument(value.getRight() < getMappingLimit());
+ return recipientRewriteTable.listSourcesReactive(Mapping.domainAlias(value.getKey()))
.map(MappingSource::asDomain)
- .flatMap(Optional::stream)).sneakyThrow()
- );
+ .handle(ReactorUtils.publishIfPresent())
+ .map(u -> Pair.of(u, value.getRight() + 1));
+ }).map(Pair::getLeft);
}
private long getMappingLimit() {
diff --git a/server/data/data-library/src/main/java/org/apache/james/rrt/lib/CanSendFromImpl.java b/server/data/data-library/src/main/java/org/apache/james/rrt/lib/CanSendFromImpl.java
index 4407bb4..5e29e04 100644
--- a/server/data/data-library/src/main/java/org/apache/james/rrt/lib/CanSendFromImpl.java
+++ b/server/data/data-library/src/main/java/org/apache/james/rrt/lib/CanSendFromImpl.java
@@ -22,9 +22,7 @@
import static org.apache.james.rrt.lib.Mapping.Type.DomainAlias;
import java.util.EnumSet;
-import java.util.List;
import java.util.Optional;
-import java.util.stream.Stream;
import jakarta.inject.Inject;
@@ -40,13 +38,14 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
public class CanSendFromImpl implements CanSendFrom {
@FunctionalInterface
interface DomainFetcher {
- List<Domain> fetch(Username user);
+ Flux<Domain> fetch(Domain user);
}
private static final Logger LOGGER = LoggerFactory.getLogger(CanSendFromImpl.class);
@@ -80,8 +79,8 @@
}
@Override
- public Stream<MailAddress> allValidFromAddressesForUser(Username user) throws RecipientRewriteTable.ErrorMappingException, RecipientRewriteTableException {
- return aliasReverseResolver.listAddresses(user);
+ public Flux<MailAddress> allValidFromAddressesForUser(Username user) throws RecipientRewriteTable.ErrorMappingException, RecipientRewriteTableException {
+ return Flux.from(aliasReverseResolver.listAddresses(user));
}
private boolean emailIsAnAliasOfTheConnectedUser(Username connectedUser, Username fromUser) throws RecipientRewriteTable.ErrorMappingException, RecipientRewriteTableException {
diff --git a/server/protocols/webadmin/webadmin-data/src/main/java/org/apache/james/webadmin/routes/UserRoutes.java b/server/protocols/webadmin/webadmin-data/src/main/java/org/apache/james/webadmin/routes/UserRoutes.java
index a06efcd..500e541 100644
--- a/server/protocols/webadmin/webadmin-data/src/main/java/org/apache/james/webadmin/routes/UserRoutes.java
+++ b/server/protocols/webadmin/webadmin-data/src/main/java/org/apache/james/webadmin/routes/UserRoutes.java
@@ -374,10 +374,10 @@
.haltError();
}
- return canSendFrom
- .allValidFromAddressesForUser(username)
+ return Flux.from(canSendFrom.allValidFromAddressesForUser(username))
.map(MailAddress::asString)
- .collect(ImmutableList.toImmutableList());
+ .collect(ImmutableList.toImmutableList())
+ .block();
} catch (RecipientRewriteTable.ErrorMappingException | RecipientRewriteTableException | UsersRepositoryException e) {
String errorMessage = String.format("Error while listing allowed From headers for user '%s'", username);
LOGGER.info(errorMessage, e);
diff --git a/server/protocols/webadmin/webadmin-jmap/src/test/java/org/apache/james/webadmin/data/jmap/UserIdentitiesRoutesTest.java b/server/protocols/webadmin/webadmin-jmap/src/test/java/org/apache/james/webadmin/data/jmap/UserIdentitiesRoutesTest.java
index e9a7a04..d07729d 100644
--- a/server/protocols/webadmin/webadmin-jmap/src/test/java/org/apache/james/webadmin/data/jmap/UserIdentitiesRoutesTest.java
+++ b/server/protocols/webadmin/webadmin-jmap/src/test/java/org/apache/james/webadmin/data/jmap/UserIdentitiesRoutesTest.java
@@ -57,7 +57,6 @@
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scala.publisher.SMono;
-import scala.jdk.javaapi.CollectionConverters;
class UserIdentitiesRoutesTest {
@@ -95,7 +94,7 @@
void listIdentitiesShouldReturnBothCustomAndServerSetIdentities() throws Exception {
// identity: server set
Mockito.when(identityFactory.listIdentities(BOB))
- .thenReturn(CollectionConverters.asScala(List.of(IdentityRepositoryTest.IDENTITY1())).toList());
+ .thenReturn(Flux.just(IdentityRepositoryTest.IDENTITY1()));
IdentityCreationRequest creationRequest = IdentityCreationRequest.fromJava(
BOB.asMailAddress(),
@@ -181,7 +180,7 @@
void listIdentitiesShouldSupportDefaultParam() throws Exception {
// identity: server set
Mockito.when(identityFactory.listIdentities(BOB))
- .thenReturn(CollectionConverters.asScala(List.of(IdentityRepositoryTest.IDENTITY1())).toList());
+ .thenReturn(Flux.just(IdentityRepositoryTest.IDENTITY1()));
Integer highPriorityOrder = 1;
Integer lowPriorityOrder = 2;
@@ -248,7 +247,7 @@
@Test
void listIdentitiesShouldReturnBadRequestWhenInvalidDefaultParam() {
Mockito.when(identityFactory.listIdentities(BOB))
- .thenReturn(CollectionConverters.asScala(List.of(IdentityRepositoryTest.IDENTITY1())).toList());
+ .thenReturn(Flux.just(IdentityRepositoryTest.IDENTITY1()));
String response = given()
.queryParam("default", "invalid")
@@ -272,7 +271,7 @@
@Test
void listIdentitiesShouldReturnNotFoundWhenCanNotQueryDefaultIdentity() {
Mockito.when(identityFactory.listIdentities(BOB))
- .thenReturn(CollectionConverters.asScala(List.<Identity>of()).toList());
+ .thenReturn(Flux.empty());
String response = given()
.queryParam("default", "true")
@@ -318,7 +317,7 @@
"";
Mockito.when(identityFactory.listIdentities(BOB))
- .thenReturn(CollectionConverters.asScala(List.of(IdentityRepositoryTest.IDENTITY1())).toList());
+ .thenReturn(Flux.just(IdentityRepositoryTest.IDENTITY1()));
given()
.body(creationRequest)
@@ -373,7 +372,7 @@
"";
Mockito.when(identityFactory.listIdentities(BOB))
- .thenReturn(CollectionConverters.asScala(List.of(IdentityRepositoryTest.IDENTITY1())).toList());
+ .thenReturn(Flux.just(IdentityRepositoryTest.IDENTITY1()));
given()
.body(creationRequest)
@@ -416,7 +415,7 @@
" }";
Mockito.when(identityFactory.listIdentities(BOB))
- .thenReturn(CollectionConverters.asScala(List.of(IdentityRepositoryTest.IDENTITY1())).toList());
+ .thenReturn(Flux.just(IdentityRepositoryTest.IDENTITY1()));
String response = given()
.body(creationRequest)
@@ -438,7 +437,7 @@
@Test
void createIdentityShouldFailWhenInvalidRequest() {
Mockito.when(identityFactory.listIdentities(BOB))
- .thenReturn(CollectionConverters.asScala(List.of(IdentityRepositoryTest.IDENTITY1())).toList());
+ .thenReturn(Flux.just(IdentityRepositoryTest.IDENTITY1()));
String response = given()
.body("invalid")
@@ -468,7 +467,7 @@
"";
Mockito.when(identityFactory.listIdentities(BOB))
- .thenReturn(CollectionConverters.asScala(List.of(IdentityRepositoryTest.IDENTITY1())).toList());
+ .thenReturn(Flux.just(IdentityRepositoryTest.IDENTITY1()));
given()
.body(creationRequest)
@@ -503,7 +502,7 @@
@Test
void updateIdentityShouldWork() throws Exception {
Mockito.when(identityFactory.listIdentities(BOB))
- .thenReturn(CollectionConverters.asScala(List.of(IdentityRepositoryTest.IDENTITY1())).toList());
+ .thenReturn(Flux.just(IdentityRepositoryTest.IDENTITY1()));
IdentityCreationRequest creationRequest = IdentityCreationRequest.fromJava(
BOB.asMailAddress(),
@@ -582,7 +581,7 @@
@Test
void updateIdentityShouldFailWhenIdNotFound() {
Mockito.when(identityFactory.listIdentities(BOB))
- .thenReturn(CollectionConverters.asScala(List.of(IdentityRepositoryTest.IDENTITY1())).toList());
+ .thenReturn(Flux.just(IdentityRepositoryTest.IDENTITY1()));
String updateRequest = "" +
" {" +
@@ -628,7 +627,7 @@
@Test
void updateIdentityShouldNotModifyAbsentPropertyInRequest() throws Exception {
Mockito.when(identityFactory.listIdentities(BOB))
- .thenReturn(CollectionConverters.asScala(List.of(IdentityRepositoryTest.IDENTITY1())).toList());
+ .thenReturn(Flux.just(IdentityRepositoryTest.IDENTITY1()));
IdentityCreationRequest creationRequest = IdentityCreationRequest.fromJava(
BOB.asMailAddress(),
@@ -695,7 +694,7 @@
@Test
void updateIdentityShouldNotAcceptChangeMayDeleteProperty() throws Exception {
Mockito.when(identityFactory.listIdentities(BOB))
- .thenReturn(CollectionConverters.asScala(List.of(IdentityRepositoryTest.IDENTITY1())).toList());
+ .thenReturn(Flux.just(IdentityRepositoryTest.IDENTITY1()));
IdentityCreationRequest creationRequest = IdentityCreationRequest.fromJava(
BOB.asMailAddress(),
