| /**************************************************************** |
| * Licensed to the Apache Software Foundation (ASF) under one * |
| * or more contributor license agreements. See the NOTICE file * |
| * distributed with this work for additional information * |
| * regarding copyright ownership. The ASF licenses this file * |
| * to you under the Apache License, Version 2.0 (the * |
| * "License"); you may not use this file except in compliance * |
| * with the License. You may obtain a copy of the License at * |
| * * |
| * http://www.apache.org/licenses/LICENSE-2.0 * |
| * * |
| * Unless required by applicable law or agreed to in writing, * |
| * software distributed under the License is distributed on an * |
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY * |
| * KIND, either express or implied. See the License for the * |
| * specific language governing permissions and limitations * |
| * under the License. * |
| ****************************************************************/ |
| |
| package org.apache.james.mailbox.store; |
| |
| import static org.apache.james.mailbox.events.MailboxEvents.Added.IS_APPENDED; |
| import static org.apache.james.mailbox.events.MailboxEvents.Added.IS_DELIVERY; |
| import static org.apache.james.util.ReactorUtils.DEFAULT_CONCURRENCY; |
| |
| import java.util.Collection; |
| import java.util.HashMap; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Optional; |
| import java.util.Set; |
| import java.util.function.Function; |
| import java.util.function.Predicate; |
| import java.util.stream.Stream; |
| |
| import javax.inject.Inject; |
| import javax.mail.Flags; |
| |
| import org.apache.commons.lang3.tuple.Pair; |
| import org.apache.james.events.EventBus; |
| import org.apache.james.mailbox.MailboxSession; |
| import org.apache.james.mailbox.MessageIdManager; |
| import org.apache.james.mailbox.MessageManager; |
| import org.apache.james.mailbox.MetadataWithMailboxId; |
| import org.apache.james.mailbox.RightManager; |
| import org.apache.james.mailbox.events.MailboxIdRegistrationKey; |
| import org.apache.james.mailbox.exception.MailboxException; |
| import org.apache.james.mailbox.exception.MailboxNotFoundException; |
| import org.apache.james.mailbox.exception.OverQuotaException; |
| import org.apache.james.mailbox.extension.PreDeletionHook; |
| import org.apache.james.mailbox.model.ComposedMessageIdWithMetaData; |
| import org.apache.james.mailbox.model.DeleteResult; |
| import org.apache.james.mailbox.model.FetchGroup; |
| import org.apache.james.mailbox.model.Mailbox; |
| import org.apache.james.mailbox.model.MailboxACL; |
| import org.apache.james.mailbox.model.MailboxACL.Right; |
| import org.apache.james.mailbox.model.MailboxId; |
| import org.apache.james.mailbox.model.MessageId; |
| import org.apache.james.mailbox.model.MessageMetaData; |
| import org.apache.james.mailbox.model.MessageMoves; |
| import org.apache.james.mailbox.model.MessageResult; |
| import org.apache.james.mailbox.model.QuotaRoot; |
| import org.apache.james.mailbox.model.UpdatedFlags; |
| import org.apache.james.mailbox.quota.QuotaManager; |
| import org.apache.james.mailbox.quota.QuotaRootResolver; |
| import org.apache.james.mailbox.store.event.EventFactory; |
| import org.apache.james.mailbox.store.mail.FetchGroupConverter; |
| import org.apache.james.mailbox.store.mail.MailboxMapper; |
| import org.apache.james.mailbox.store.mail.MessageIdMapper; |
| import org.apache.james.mailbox.store.mail.MessageMapper; |
| import org.apache.james.mailbox.store.mail.model.FlagsFactory; |
| import org.apache.james.mailbox.store.mail.model.FlagsFilter; |
| import org.apache.james.mailbox.store.mail.model.MailboxMessage; |
| import org.apache.james.mailbox.store.mail.model.Message; |
| import org.apache.james.mailbox.store.quota.QuotaChecker; |
| import org.apache.james.util.FunctionalUtils; |
| import org.reactivestreams.Publisher; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import com.github.fge.lambdas.Throwing; |
| import com.github.fge.lambdas.functions.ThrowingFunction; |
| import com.google.common.collect.ImmutableList; |
| import com.google.common.collect.ImmutableListMultimap; |
| import com.google.common.collect.ImmutableMap; |
| import com.google.common.collect.ImmutableSet; |
| import com.google.common.collect.Sets; |
| |
| import reactor.core.publisher.Flux; |
| import reactor.core.publisher.Mono; |
| import reactor.core.scheduler.Scheduler; |
| import reactor.core.scheduler.Schedulers; |
| |
| public class StoreMessageIdManager implements MessageIdManager { |
| |
| public static ImmutableSet<MailboxId> toMailboxIds(List<MailboxMessage> mailboxMessages) { |
| return mailboxMessages |
| .stream() |
| .map(MailboxMessage::getMailboxId) |
| .collect(ImmutableSet.toImmutableSet()); |
| } |
| |
| private static final Logger LOGGER = LoggerFactory.getLogger(StoreMessageIdManager.class); |
| |
| private final RightManager rightManager; |
| private final MailboxSessionMapperFactory mailboxSessionMapperFactory; |
| private final EventBus eventBus; |
| private final QuotaManager quotaManager; |
| private final QuotaRootResolver quotaRootResolver; |
| private final PreDeletionHooks preDeletionHooks; |
| |
| @Inject |
| public StoreMessageIdManager(RightManager rightManager, MailboxSessionMapperFactory mailboxSessionMapperFactory, |
| EventBus eventBus, QuotaManager quotaManager, QuotaRootResolver quotaRootResolver, |
| PreDeletionHooks preDeletionHooks) { |
| this.rightManager = rightManager; |
| this.mailboxSessionMapperFactory = mailboxSessionMapperFactory; |
| this.eventBus = eventBus; |
| this.quotaManager = quotaManager; |
| this.quotaRootResolver = quotaRootResolver; |
| this.preDeletionHooks = preDeletionHooks; |
| } |
| |
| @Override |
| public void setFlags(Flags newState, MessageManager.FlagsUpdateMode replace, MessageId messageId, List<MailboxId> mailboxIds, MailboxSession mailboxSession) throws MailboxException { |
| MailboxReactorUtils.block(setFlagsReactive(newState, replace, messageId, mailboxIds, mailboxSession)); |
| } |
| |
| @Override |
| public Mono<Void> setFlagsReactive(Flags newState, MessageManager.FlagsUpdateMode replace, MessageId messageId, List<MailboxId> mailboxIds, MailboxSession mailboxSession) { |
| MessageIdMapper messageIdMapper = mailboxSessionMapperFactory.getMessageIdMapper(mailboxSession); |
| MailboxMapper mailboxMapper = mailboxSessionMapperFactory.getMailboxMapper(mailboxSession); |
| |
| int concurrency = 4; |
| |
| return Flux.fromIterable(mailboxIds) |
| .flatMap(mailboxMapper::findMailboxById, concurrency) |
| .collect(ImmutableList.toImmutableList()) |
| .flatMap(Throwing.<List<Mailbox>, Mono<Void>>function(targetMailboxes -> { |
| assertRightsOnMailboxes(targetMailboxes, mailboxSession, Right.Write); |
| |
| return messageIdMapper.setFlags(messageId, mailboxIds, newState, replace) |
| .flatMapIterable(updatedFlags -> updatedFlags.asMap().entrySet()) |
| .concatMap(entry -> dispatchFlagsChange(mailboxSession, entry.getKey(), ImmutableList.copyOf(entry.getValue()), targetMailboxes)) |
| .then(); |
| }).sneakyThrow()); |
| } |
| |
| @Override |
| public Set<MessageId> accessibleMessages(Collection<MessageId> messageIds, MailboxSession mailboxSession) { |
| return accessibleMessagesReactive(messageIds, mailboxSession).block(); |
| } |
| |
| @Override |
| public Mono<Set<MessageId>> accessibleMessagesReactive(Collection<MessageId> messageIds, MailboxSession mailboxSession) { |
| MessageIdMapper messageIdMapper = mailboxSessionMapperFactory.getMessageIdMapper(mailboxSession); |
| return Flux.fromIterable(messageIds) |
| .flatMap(messageIdMapper::findMetadata, DEFAULT_CONCURRENCY) |
| .collect(ImmutableList.toImmutableList()) |
| .flatMap(idList -> getAllowedMailboxIds(mailboxSession, idList.stream() |
| .map(id -> id.getComposedMessageId().getMailboxId()), Right.Read) |
| .map(allowedMailboxIds -> idList.stream() |
| .filter(id -> allowedMailboxIds.contains(id.getComposedMessageId().getMailboxId())) |
| .map(id -> id.getComposedMessageId().getMessageId()) |
| .collect(ImmutableSet.toImmutableSet()))); |
| } |
| |
| @Override |
| public List<MessageResult> getMessages(Collection<MessageId> messageIds, FetchGroup fetchGroup, MailboxSession mailboxSession) { |
| return getMessagesReactive(messageIds, fetchGroup, mailboxSession) |
| .collectList() |
| .block(); |
| } |
| |
| @Override |
| public Flux<MessageResult> getMessagesReactive(Collection<MessageId> messageIds, FetchGroup fetchGroup, MailboxSession mailboxSession) { |
| MessageIdMapper messageIdMapper = mailboxSessionMapperFactory.getMessageIdMapper(mailboxSession); |
| |
| MessageMapper.FetchType fetchType = FetchGroupConverter.getFetchType(fetchGroup); |
| return messageIdMapper.findReactive(messageIds, fetchType) |
| .groupBy(MailboxMessage::getMailboxId) |
| .filterWhen(groupedFlux -> hasRightsOnMailboxReactive(mailboxSession, Right.Read).apply(groupedFlux.key()), DEFAULT_CONCURRENCY) |
| .flatMap(Function.identity(), DEFAULT_CONCURRENCY) |
| .publishOn(forFetchType(fetchType)) |
| .map(Throwing.function(messageResultConverter(fetchGroup)).sneakyThrow()); |
| } |
| |
| private Scheduler forFetchType(MessageMapper.FetchType fetchType) { |
| if (fetchType == MessageMapper.FetchType.FULL) { |
| return Schedulers.parallel(); |
| } |
| return Schedulers.immediate(); |
| } |
| |
| @Override |
| public Publisher<ComposedMessageIdWithMetaData> messagesMetadata(Collection<MessageId> ids, MailboxSession session) { |
| MessageIdMapper messageIdMapper = mailboxSessionMapperFactory.getMessageIdMapper(session); |
| int concurrency = 4; |
| return Flux.fromIterable(ids) |
| .flatMap(messageIdMapper::findMetadata, concurrency) |
| .groupBy(metaData -> metaData.getComposedMessageId().getMailboxId()) |
| .filterWhen(groupedFlux -> hasRightsOnMailboxReactive(session, Right.Read).apply(groupedFlux.key()), DEFAULT_CONCURRENCY) |
| .flatMap(Function.identity(), DEFAULT_CONCURRENCY); |
| } |
| |
| private Mono<ImmutableSet<MailboxId>> getAllowedMailboxIds(MailboxSession mailboxSession, Stream<MailboxId> idList, Right... rights) { |
| return Flux.fromStream(idList) |
| .distinct() |
| .filterWhen(hasRightsOnMailboxReactive(mailboxSession, rights), DEFAULT_CONCURRENCY) |
| .collect(ImmutableSet.toImmutableSet()); |
| } |
| |
| @Override |
| public DeleteResult delete(MessageId messageId, List<MailboxId> mailboxIds, MailboxSession mailboxSession) throws MailboxException { |
| return MailboxReactorUtils.block(deleteReactive(ImmutableList.of(messageId), mailboxIds, mailboxSession)); |
| } |
| |
| @Override |
| public Mono<DeleteResult> deleteReactive(List<MessageId> messageIds, List<MailboxId> mailboxIds, MailboxSession mailboxSession) { |
| MessageIdMapper messageIdMapper = mailboxSessionMapperFactory.getMessageIdMapper(mailboxSession); |
| |
| return assertRightsOnMailboxIds(mailboxIds, mailboxSession, Right.DeleteMessages) |
| .then(messageIdMapper.findReactive(messageIds, MessageMapper.FetchType.METADATA) |
| .filter(inMailboxes(mailboxIds)) |
| .collectList()) |
| .flatMap(messageList -> { |
| Set<MessageId> found = messageList.stream() |
| .map(Message::getMessageId) |
| .collect(ImmutableSet.toImmutableSet()); |
| Set<MessageId> notFound = Sets.difference(ImmutableSet.copyOf(messageIds), found); |
| |
| DeleteResult result = DeleteResult.builder() |
| .addDestroyed(found) |
| .addNotFound(notFound) |
| .build(); |
| |
| |
| if (!messageList.isEmpty()) { |
| return deleteWithPreHooks(messageIdMapper, messageList, mailboxSession) |
| .thenReturn(result); |
| } |
| return Mono.just(result); |
| }); |
| } |
| |
| @Override |
| public Mono<DeleteResult> delete(List<MessageId> messageIds, MailboxSession mailboxSession) { |
| MessageIdMapper messageIdMapper = mailboxSessionMapperFactory.getMessageIdMapper(mailboxSession); |
| |
| return messageIdMapper.findReactive(messageIds, MessageMapper.FetchType.METADATA) |
| .collectList() |
| .flatMap(messageList -> |
| getAllowedMailboxIds(mailboxSession, |
| messageList |
| .stream() |
| .map(MailboxMessage::getMailboxId), |
| Right.DeleteMessages) |
| .flatMap(allowedMailboxIds -> deleteInAllowedMailboxes(messageIds, mailboxSession, messageIdMapper, messageList, allowedMailboxIds))); |
| } |
| |
| private Mono<DeleteResult> deleteInAllowedMailboxes(List<MessageId> messageIds, MailboxSession mailboxSession, MessageIdMapper messageIdMapper, List<MailboxMessage> messageList, ImmutableSet<MailboxId> allowedMailboxIds) { |
| List<MailboxMessage> accessibleMessages = messageList.stream() |
| .filter(message -> allowedMailboxIds.contains(message.getMailboxId())) |
| .collect(ImmutableList.toImmutableList()); |
| ImmutableSet<MessageId> accessibleMessageIds = accessibleMessages.stream() |
| .map(MailboxMessage::getMessageId) |
| .distinct() |
| .collect(ImmutableSet.toImmutableSet()); |
| Sets.SetView<MessageId> nonAccessibleMessages = Sets.difference(ImmutableSet.copyOf(messageIds), accessibleMessageIds); |
| |
| return deleteWithPreHooks(messageIdMapper, accessibleMessages, mailboxSession) |
| .thenReturn(DeleteResult.builder() |
| .addDestroyed(accessibleMessageIds) |
| .addNotFound(nonAccessibleMessages) |
| .build()); |
| } |
| |
| private Mono<Void> deleteWithPreHooks(MessageIdMapper messageIdMapper, List<MailboxMessage> messageList, MailboxSession mailboxSession) { |
| ImmutableMap<MetadataWithMailboxId, MessageMetaData> metadataWithMailbox = messageList.stream() |
| .collect(ImmutableMap.toImmutableMap( |
| message -> MetadataWithMailboxId.from(message.metaData(), message.getMailboxId()), |
| MailboxMessage::metaData)); |
| |
| return preDeletionHooks.runHooks(PreDeletionHook.DeleteOperation.from(metadataWithMailbox.keySet().asList())) |
| .then(delete(messageIdMapper, messageList, mailboxSession, metadataWithMailbox)); |
| } |
| |
| private Mono<Void> delete(MessageIdMapper messageIdMapper, List<MailboxMessage> messageList, MailboxSession mailboxSession, Map<MetadataWithMailboxId, MessageMetaData> metadataWithMailbox) { |
| MailboxMapper mailboxMapper = mailboxSessionMapperFactory.getMailboxMapper(mailboxSession); |
| |
| return messageIdMapper.deleteReactive( |
| messageList.stream() |
| .collect(ImmutableListMultimap.toImmutableListMultimap( |
| Message::getMessageId, |
| MailboxMessage::getMailboxId))) |
| .then( |
| Flux.fromIterable(metadataWithMailbox.entrySet()) |
| .flatMap(metadataWithMailboxId -> mailboxMapper.findMailboxById(metadataWithMailboxId.getKey().getMailboxId()) |
| .flatMap(mailbox -> eventBus.dispatch(EventFactory.expunged() |
| .randomEventId() |
| .mailboxSession(mailboxSession) |
| .mailbox(mailbox) |
| .addMetaData(metadataWithMailboxId.getValue()) |
| .build(), |
| new MailboxIdRegistrationKey(metadataWithMailboxId.getKey().getMailboxId()))), DEFAULT_CONCURRENCY) |
| .then()); |
| } |
| |
| @Override |
| public void setInMailboxes(MessageId messageId, Collection<MailboxId> targetMailboxIds, MailboxSession mailboxSession) throws MailboxException { |
| MailboxReactorUtils.block(setInMailboxesReactive(messageId, targetMailboxIds, mailboxSession)); |
| } |
| |
| @Override |
| public Mono<Void> setInMailboxesReactive(MessageId messageId, Collection<MailboxId> targetMailboxIds, MailboxSession mailboxSession) { |
| return findRelatedMailboxMessages(messageId, mailboxSession) |
| .flatMap(currentMailboxMessages -> messageMovesWithMailbox(MessageMoves.builder() |
| .targetMailboxIds(targetMailboxIds) |
| .previousMailboxIds(toMailboxIds(currentMailboxMessages)) |
| .build(), mailboxSession) |
| .flatMap(Throwing.<MessageMovesWithMailbox, Mono<Void>>function(messageMove -> { |
| MessageMovesWithMailbox refined = messageMove.filterPrevious(hasRightsOnMailbox(mailboxSession, Right.Read)); |
| |
| if (messageMove.getPreviousMailboxes().isEmpty()) { |
| LOGGER.info("Tried to access {} not accessible for {}", messageId, mailboxSession.getUser().asString()); |
| return Mono.empty(); |
| } |
| if (refined.isChange()) { |
| return applyMessageMoves(mailboxSession, currentMailboxMessages, refined); |
| } |
| return Mono.empty(); |
| }).sneakyThrow())); |
| } |
| |
| public void setInMailboxesNoCheck(MessageId messageId, MailboxId targetMailboxId, MailboxSession mailboxSession) throws MailboxException { |
| MessageIdMapper messageIdMapper = mailboxSessionMapperFactory.getMessageIdMapper(mailboxSession); |
| List<MailboxMessage> currentMailboxMessages = messageIdMapper.find(ImmutableList.of(messageId), MessageMapper.FetchType.METADATA); |
| |
| |
| MailboxReactorUtils.block(messageMovesWithMailbox(MessageMoves.builder() |
| .targetMailboxIds(targetMailboxId) |
| .previousMailboxIds(toMailboxIds(currentMailboxMessages)) |
| .build(), mailboxSession) |
| .flatMap(messageMove -> { |
| if (messageMove.isChange()) { |
| return applyMessageMoveNoMailboxChecks(mailboxSession, currentMailboxMessages, messageMove); |
| } |
| return Mono.empty(); |
| })); |
| } |
| |
| private Mono<List<MailboxMessage>> findRelatedMailboxMessages(MessageId messageId, MailboxSession mailboxSession) { |
| MessageIdMapper messageIdMapper = mailboxSessionMapperFactory.getMessageIdMapper(mailboxSession); |
| |
| return messageIdMapper.findReactive(ImmutableList.of(messageId), MessageMapper.FetchType.METADATA) |
| .collect(ImmutableList.toImmutableList()); |
| } |
| |
| private Mono<Void> applyMessageMoves(MailboxSession mailboxSession, List<MailboxMessage> currentMailboxMessages, MessageMovesWithMailbox messageMoves) throws MailboxNotFoundException { |
| assertRightsOnMailboxes(messageMoves.addedMailboxes(), mailboxSession, Right.Insert); |
| assertRightsOnMailboxes(messageMoves.removedMailboxes(), mailboxSession, Right.DeleteMessages); |
| assertRightsOnMailboxes(messageMoves.getTargetMailboxes(), mailboxSession, Right.Read); |
| |
| return applyMessageMoveNoMailboxChecks(mailboxSession, currentMailboxMessages, messageMoves); |
| } |
| |
| private Mono<Void> applyMessageMoveNoMailboxChecks(MailboxSession mailboxSession, List<MailboxMessage> currentMailboxMessages, MessageMovesWithMailbox messageMoves) { |
| Optional<MailboxMessage> mailboxMessage = currentMailboxMessages.stream().findAny(); |
| |
| if (mailboxMessage.isEmpty()) { |
| return Mono.error(new MailboxNotFoundException("can't load message")); |
| } |
| List<Pair<MailboxMessage, Mailbox>> messagesToRemove = currentMailboxMessages.stream() |
| .flatMap(message -> messageMoves.removedMailboxes() |
| .stream() |
| .filter(mailbox -> mailbox.getMailboxId().equals(message.getMailboxId())) |
| .map(mailbox -> Pair.of(message, mailbox))) |
| .collect(ImmutableList.toImmutableList()); |
| |
| return validateQuota(messageMoves, mailboxMessage.get()) |
| .then(addMessageToMailboxes(mailboxMessage.get(), messageMoves, mailboxSession)) |
| .then(expungeMessageFromMailboxes(mailboxMessage.get().getMessageId(), messagesToRemove, mailboxSession, messageMoves)) |
| .then(eventBus.dispatch(EventFactory.moved() |
| .session(mailboxSession) |
| .messageMoves(messageMoves.asMessageMoves()) |
| .messageId(mailboxMessage.get().getMessageId()) |
| .build(), |
| messageMoves.impactedMailboxes() |
| .map(Mailbox::getMailboxId) |
| .map(MailboxIdRegistrationKey::new) |
| .collect(ImmutableSet.toImmutableSet()))); |
| } |
| |
| private Mono<Void> expungeMessageFromMailboxes(MessageId messageId, List<Pair<MailboxMessage, Mailbox>> messages, MailboxSession mailboxSession, MessageMovesWithMailbox messageMoves) { |
| if (messages.isEmpty()) { |
| return Mono.empty(); |
| } |
| |
| MessageIdMapper messageIdMapper = mailboxSessionMapperFactory.getMessageIdMapper(mailboxSession); |
| |
| ImmutableList<MailboxId> mailboxIds = messages.stream() |
| .map(pair -> pair.getRight().getMailboxId()) |
| .collect(ImmutableList.toImmutableList()); |
| |
| return Mono.from(messageIdMapper.deleteReactive(messageId, mailboxIds)) |
| .then(Flux.fromIterable(messages) |
| .flatMap(message -> dispatchExpungedEvent(message, mailboxSession, messageMoves), DEFAULT_CONCURRENCY) |
| .then()); |
| } |
| |
| private Mono<Void> dispatchExpungedEvent(Pair<MailboxMessage, Mailbox> message, MailboxSession mailboxSession, MessageMovesWithMailbox messageMoves) { |
| return Mono.just(EventFactory.expunged() |
| .randomEventId() |
| .mailboxSession(mailboxSession) |
| .mailbox(message.getRight()) |
| .addMetaData(message.getLeft().metaData())) |
| .map(eventBuilder -> { |
| if (isSingleMove(messageMoves)) { |
| return eventBuilder |
| .movedTo(messageMoves.addedMailboxes().iterator().next().getMailboxId()) |
| .build(); |
| } else { |
| return eventBuilder.build(); |
| } |
| }) |
| .flatMap(event -> eventBus.dispatch(event, new MailboxIdRegistrationKey(message.getRight().getMailboxId()))); |
| } |
| |
| private Mono<Void> dispatchFlagsChange(MailboxSession mailboxSession, MailboxId mailboxId, ImmutableList<UpdatedFlags> updatedFlags, List<Mailbox> knownMailboxes) { |
| if (updatedFlags.stream().anyMatch(UpdatedFlags::flagsChanged)) { |
| return knownMailboxes.stream() |
| .filter(knownMailbox -> knownMailbox.getMailboxId().equals(mailboxId)) |
| .findFirst() |
| .map(Mono::just) |
| .orElseGet(() -> mailboxSessionMapperFactory.getMailboxMapper(mailboxSession).findMailboxById(mailboxId)) |
| .flatMap(mailbox -> |
| eventBus.dispatch(EventFactory.flagsUpdated() |
| .randomEventId() |
| .mailboxSession(mailboxSession) |
| .mailbox(mailbox) |
| .updatedFlags(updatedFlags) |
| .build(), |
| new MailboxIdRegistrationKey(mailboxId))); |
| } |
| return Mono.empty(); |
| } |
| |
| private Mono<Void> validateQuota(MessageMovesWithMailbox messageMoves, MailboxMessage mailboxMessage) { |
| Map<QuotaRoot, Integer> messageCountByQuotaRoot = buildMapQuotaRoot(messageMoves); |
| |
| return Flux.fromIterable(messageCountByQuotaRoot.entrySet()) |
| .filter(entry -> entry.getValue() > 0) |
| .flatMap(entry -> Mono.from(quotaManager.getQuotasReactive(entry.getKey())) |
| .map(quotas -> new QuotaChecker(quotas, entry.getKey())) |
| .handle((quotaChecker, sink) -> { |
| Integer additionalCopyCount = entry.getValue(); |
| long additionalOccupiedSpace = additionalCopyCount * mailboxMessage.getFullContentOctets(); |
| try { |
| quotaChecker.tryAddition(additionalCopyCount, additionalOccupiedSpace); |
| } catch (OverQuotaException e) { |
| sink.error(e); |
| } |
| })) |
| .then(); |
| } |
| |
| private Map<QuotaRoot, Integer> buildMapQuotaRoot(MessageMovesWithMailbox messageMoves) { |
| try { |
| Map<QuotaRoot, Integer> messageCountByQuotaRoot = new HashMap<>(); |
| for (Mailbox mailbox : messageMoves.addedMailboxes()) { |
| QuotaRoot quotaRoot = quotaRootResolver.getQuotaRoot(mailbox); |
| int currentCount = Optional.ofNullable(messageCountByQuotaRoot.get(quotaRoot)) |
| .orElse(0); |
| messageCountByQuotaRoot.put(quotaRoot, currentCount + 1); |
| } |
| for (Mailbox mailbox : messageMoves.removedMailboxes()) { |
| QuotaRoot quotaRoot = quotaRootResolver.getQuotaRoot(mailbox); |
| int currentCount = Optional.ofNullable(messageCountByQuotaRoot.get(quotaRoot)) |
| .orElse(0); |
| messageCountByQuotaRoot.put(quotaRoot, currentCount - 1); |
| } |
| return messageCountByQuotaRoot; |
| } catch (MailboxException e) { |
| throw new RuntimeException(e); |
| } |
| } |
| |
| private Mono<Void> addMessageToMailboxes(MailboxMessage mailboxMessage, MessageMovesWithMailbox messageMoves, MailboxSession mailboxSession) { |
| MessageIdMapper messageIdMapper = mailboxSessionMapperFactory.getMessageIdMapper(mailboxSession); |
| |
| return Flux.fromIterable(messageMoves.addedMailboxes()) |
| .flatMap(Throwing.<Mailbox, Mono<Void>>function(mailbox -> { |
| MailboxACL.Rfc4314Rights myRights = rightManager.myRights(mailbox, mailboxSession); |
| boolean shouldPreserveFlags = myRights.contains(Right.Write); |
| MailboxMessage copy = mailboxMessage.copy(mailbox); |
| copy.setFlags(FlagsFactory |
| .builder() |
| .flags(mailboxMessage.createFlags()) |
| .filteringFlags( |
| FlagsFilter.builder() |
| .systemFlagFilter(f -> shouldPreserveFlags) |
| .userFlagFilter(f -> shouldPreserveFlags) |
| .build()) |
| .build()); |
| |
| return save(messageIdMapper, copy, mailbox) |
| .flatMap(metadata -> dispatchAddedEvent(mailboxSession, mailbox, metadata, messageMoves)); |
| }).sneakyThrow()) |
| .then(); |
| } |
| |
| private Mono<Void> dispatchAddedEvent(MailboxSession mailboxSession, Mailbox mailbox, MessageMetaData messageMetaData, MessageMovesWithMailbox messageMoves) { |
| return Mono.just(EventFactory.added() |
| .randomEventId() |
| .mailboxSession(mailboxSession) |
| .mailbox(mailbox) |
| .addMetaData(messageMetaData) |
| .isDelivery(!IS_DELIVERY) |
| .isAppended(!IS_APPENDED)) |
| .map(eventBuilder -> { |
| if (isSingleMove(messageMoves)) { |
| return eventBuilder |
| .movedFrom(messageMoves.removedMailboxes().iterator().next().getMailboxId()) |
| .build(); |
| } else { |
| return eventBuilder.build(); |
| } |
| }).flatMap(event -> eventBus.dispatch(event, new MailboxIdRegistrationKey(mailbox.getMailboxId()))); |
| } |
| |
| private boolean isSingleMove(MessageMovesWithMailbox messageMoves) { |
| return messageMoves.addedMailboxes().size() == 1 && messageMoves.removedMailboxes().size() == 1; |
| } |
| |
| private Mono<MessageMetaData> save(MessageIdMapper messageIdMapper, MailboxMessage mailboxMessage, Mailbox mailbox) { |
| return Mono.zip( |
| mailboxSessionMapperFactory.getModSeqProvider().nextModSeqReactive(mailbox.getMailboxId()), |
| mailboxSessionMapperFactory.getUidProvider().nextUidReactive(mailbox.getMailboxId())) |
| .flatMap(modSeqAndUid -> { |
| mailboxMessage.setModSeq(modSeqAndUid.getT1()); |
| mailboxMessage.setUid(modSeqAndUid.getT2()); |
| |
| return messageIdMapper.copyInMailboxReactive(mailboxMessage, mailbox) |
| .thenReturn(mailboxMessage.metaData()); |
| }); |
| } |
| |
| private ThrowingFunction<MailboxMessage, MessageResult> messageResultConverter(FetchGroup fetchGroup) { |
| return input -> ResultUtils.loadMessageResult(input, fetchGroup); |
| } |
| |
| private Predicate<MailboxMessage> inMailboxes(Collection<MailboxId> mailboxIds) { |
| return mailboxMessage -> mailboxIds.contains(mailboxMessage.getMailboxId()); |
| } |
| |
| private Function<MailboxId, Mono<Boolean>> hasRightsOnMailboxReactive(MailboxSession session, Right... rights) { |
| return mailboxId -> Mono.from(rightManager.myRights(mailboxId, session)) |
| .map(myRights -> myRights.contains(rights)) |
| .onErrorResume(any -> Mono.just(false)); |
| } |
| |
| private Predicate<Mailbox> hasRightsOnMailbox(MailboxSession session, Right... rights) { |
| return mailbox -> rightManager.myRights(mailbox, session).contains(rights); |
| } |
| |
| private Mono<Void> assertRightsOnMailboxIds(Collection<MailboxId> mailboxIds, MailboxSession mailboxSession, Right... rights) { |
| return Flux.fromIterable(mailboxIds) |
| .filterWhen(hasRightsOnMailboxReactive(mailboxSession, rights).andThen(result -> result.map(FunctionalUtils.negate())), DEFAULT_CONCURRENCY) |
| .next() |
| .flatMap(mailboxForbidden -> { |
| LOGGER.info("Mailbox with Id {} does not belong to {}", mailboxForbidden, mailboxSession.getUser().asString()); |
| return Mono.error(new MailboxNotFoundException(mailboxForbidden)); |
| }) |
| .then(); |
| } |
| |
| private void assertRightsOnMailboxes(Collection<Mailbox> mailboxes, MailboxSession mailboxSession, Right... rights) throws MailboxNotFoundException { |
| Optional<Mailbox> firstForbiddenMailbox = mailboxes.stream() |
| .filter(Predicate.not(hasRightsOnMailbox(mailboxSession, rights))) |
| .findFirst(); |
| |
| if (firstForbiddenMailbox.isPresent()) { |
| MailboxId mailboxId = firstForbiddenMailbox.get().getMailboxId(); |
| LOGGER.info("Mailbox with Id {} does not belong to {}", mailboxId, mailboxSession.getUser().asString()); |
| throw new MailboxNotFoundException(firstForbiddenMailbox.get().generateAssociatedPath()); |
| } |
| } |
| |
| private Mono<MessageMovesWithMailbox> messageMovesWithMailbox(MessageMoves messageMoves, MailboxSession session) { |
| MailboxMapper mailboxMapper = mailboxSessionMapperFactory.getMailboxMapper(session); |
| |
| Mono<List<Mailbox>> target = Flux.fromIterable(messageMoves.getTargetMailboxIds()) |
| .flatMap(mailboxMapper::findMailboxById, DEFAULT_CONCURRENCY) |
| .collect(ImmutableList.toImmutableList()); |
| Mono<List<Mailbox>> previous = Flux.fromIterable(messageMoves.getPreviousMailboxIds()) |
| .flatMap(mailboxMapper::findMailboxById, DEFAULT_CONCURRENCY) |
| .collect(ImmutableList.toImmutableList()); |
| |
| return target.zipWith(previous) |
| .map(tuple -> MessageMovesWithMailbox.builder() |
| .targetMailboxes(tuple.getT1()) |
| .previousMailboxes(tuple.getT2()) |
| .build()); |
| } |
| } |
