MAILBOX-331 Implement Quota Threshold notifier as an EventSourcing project
Project: http://git-wip-us.apache.org/repos/asf/james-project/repo Commit: http://git-wip-us.apache.org/repos/asf/james-project/commit/0e437d25 Tree: http://git-wip-us.apache.org/repos/asf/james-project/tree/0e437d25 Diff: http://git-wip-us.apache.org/repos/asf/james-project/diff/0e437d25 Branch: refs/heads/master Commit: 0e437d2593d3b666de51af493283d1d792761d35 Parents: b035747 Author: benwa <[email protected]> Authored: Thu Apr 26 12:32:37 2018 +0700 Committer: benwa <[email protected]> Committed: Tue May 8 15:42:03 2018 +0700 ---------------------------------------------------------------------- .../apache/james/mailbox/MailboxListener.java | 14 +- .../james/mailbox/MailboxManagerTest.java | 14 +- mailbox/memory/pom.xml | 1 + mailbox/plugin/quota-mailing-memory/pom.xml | 96 +++++++ .../james/eventsource/InMemoryEventStore.java | 104 ++++++++ .../InMemoryEventSourcingSystemTest.java | 27 ++ .../InMemoryEventStoreExtension.java | 39 +++ .../eventsourcing/InMemoryEventStoreTest.java | 27 ++ ...oryQuotaMailingListenersIntegrationTest.java | 29 +++ mailbox/plugin/quota-mailing/pom.xml | 103 ++++++++ .../apache/james/eventsourcing/AggregateId.java | 24 ++ .../james/eventsourcing/CommandDispatcher.java | 115 +++++++++ .../org/apache/james/eventsourcing/Event.java | 32 +++ .../apache/james/eventsourcing/EventBus.java | 61 +++++ .../org/apache/james/eventsourcing/EventId.java | 78 ++++++ .../eventsourcing/EventSourcingSystem.java | 36 +++ .../apache/james/eventsourcing/EventStore.java | 114 +++++++++ .../apache/james/eventsourcing/Subscriber.java | 24 ++ .../QuotaMailingListenerConfiguration.java | 74 ++++++ .../mailing/aggregates/UserQuotaThresholds.java | 163 ++++++++++++ .../commands/DetectThresholdCrossing.java | 77 ++++++ .../DetectThresholdCrossingHandler.java | 59 +++++ .../events/QuotaThresholdChangedEvent.java | 75 ++++++ .../QuotaThresholdCrossingListener.java | 71 ++++++ .../subscribers/QuotaThresholdMailer.java | 79 ++++++ .../subscribers/QuotaThresholdNotice.java | 199 +++++++++++++++ .../mailbox/quota/model/HistoryEvolution.java | 111 ++++++++ .../mailbox/quota/model/QuotaThreshold.java | 99 ++++++++ .../quota/model/QuotaThresholdChange.java | 71 ++++++ .../quota/model/QuotaThresholdHistory.java | 109 ++++++++ .../mailbox/quota/model/QuotaThresholds.java | 58 +++++ .../eventsourcing/DataCollectorSubscriber.java | 46 ++++ .../apache/james/eventsourcing/EventIdTest.java | 85 +++++++ .../eventsourcing/EventSourcingSystemTest.java | 251 +++++++++++++++++++ .../james/eventsourcing/EventStoreTest.java | 75 ++++++ .../apache/james/eventsourcing/HistoryTest.java | 87 +++++++ .../james/eventsourcing/TestAggregateId.java | 64 +++++ .../apache/james/eventsourcing/TestEvent.java | 82 ++++++ .../aggregates/UserQuotaThresholdsTest.java | 47 ++++ .../commands/DetectThresholdCrossingTest.java | 33 +++ .../QuotaThresholdListenersTestSystem.java | 58 +++++ .../QuotaThresholdMailingIntegrationTest.java | 217 ++++++++++++++++ .../subscribers/QuotaThresholdNoticeTest.java | 202 +++++++++++++++ .../quota/model/HistoryEvolutionTest.java | 104 ++++++++ .../quota/model/QuotaThresholdChangeTest.java | 59 +++++ .../quota/model/QuotaThresholdFixture.java | 149 +++++++++++ .../quota/model/QuotaThresholdHistoryTest.java | 87 +++++++ .../mailbox/quota/model/QuotaThresholdTest.java | 141 +++++++++++ .../quota/model/QuotaThresholdsTest.java | 83 ++++++ mailbox/pom.xml | 2 + .../store/event/MailboxEventDispatcher.java | 8 +- .../apache/mailet/base/MailAddressFixture.java | 1 + .../mailet/base/test/FakeMailContext.java | 18 +- pom.xml | 11 + 54 files changed, 4071 insertions(+), 22 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/james-project/blob/0e437d25/mailbox/api/src/main/java/org/apache/james/mailbox/MailboxListener.java ---------------------------------------------------------------------- diff --git a/mailbox/api/src/main/java/org/apache/james/mailbox/MailboxListener.java b/mailbox/api/src/main/java/org/apache/james/mailbox/MailboxListener.java index de16426..664a107 100644 --- a/mailbox/api/src/main/java/org/apache/james/mailbox/MailboxListener.java +++ b/mailbox/api/src/main/java/org/apache/james/mailbox/MailboxListener.java @@ -20,6 +20,7 @@ package org.apache.james.mailbox; import java.io.Serializable; +import java.time.Instant; import java.util.List; import java.util.Objects; @@ -71,12 +72,14 @@ public interface MailboxListener { private final QuotaRoot quotaRoot; private final Quota<QuotaCount> countQuota; private final Quota<QuotaSize> sizeQuota; + private final Instant instant; - public QuotaUsageUpdatedEvent(MailboxSession session, QuotaRoot quotaRoot, Quota<QuotaCount> countQuota, Quota<QuotaSize> sizeQuota) { + public QuotaUsageUpdatedEvent(MailboxSession session, QuotaRoot quotaRoot, Quota<QuotaCount> countQuota, Quota<QuotaSize> sizeQuota, Instant instant) { this.session = session; this.quotaRoot = quotaRoot; this.countQuota = countQuota; this.sizeQuota = sizeQuota; + this.instant = instant; } @Override @@ -97,6 +100,10 @@ public interface MailboxListener { return quotaRoot; } + public Instant getInstant() { + return instant; + } + @Override public final boolean equals(Object o) { if (o instanceof QuotaUsageUpdatedEvent) { @@ -105,14 +112,15 @@ public interface MailboxListener { return Objects.equals(this.session, that.session) && Objects.equals(this.quotaRoot, that.quotaRoot) && Objects.equals(this.countQuota, that.countQuota) - && Objects.equals(this.sizeQuota, that.sizeQuota); + && Objects.equals(this.sizeQuota, that.sizeQuota) + && Objects.equals(this.instant, that.instant); } return false; } @Override public final int hashCode() { - return Objects.hash(session, quotaRoot, countQuota, sizeQuota); + return Objects.hash(session, quotaRoot, countQuota, sizeQuota, instant); } } http://git-wip-us.apache.org/repos/asf/james-project/blob/0e437d25/mailbox/api/src/test/java/org/apache/james/mailbox/MailboxManagerTest.java ---------------------------------------------------------------------- diff --git a/mailbox/api/src/test/java/org/apache/james/mailbox/MailboxManagerTest.java b/mailbox/api/src/test/java/org/apache/james/mailbox/MailboxManagerTest.java index 93573f1..cda734b 100644 --- a/mailbox/api/src/test/java/org/apache/james/mailbox/MailboxManagerTest.java +++ b/mailbox/api/src/test/java/org/apache/james/mailbox/MailboxManagerTest.java @@ -23,6 +23,7 @@ import static org.assertj.core.api.Assertions.assertThatCode; import java.io.UnsupportedEncodingException; import java.nio.charset.StandardCharsets; +import java.time.Instant; import java.util.List; import java.util.Optional; @@ -928,17 +929,8 @@ public abstract class MailboxManagerTest { .build(message), session); assertThat(listener.getEvents()) - .contains(new MailboxListener.QuotaUsageUpdatedEvent( - session, - QuotaRoot.quotaRoot("#private&" + USER_1, Optional.empty()), - Quota.<QuotaCount>builder() - .used(QuotaCount.count(1)) - .computedLimit(QuotaCount.unlimited()) - .build(), - Quota.<QuotaSize>builder() - .used(QuotaSize.size(85)) - .computedLimit(QuotaSize.unlimited()) - .build())); + .filteredOn(event -> event instanceof MailboxListener.QuotaUsageUpdatedEvent) + .isNotEmpty(); } @Test http://git-wip-us.apache.org/repos/asf/james-project/blob/0e437d25/mailbox/memory/pom.xml ---------------------------------------------------------------------- diff --git a/mailbox/memory/pom.xml b/mailbox/memory/pom.xml index 3994681..21b245c 100644 --- a/mailbox/memory/pom.xml +++ b/mailbox/memory/pom.xml @@ -64,6 +64,7 @@ <dependency> <groupId>junit</groupId> <artifactId>junit</artifactId> + <scope>test</scope> </dependency> <dependency> <groupId>nl.jqno.equalsverifier</groupId> http://git-wip-us.apache.org/repos/asf/james-project/blob/0e437d25/mailbox/plugin/quota-mailing-memory/pom.xml ---------------------------------------------------------------------- diff --git a/mailbox/plugin/quota-mailing-memory/pom.xml b/mailbox/plugin/quota-mailing-memory/pom.xml new file mode 100644 index 0000000..612ed6a --- /dev/null +++ b/mailbox/plugin/quota-mailing-memory/pom.xml @@ -0,0 +1,96 @@ +<?xml version="1.0" encoding="UTF-8"?> +<!-- + 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. +--> +<project xmlns="http://maven.apache.org/POM/4.0.0"; + xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; + xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd";> + <modelVersion>4.0.0</modelVersion> + + <parent> + <artifactId>apache-james-mailbox</artifactId> + <groupId>org.apache.james</groupId> + <version>3.1.0-SNAPSHOT</version> + <relativePath>../../pom.xml</relativePath> + </parent> + + <artifactId>quota-mailing-memory</artifactId> + <name>Apache James :: Mailbox :: Plugin :: Quota Mailing :: Memory</name> + <description>Apache James Mailbox memory implementation of Quota mailing listener</description> + + <dependencies> + <dependency> + <groupId>${project.groupId}</groupId> + <artifactId>apache-james-mailbox-api</artifactId> + <type>test-jar</type> + <scope>test</scope> + </dependency> + <dependency> + <groupId>${project.groupId}</groupId> + <artifactId>apache-james-mailbox-quota-mailing</artifactId> + </dependency> + <dependency> + <groupId>${project.groupId}</groupId> + <artifactId>apache-james-mailbox-quota-mailing</artifactId> + <type>test-jar</type> + <scope>test</scope> + </dependency> + <dependency> + <groupId>${project.groupId}</groupId> + <artifactId>apache-james-mailbox-store</artifactId> + <scope>test</scope> + </dependency> + <dependency> + <groupId>org.apache.james</groupId> + <artifactId>apache-mailet-base</artifactId> + <type>test-jar</type> + <scope>test</scope> + </dependency> + <dependency> + <groupId>${project.groupId}</groupId> + <artifactId>james-server-data-memory</artifactId> + <scope>test</scope> + </dependency> + <dependency> + <groupId>nl.jqno.equalsverifier</groupId> + <artifactId>equalsverifier</artifactId> + <scope>test</scope> + </dependency> + <dependency> + <groupId>org.assertj</groupId> + <artifactId>assertj-core</artifactId> + <scope>test</scope> + </dependency> + <dependency> + <groupId>org.junit.jupiter</groupId> + <artifactId>junit-jupiter-engine</artifactId> + <scope>test</scope> + </dependency> + <dependency> + <groupId>org.junit.platform</groupId> + <artifactId>junit-platform-launcher</artifactId> + <scope>test</scope> + </dependency> + <dependency> + <groupId>org.mockito</groupId> + <artifactId>mockito-core</artifactId> + <scope>test</scope> + </dependency> + </dependencies> + +</project> \ No newline at end of file http://git-wip-us.apache.org/repos/asf/james-project/blob/0e437d25/mailbox/plugin/quota-mailing-memory/src/main/java/org/apache/james/eventsource/InMemoryEventStore.java ---------------------------------------------------------------------- diff --git a/mailbox/plugin/quota-mailing-memory/src/main/java/org/apache/james/eventsource/InMemoryEventStore.java b/mailbox/plugin/quota-mailing-memory/src/main/java/org/apache/james/eventsource/InMemoryEventStore.java new file mode 100644 index 0000000..91cdbb9 --- /dev/null +++ b/mailbox/plugin/quota-mailing-memory/src/main/java/org/apache/james/eventsource/InMemoryEventStore.java @@ -0,0 +1,104 @@ +/**************************************************************** + * 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.eventsource; + +import java.util.List; +import java.util.Optional; +import java.util.concurrent.ConcurrentHashMap; + +import org.apache.james.eventsourcing.AggregateId; +import org.apache.james.eventsourcing.Event; +import org.apache.james.eventsourcing.EventStore; + +import com.google.common.base.Preconditions; +import com.google.common.collect.ImmutableList; + +public class InMemoryEventStore implements EventStore { + + private final ConcurrentHashMap<AggregateId, History> store; + + public InMemoryEventStore() { + this.store = new ConcurrentHashMap<>(); + } + + @Override + public void appendAll(List<Event> events) { + if (events.isEmpty()) { + return; + } + AggregateId aggregateId = getAggregateId(events); + + if (!store.containsKey(aggregateId)) { + appendToEmptyHistory(aggregateId, events); + } else { + appendToExistingHistory(aggregateId, events); + } + } + + private AggregateId getAggregateId(List<? extends Event> events) { + Preconditions.checkArgument(!events.isEmpty()); + Preconditions.checkArgument(belongsToSameAggregate(events)); + return events.stream() + .map(Event::getAggregateId) + .findFirst() + .get(); + } + + private boolean belongsToSameAggregate(List<? extends Event> events) { + return events.stream() + .map(Event::getAggregateId) + .distinct() + .limit(2) + .count() <= 1; + } + + private void appendToEmptyHistory(AggregateId aggregateId, List<Event> events) { + History newHistory = History.of(events); + + History previousHistory = store.putIfAbsent(aggregateId, newHistory); + if (previousHistory != null) { + throw new EventStore.EventStoreFailedException(); + } + } + + private void appendToExistingHistory(AggregateId aggregateId, List<? extends Event> events) { + History currentHistory = store.get(aggregateId); + List<Event> updatedEvents = updatedEvents(currentHistory, events); + History updatedHistory = History.of(updatedEvents); + + boolean isReplaced = store.replace(aggregateId, currentHistory, updatedHistory); + if (!isReplaced) { + throw new EventStore.EventStoreFailedException(); + } + } + + private List<Event> updatedEvents(History currentHistory, List<? extends Event> newEvents) { + return ImmutableList.<Event>builder() + .addAll(currentHistory.getEvents()) + .addAll(newEvents) + .build(); + } + + @Override + public History getEventsOfAggregate(AggregateId aggregateId) { + return Optional.ofNullable(store.get(aggregateId)) + .orElse(History.empty()); + } +} http://git-wip-us.apache.org/repos/asf/james-project/blob/0e437d25/mailbox/plugin/quota-mailing-memory/src/test/java/org/apache/james/eventsourcing/InMemoryEventSourcingSystemTest.java ---------------------------------------------------------------------- diff --git a/mailbox/plugin/quota-mailing-memory/src/test/java/org/apache/james/eventsourcing/InMemoryEventSourcingSystemTest.java b/mailbox/plugin/quota-mailing-memory/src/test/java/org/apache/james/eventsourcing/InMemoryEventSourcingSystemTest.java new file mode 100644 index 0000000..c2f9935 --- /dev/null +++ b/mailbox/plugin/quota-mailing-memory/src/test/java/org/apache/james/eventsourcing/InMemoryEventSourcingSystemTest.java @@ -0,0 +1,27 @@ +/**************************************************************** + * 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.eventsourcing; + +import org.junit.jupiter.api.extension.ExtendWith; + +@ExtendWith(InMemoryEventStoreExtension.class) +public class InMemoryEventSourcingSystemTest implements EventSourcingSystemTest { + +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/james-project/blob/0e437d25/mailbox/plugin/quota-mailing-memory/src/test/java/org/apache/james/eventsourcing/InMemoryEventStoreExtension.java ---------------------------------------------------------------------- diff --git a/mailbox/plugin/quota-mailing-memory/src/test/java/org/apache/james/eventsourcing/InMemoryEventStoreExtension.java b/mailbox/plugin/quota-mailing-memory/src/test/java/org/apache/james/eventsourcing/InMemoryEventStoreExtension.java new file mode 100644 index 0000000..c4b345e --- /dev/null +++ b/mailbox/plugin/quota-mailing-memory/src/test/java/org/apache/james/eventsourcing/InMemoryEventStoreExtension.java @@ -0,0 +1,39 @@ +/**************************************************************** + * 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.eventsourcing; + +import org.apache.james.eventsource.InMemoryEventStore; +import org.junit.jupiter.api.extension.ExtensionContext; +import org.junit.jupiter.api.extension.ParameterContext; +import org.junit.jupiter.api.extension.ParameterResolutionException; +import org.junit.jupiter.api.extension.ParameterResolver; + +public class InMemoryEventStoreExtension implements ParameterResolver { + + @Override + public boolean supportsParameter(ParameterContext parameterContext, ExtensionContext extensionContext) throws ParameterResolutionException { + return (parameterContext.getParameter().getType() == EventStore.class); + } + + @Override + public Object resolveParameter(ParameterContext parameterContext, ExtensionContext extensionContext) throws ParameterResolutionException { + return new InMemoryEventStore(); + } +} http://git-wip-us.apache.org/repos/asf/james-project/blob/0e437d25/mailbox/plugin/quota-mailing-memory/src/test/java/org/apache/james/eventsourcing/InMemoryEventStoreTest.java ---------------------------------------------------------------------- diff --git a/mailbox/plugin/quota-mailing-memory/src/test/java/org/apache/james/eventsourcing/InMemoryEventStoreTest.java b/mailbox/plugin/quota-mailing-memory/src/test/java/org/apache/james/eventsourcing/InMemoryEventStoreTest.java new file mode 100644 index 0000000..eddc2d0 --- /dev/null +++ b/mailbox/plugin/quota-mailing-memory/src/test/java/org/apache/james/eventsourcing/InMemoryEventStoreTest.java @@ -0,0 +1,27 @@ +/**************************************************************** + * 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.eventsourcing; + +import org.junit.jupiter.api.extension.ExtendWith; + +@ExtendWith(InMemoryEventStoreExtension.class) +public class InMemoryEventStoreTest implements EventStoreTest { + +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/james-project/blob/0e437d25/mailbox/plugin/quota-mailing-memory/src/test/java/org/apache/james/mailbox/quota/memory/listeners/InMemoryQuotaMailingListenersIntegrationTest.java ---------------------------------------------------------------------- diff --git a/mailbox/plugin/quota-mailing-memory/src/test/java/org/apache/james/mailbox/quota/memory/listeners/InMemoryQuotaMailingListenersIntegrationTest.java b/mailbox/plugin/quota-mailing-memory/src/test/java/org/apache/james/mailbox/quota/memory/listeners/InMemoryQuotaMailingListenersIntegrationTest.java new file mode 100644 index 0000000..4b17682 --- /dev/null +++ b/mailbox/plugin/quota-mailing-memory/src/test/java/org/apache/james/mailbox/quota/memory/listeners/InMemoryQuotaMailingListenersIntegrationTest.java @@ -0,0 +1,29 @@ +/**************************************************************** + * 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.quota.memory.listeners; + +import org.apache.james.eventsourcing.InMemoryEventStoreExtension; +import org.apache.james.mailbox.quota.mailing.listeners.QuotaThresholdMailingIntegrationTest; +import org.junit.jupiter.api.extension.ExtendWith; + +@ExtendWith(InMemoryEventStoreExtension.class) +public class InMemoryQuotaMailingListenersIntegrationTest implements QuotaThresholdMailingIntegrationTest { + +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/james-project/blob/0e437d25/mailbox/plugin/quota-mailing/pom.xml ---------------------------------------------------------------------- diff --git a/mailbox/plugin/quota-mailing/pom.xml b/mailbox/plugin/quota-mailing/pom.xml new file mode 100644 index 0000000..480bae9 --- /dev/null +++ b/mailbox/plugin/quota-mailing/pom.xml @@ -0,0 +1,103 @@ +<?xml version="1.0" encoding="UTF-8"?> +<!-- + 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. +--> +<project xmlns="http://maven.apache.org/POM/4.0.0"; + xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; + xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd";> + <modelVersion>4.0.0</modelVersion> + + <parent> + <artifactId>apache-james-mailbox</artifactId> + <groupId>org.apache.james</groupId> + <version>3.1.0-SNAPSHOT</version> + <relativePath>../../pom.xml</relativePath> + </parent> + + <artifactId>apache-james-mailbox-quota-mailing</artifactId> + <name>Apache James :: Mailbox :: Plugin :: Quota Mailing</name> + <description>Apache James Mailbox Quota mailing listener</description> + + <dependencies> + <dependency> + <groupId>${project.groupId}</groupId> + <artifactId>apache-james-mailbox-api</artifactId> + </dependency> + <dependency> + <groupId>${project.groupId}</groupId> + <artifactId>apache-james-mailbox-api</artifactId> + <type>test-jar</type> + <scope>test</scope> + </dependency> + <dependency> + <groupId>${project.groupId}</groupId> + <artifactId>apache-james-mailbox-store</artifactId> + <scope>test</scope> + </dependency> + <dependency> + <groupId>${project.groupId}</groupId> + <artifactId>apache-mailet-api</artifactId> + </dependency> + <dependency> + <groupId>org.apache.james</groupId> + <artifactId>apache-mailet-base</artifactId> + <type>test-jar</type> + <scope>test</scope> + </dependency> + <dependency> + <groupId>${project.groupId}</groupId> + <artifactId>james-server-data-api</artifactId> + </dependency> + <dependency> + <groupId>${project.groupId}</groupId> + <artifactId>james-server-data-memory</artifactId> + <scope>test</scope> + </dependency> + <dependency> + <groupId>javax.inject</groupId> + <artifactId>javax.inject</artifactId> + </dependency> + <dependency> + <groupId>nl.jqno.equalsverifier</groupId> + <artifactId>equalsverifier</artifactId> + <scope>test</scope> + </dependency> + <dependency> + <groupId>org.assertj</groupId> + <artifactId>assertj-core</artifactId> + <scope>test</scope> + </dependency> + <dependency> + <groupId>org.junit.jupiter</groupId> + <artifactId>junit-jupiter-engine</artifactId> + <scope>test</scope> + </dependency> + <dependency> + <groupId>org.junit.platform</groupId> + <artifactId>junit-platform-launcher</artifactId> + <scope>test</scope> + </dependency> + <dependency> + <groupId>org.mockito</groupId> + <artifactId>mockito-core</artifactId> + <scope>test</scope> + </dependency> + </dependencies> + + +</project> \ No newline at end of file http://git-wip-us.apache.org/repos/asf/james-project/blob/0e437d25/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/eventsourcing/AggregateId.java ---------------------------------------------------------------------- diff --git a/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/eventsourcing/AggregateId.java b/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/eventsourcing/AggregateId.java new file mode 100644 index 0000000..18c6224 --- /dev/null +++ b/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/eventsourcing/AggregateId.java @@ -0,0 +1,24 @@ +/**************************************************************** + * 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.eventsourcing; + +public interface AggregateId { + String asAggregateKey(); +} http://git-wip-us.apache.org/repos/asf/james-project/blob/0e437d25/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/eventsourcing/CommandDispatcher.java ---------------------------------------------------------------------- diff --git a/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/eventsourcing/CommandDispatcher.java b/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/eventsourcing/CommandDispatcher.java new file mode 100644 index 0000000..ef5bfd4 --- /dev/null +++ b/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/eventsourcing/CommandDispatcher.java @@ -0,0 +1,115 @@ +/**************************************************************** + * 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.eventsourcing; + +import java.util.Collection; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.function.Supplier; +import java.util.stream.IntStream; + +import javax.inject.Inject; + +import com.github.steveash.guavate.Guavate; + +public class CommandDispatcher { + + private static final int MAX_RETRY = 10; + + public interface Command { + } + + public class UnknownCommandException extends RuntimeException { + private final Command command; + + public UnknownCommandException(Command command) { + super(String.format("Unknown command %s", command)); + this.command = command; + } + + public Command getCommand() { + return command; + } + } + + public class TooManyRetries extends RuntimeException { + private final Command command; + private final int retries; + + + public TooManyRetries(Command command, int retries) { + super(String.format("Too much retries for command %s. Store failure after %d retries", command, retries)); + this.command = command; + this.retries = retries; + } + + + public Command getCommand() { + return command; + } + + public int getRetries() { + return retries; + } + } + + public interface CommandHandler<C extends Command> { + Class<C> handledClass(); + + List<? extends Event> handle(C c); + } + + private final EventBus eventBus; + private final Map<Class, CommandHandler> handlers; + + @Inject + public CommandDispatcher(EventBus eventBus, Collection<CommandHandler> handlers) { + this.eventBus = eventBus; + this.handlers = handlers.stream() + .collect(Guavate.toImmutableMap(CommandHandler::handledClass, handler -> handler)); + } + + public void dispatch(Command c) { + trySeveralTimes(() -> tryDispatch(c)) + .orElseThrow(() -> new TooManyRetries(c, MAX_RETRY)); + } + + public Optional<Integer> trySeveralTimes(Supplier<Boolean> singleTry) { + return IntStream.range(0, MAX_RETRY) + .boxed() + .filter(any -> singleTry.get()) + .findFirst(); + } + + private boolean tryDispatch(Command c) { + try { + List<Event> events = + Optional.ofNullable(handlers.get(c.getClass())) + .map(f -> f.handle(c)) + .orElseThrow(() -> new UnknownCommandException(c)); + + eventBus.publish(events); + return true; + } catch (EventStore.EventStoreFailedException e) { + return false; + } + } +} http://git-wip-us.apache.org/repos/asf/james-project/blob/0e437d25/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/eventsourcing/Event.java ---------------------------------------------------------------------- diff --git a/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/eventsourcing/Event.java b/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/eventsourcing/Event.java new file mode 100644 index 0000000..7b3bc00 --- /dev/null +++ b/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/eventsourcing/Event.java @@ -0,0 +1,32 @@ +/**************************************************************** + * 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.eventsourcing; + +public interface Event extends Comparable<Event> { + + EventId eventId(); + + AggregateId getAggregateId(); + + @Override + default int compareTo(Event o) { + return eventId().compareTo(o.eventId()); + } +} http://git-wip-us.apache.org/repos/asf/james-project/blob/0e437d25/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/eventsourcing/EventBus.java ---------------------------------------------------------------------- diff --git a/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/eventsourcing/EventBus.java b/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/eventsourcing/EventBus.java new file mode 100644 index 0000000..067d432 --- /dev/null +++ b/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/eventsourcing/EventBus.java @@ -0,0 +1,61 @@ +/**************************************************************** + * 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.eventsourcing; + +import java.util.List; +import java.util.Set; + +import javax.inject.Inject; + +import org.apache.commons.lang3.tuple.Pair; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.google.common.collect.ImmutableSet; + +public class EventBus { + + public static final Logger LOGGER = LoggerFactory.getLogger(EventBus.class); + private final EventStore eventStore; + private final Set<Subscriber> subscribers; + + @Inject + public EventBus(EventStore eventStore, Set<Subscriber> subscribers) { + this.eventStore = eventStore; + this.subscribers = ImmutableSet.copyOf(subscribers); + } + + public void publish(List<Event> events) throws EventStore.EventStoreFailedException { + eventStore.appendAll(events); + events.stream() + .flatMap(event -> subscribers.stream().map(subscriber -> Pair.of(event, subscriber))) + .forEach(this::handle); + } + + public void handle(Pair<Event, Subscriber> pair) { + Subscriber subscriber = pair.getRight(); + Event event = pair.getLeft(); + try { + subscriber.handle(event); + } catch (Exception e) { + LOGGER.error("Error while calling {} for {}", subscriber, event, e); + } + } +} http://git-wip-us.apache.org/repos/asf/james-project/blob/0e437d25/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/eventsourcing/EventId.java ---------------------------------------------------------------------- diff --git a/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/eventsourcing/EventId.java b/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/eventsourcing/EventId.java new file mode 100644 index 0000000..1d636f6 --- /dev/null +++ b/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/eventsourcing/EventId.java @@ -0,0 +1,78 @@ +/**************************************************************** + * 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.eventsourcing; + +import java.util.Objects; +import java.util.Optional; + +import com.google.common.base.MoreObjects; +import com.google.common.base.Preconditions; + +public class EventId implements Comparable<EventId> { + + public static EventId first() { + return new EventId(0); + } + + private final long value; + + private EventId(long value) { + Preconditions.checkArgument(value >= 0, "EventId can not be negative"); + this.value = value; + } + + public EventId next() { + return new EventId(value + 1); + } + + public Optional<EventId> previous() { + if (value > 0) { + return Optional.of(new EventId(value - 1)); + } + return Optional.empty(); + } + + @Override + public int compareTo(EventId o) { + return Long.compare(value, o.value); + } + + @Override + public final boolean equals(Object o) { + if (o instanceof EventId) { + EventId eventId = (EventId) o; + + return Objects.equals(this.value, eventId.value); + } + return false; + } + + @Override + public final int hashCode() { + return Objects.hash(value); + } + + @Override + public String toString() { + return MoreObjects.toStringHelper(this) + .add("value", value) + .toString(); + } +} http://git-wip-us.apache.org/repos/asf/james-project/blob/0e437d25/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/eventsourcing/EventSourcingSystem.java ---------------------------------------------------------------------- diff --git a/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/eventsourcing/EventSourcingSystem.java b/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/eventsourcing/EventSourcingSystem.java new file mode 100644 index 0000000..e627224 --- /dev/null +++ b/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/eventsourcing/EventSourcingSystem.java @@ -0,0 +1,36 @@ +/**************************************************************** + * 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.eventsourcing; + +import java.util.Set; + +public class EventSourcingSystem { + private final EventBus eventBus; + private final CommandDispatcher commandDispatcher; + + public EventSourcingSystem(Set<CommandDispatcher.CommandHandler> handlers, Set<Subscriber> subscribers, EventStore eventStore) { + this.eventBus = new EventBus(eventStore, subscribers); + this.commandDispatcher = new CommandDispatcher(eventBus, handlers); + } + + public void dispatch(CommandDispatcher.Command c) { + commandDispatcher.dispatch(c); + } +} http://git-wip-us.apache.org/repos/asf/james-project/blob/0e437d25/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/eventsourcing/EventStore.java ---------------------------------------------------------------------- diff --git a/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/eventsourcing/EventStore.java b/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/eventsourcing/EventStore.java new file mode 100644 index 0000000..1ba8028 --- /dev/null +++ b/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/eventsourcing/EventStore.java @@ -0,0 +1,114 @@ +/**************************************************************** + * 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.eventsourcing; + +import java.util.Comparator; +import java.util.List; +import java.util.Objects; +import java.util.Optional; +import java.util.Set; + +import com.github.steveash.guavate.Guavate; +import com.google.common.collect.ImmutableList; + +public interface EventStore { + + class EventStoreFailedException extends RuntimeException { + + } + + class History { + public static History empty() { + return new History(ImmutableList.of()); + } + + public static History of(List<Event> events) { + return new History(ImmutableList.copyOf(events)); + } + + public static History of(Event... events) { + return of(ImmutableList.copyOf(events)); + } + + private final List<Event> events; + + private History(List<Event> events) { + if (hasEventIdDuplicates(events)) { + throw new EventStoreFailedException(); + } + this.events = events; + } + + public boolean hasEventIdDuplicates(List<Event> events) { + Set<EventId> eventIds = events.stream() + .map(Event::eventId) + .collect(Guavate.toImmutableSet()); + + return eventIds.size() != events.size(); + } + + public Optional<EventId> getVersion() { + return events.stream() + .map(Event::eventId) + .max(Comparator.naturalOrder()); + } + + public List<Event> getEvents() { + return events; + } + + public EventId getNextEventId() { + return getVersion() + .map(EventId::next) + .orElse(EventId.first()); + } + + @Override + public final boolean equals(Object o) { + if (o instanceof History) { + History history = (History) o; + + return Objects.equals(this.events, history.events); + } + return false; + } + + @Override + public final int hashCode() { + return Objects.hash(events); + } + } + + default void append(Event event) { + appendAll(ImmutableList.of(event)); + } + + default void appendAll(Event... events) { + appendAll(ImmutableList.copyOf(events)); + } + + /** + * This method should check that no input event has an id already stored and throw otherwise + * It should also check that all events belong to the same aggregate + */ + void appendAll(List<Event> events); + + History getEventsOfAggregate(AggregateId aggregateId); +} http://git-wip-us.apache.org/repos/asf/james-project/blob/0e437d25/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/eventsourcing/Subscriber.java ---------------------------------------------------------------------- diff --git a/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/eventsourcing/Subscriber.java b/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/eventsourcing/Subscriber.java new file mode 100644 index 0000000..42a804d --- /dev/null +++ b/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/eventsourcing/Subscriber.java @@ -0,0 +1,24 @@ +/**************************************************************** + * 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.eventsourcing; + +public interface Subscriber { + void handle(Event event); +} http://git-wip-us.apache.org/repos/asf/james-project/blob/0e437d25/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/mailbox/quota/mailing/QuotaMailingListenerConfiguration.java ---------------------------------------------------------------------- diff --git a/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/mailbox/quota/mailing/QuotaMailingListenerConfiguration.java b/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/mailbox/quota/mailing/QuotaMailingListenerConfiguration.java new file mode 100644 index 0000000..25d504c --- /dev/null +++ b/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/mailbox/quota/mailing/QuotaMailingListenerConfiguration.java @@ -0,0 +1,74 @@ +/**************************************************************** + * 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.quota.mailing; + +import java.time.Duration; +import java.util.Objects; + +import org.apache.james.mailbox.quota.model.QuotaThresholds; + +import com.google.common.base.MoreObjects; +import com.google.common.collect.ImmutableList; + +public class QuotaMailingListenerConfiguration { + public static QuotaMailingListenerConfiguration DEFAULT = new QuotaMailingListenerConfiguration( + new QuotaThresholds(ImmutableList.of()), + Duration.ofDays(1)); + + private final QuotaThresholds thresholds; + private final Duration gracePeriod; + + public QuotaMailingListenerConfiguration(QuotaThresholds thresholds, Duration gracePeriod) { + this.thresholds = thresholds; + this.gracePeriod = gracePeriod; + } + + public QuotaThresholds getThresholds() { + return thresholds; + } + + public Duration getGracePeriod() { + return gracePeriod; + } + + @Override + public final boolean equals(Object o) { + if (o instanceof QuotaMailingListenerConfiguration) { + QuotaMailingListenerConfiguration that = (QuotaMailingListenerConfiguration) o; + + return Objects.equals(this.thresholds, that.thresholds) + && Objects.equals(this.gracePeriod, that.gracePeriod); + } + return false; + } + + @Override + public final int hashCode() { + return Objects.hash(thresholds, gracePeriod); + } + + @Override + public String toString() { + return MoreObjects.toStringHelper(this) + .add("thresholds", thresholds) + .add("gracePeriod", gracePeriod) + .toString(); + } +} http://git-wip-us.apache.org/repos/asf/james-project/blob/0e437d25/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/mailbox/quota/mailing/aggregates/UserQuotaThresholds.java ---------------------------------------------------------------------- diff --git a/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/mailbox/quota/mailing/aggregates/UserQuotaThresholds.java b/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/mailbox/quota/mailing/aggregates/UserQuotaThresholds.java new file mode 100644 index 0000000..8290baf --- /dev/null +++ b/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/mailbox/quota/mailing/aggregates/UserQuotaThresholds.java @@ -0,0 +1,163 @@ +/**************************************************************** + * 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.quota.mailing.aggregates; + +import java.time.Duration; +import java.time.Instant; +import java.util.List; +import java.util.Objects; +import java.util.stream.Collectors; + +import org.apache.james.core.User; +import org.apache.james.eventsourcing.AggregateId; +import org.apache.james.eventsourcing.EventStore; +import org.apache.james.mailbox.model.Quota; +import org.apache.james.mailbox.quota.QuotaCount; +import org.apache.james.mailbox.quota.QuotaSize; +import org.apache.james.mailbox.quota.mailing.QuotaMailingListenerConfiguration; +import org.apache.james.mailbox.quota.mailing.commands.DetectThresholdCrossing; +import org.apache.james.mailbox.quota.mailing.events.QuotaThresholdChangedEvent; +import org.apache.james.mailbox.quota.model.HistoryEvolution; +import org.apache.james.mailbox.quota.model.QuotaThresholdChange; +import org.apache.james.mailbox.quota.model.QuotaThresholdHistory; +import org.apache.james.mailbox.quota.model.QuotaThresholds; +import org.apache.james.util.OptionalUtils; + +import com.github.steveash.guavate.Guavate; +import com.google.common.collect.ImmutableList; + +public class UserQuotaThresholds { + + public static class Id implements AggregateId { + + public static Id from(User user) { + return new Id(user); + } + + private final User user; + + private Id(User user) { + this.user = user; + } + + public User getUser() { + return user; + } + + @Override + public String asAggregateKey() { + return "QuotaThreasholdEvents-" + user.asString(); + } + + @Override + public final boolean equals(Object o) { + if (o instanceof Id) { + Id id = (Id) o; + + return Objects.equals(this.user, id.user); + } + return false; + } + + @Override + public final int hashCode() { + return Objects.hash(user); + } + } + + public static UserQuotaThresholds fromEvents(Id aggregateId, EventStore.History history) { + return new UserQuotaThresholds(aggregateId, history); + } + + private final Id aggregateId; + private final EventStore.History history; + private final List<QuotaThresholdChangedEvent> events; + + private UserQuotaThresholds(Id aggregateId, EventStore.History history) { + this.aggregateId = aggregateId; + this.history = history; + this.events = history.getEvents().stream() + .map(QuotaThresholdChangedEvent.class::cast) + .collect(Collectors.toList()); + } + + public List<QuotaThresholdChangedEvent> detectThresholdCrossing(QuotaMailingListenerConfiguration configuration, + DetectThresholdCrossing command) { + + List<QuotaThresholdChangedEvent> events = generateEvents( + configuration.getThresholds(), + configuration.getGracePeriod(), + command.getCountQuota(), + command.getSizeQuota(), + command.getInstant()); + events.forEach(this::apply); + return events; + } + + private List<QuotaThresholdChangedEvent> generateEvents(QuotaThresholds configuration, Duration gracePeriod, Quota<QuotaCount> countQuota, Quota<QuotaSize> sizeQuota, Instant now) { + QuotaThresholdChange countThresholdChange = new QuotaThresholdChange(configuration.highestExceededThreshold(countQuota), now); + QuotaThresholdChange sizeThresholdChange = new QuotaThresholdChange(configuration.highestExceededThreshold(sizeQuota), now); + + HistoryEvolution countHistoryEvolution = computeCountHistory() + .compareWithCurrentThreshold(countThresholdChange, gracePeriod); + HistoryEvolution sizeHistoryEvolution = computeSizeHistory() + .compareWithCurrentThreshold(sizeThresholdChange, gracePeriod); + + return generateEvents(countHistoryEvolution, sizeHistoryEvolution, countQuota, sizeQuota); + } + + private QuotaThresholdHistory computeSizeHistory() { + return new QuotaThresholdHistory( + events.stream() + .map(QuotaThresholdChangedEvent::getSizeHistoryEvolution) + .map(HistoryEvolution::getThresholdChange) + .flatMap(OptionalUtils::toStream) + .collect(Guavate.toImmutableList())); + } + + private QuotaThresholdHistory computeCountHistory() { + return new QuotaThresholdHistory( + events.stream() + .map(QuotaThresholdChangedEvent::getCountHistoryEvolution) + .map(HistoryEvolution::getThresholdChange) + .flatMap(OptionalUtils::toStream) + .collect(Guavate.toImmutableList())); + } + + private List<QuotaThresholdChangedEvent> generateEvents(HistoryEvolution countHistoryEvolution, HistoryEvolution sizeHistoryEvolution, Quota<QuotaCount> countQuota, Quota<QuotaSize> sizeQuota) { + if (countHistoryEvolution.isChange() || sizeHistoryEvolution.isChange()) { + return ImmutableList.of( + new QuotaThresholdChangedEvent( + history.getNextEventId(), + sizeHistoryEvolution, + countHistoryEvolution, + sizeQuota, + countQuota, + aggregateId)); + } + + return ImmutableList.of(); + } + + private void apply(QuotaThresholdChangedEvent event) { + events.add(event); + } + +} http://git-wip-us.apache.org/repos/asf/james-project/blob/0e437d25/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/mailbox/quota/mailing/commands/DetectThresholdCrossing.java ---------------------------------------------------------------------- diff --git a/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/mailbox/quota/mailing/commands/DetectThresholdCrossing.java b/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/mailbox/quota/mailing/commands/DetectThresholdCrossing.java new file mode 100644 index 0000000..0de0023 --- /dev/null +++ b/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/mailbox/quota/mailing/commands/DetectThresholdCrossing.java @@ -0,0 +1,77 @@ +/**************************************************************** + * 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.quota.mailing.commands; + +import java.time.Instant; +import java.util.Objects; + +import org.apache.james.core.User; +import org.apache.james.eventsourcing.CommandDispatcher; +import org.apache.james.mailbox.model.Quota; +import org.apache.james.mailbox.quota.QuotaCount; +import org.apache.james.mailbox.quota.QuotaSize; + +public class DetectThresholdCrossing implements CommandDispatcher.Command { + + private final User user; + private final Quota<QuotaCount> countQuota; + private final Quota<QuotaSize> sizeQuota; + private final Instant instant; + + public DetectThresholdCrossing(User user, Quota<QuotaCount> countQuota, Quota<QuotaSize> sizeQuota, Instant instant) { + this.user = user; + this.countQuota = countQuota; + this.sizeQuota = sizeQuota; + this.instant = instant; + } + + public User getUser() { + return user; + } + + public Quota<QuotaCount> getCountQuota() { + return countQuota; + } + + public Quota<QuotaSize> getSizeQuota() { + return sizeQuota; + } + + public Instant getInstant() { + return instant; + } + + @Override + public final boolean equals(Object o) { + if (o instanceof DetectThresholdCrossing) { + DetectThresholdCrossing that = (DetectThresholdCrossing) o; + + return Objects.equals(this.user, that.user) + && Objects.equals(this.countQuota, that.countQuota) + && Objects.equals(this.sizeQuota, that.sizeQuota); + } + return false; + } + + @Override + public final int hashCode() { + return Objects.hash(user, countQuota, sizeQuota); + } +} http://git-wip-us.apache.org/repos/asf/james-project/blob/0e437d25/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/mailbox/quota/mailing/commands/DetectThresholdCrossingHandler.java ---------------------------------------------------------------------- diff --git a/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/mailbox/quota/mailing/commands/DetectThresholdCrossingHandler.java b/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/mailbox/quota/mailing/commands/DetectThresholdCrossingHandler.java new file mode 100644 index 0000000..39b2013 --- /dev/null +++ b/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/mailbox/quota/mailing/commands/DetectThresholdCrossingHandler.java @@ -0,0 +1,59 @@ +/**************************************************************** + * 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.quota.mailing.commands; + +import java.util.List; + +import javax.inject.Inject; + +import org.apache.james.eventsourcing.CommandDispatcher; +import org.apache.james.eventsourcing.Event; +import org.apache.james.eventsourcing.EventStore; +import org.apache.james.mailbox.quota.mailing.QuotaMailingListenerConfiguration; +import org.apache.james.mailbox.quota.mailing.aggregates.UserQuotaThresholds; + +public class DetectThresholdCrossingHandler implements CommandDispatcher.CommandHandler<DetectThresholdCrossing> { + + private final EventStore eventStore; + private final QuotaMailingListenerConfiguration quotaMailingListenerConfiguration; + + @Inject + public DetectThresholdCrossingHandler(EventStore eventStore, QuotaMailingListenerConfiguration quotaMailingListenerConfiguration) { + this.eventStore = eventStore; + this.quotaMailingListenerConfiguration = quotaMailingListenerConfiguration; + } + + @Override + public List<? extends Event> handle(DetectThresholdCrossing command) { + return loadAggregate(command) + .detectThresholdCrossing(quotaMailingListenerConfiguration, command); + } + + private UserQuotaThresholds loadAggregate(DetectThresholdCrossing command) { + UserQuotaThresholds.Id aggregateId = UserQuotaThresholds.Id.from(command.getUser()); + EventStore.History history = eventStore.getEventsOfAggregate(aggregateId); + return UserQuotaThresholds.fromEvents(aggregateId, history); + } + + @Override + public Class<DetectThresholdCrossing> handledClass() { + return DetectThresholdCrossing.class; + } +} http://git-wip-us.apache.org/repos/asf/james-project/blob/0e437d25/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/mailbox/quota/mailing/events/QuotaThresholdChangedEvent.java ---------------------------------------------------------------------- diff --git a/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/mailbox/quota/mailing/events/QuotaThresholdChangedEvent.java b/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/mailbox/quota/mailing/events/QuotaThresholdChangedEvent.java new file mode 100644 index 0000000..3c21a3f --- /dev/null +++ b/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/mailbox/quota/mailing/events/QuotaThresholdChangedEvent.java @@ -0,0 +1,75 @@ +/**************************************************************** + * 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.quota.mailing.events; + +import org.apache.james.eventsourcing.Event; +import org.apache.james.eventsourcing.EventId; +import org.apache.james.mailbox.model.Quota; +import org.apache.james.mailbox.quota.QuotaCount; +import org.apache.james.mailbox.quota.QuotaSize; +import org.apache.james.mailbox.quota.mailing.aggregates.UserQuotaThresholds; +import org.apache.james.mailbox.quota.model.HistoryEvolution; + +public class QuotaThresholdChangedEvent implements Event { + + private final EventId eventId; + private final HistoryEvolution sizeHistoryEvolution; + private final HistoryEvolution countHistoryEvolution; + private final Quota<QuotaSize> sizeQuota; + private final Quota<QuotaCount> countQuota; + private final UserQuotaThresholds.Id aggregateId; + + public QuotaThresholdChangedEvent(EventId eventId, HistoryEvolution sizeHistoryEvolution, HistoryEvolution countHistoryEvolution, Quota<QuotaSize> sizeQuota, Quota<QuotaCount> countQuota, UserQuotaThresholds.Id aggregateId) { + this.eventId = eventId; + this.sizeHistoryEvolution = sizeHistoryEvolution; + this.countHistoryEvolution = countHistoryEvolution; + this.sizeQuota = sizeQuota; + this.countQuota = countQuota; + this.aggregateId = aggregateId; + } + + public HistoryEvolution getSizeHistoryEvolution() { + return sizeHistoryEvolution; + } + + public HistoryEvolution getCountHistoryEvolution() { + return countHistoryEvolution; + } + + public Quota<QuotaSize> getSizeQuota() { + return sizeQuota; + } + + public Quota<QuotaCount> getCountQuota() { + return countQuota; + } + + @Override + public EventId eventId() { + return eventId; + } + + @Override + public UserQuotaThresholds.Id getAggregateId() { + return aggregateId; + } + + +} http://git-wip-us.apache.org/repos/asf/james-project/blob/0e437d25/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/mailbox/quota/mailing/listeners/QuotaThresholdCrossingListener.java ---------------------------------------------------------------------- diff --git a/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/mailbox/quota/mailing/listeners/QuotaThresholdCrossingListener.java b/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/mailbox/quota/mailing/listeners/QuotaThresholdCrossingListener.java new file mode 100644 index 0000000..a4b1094 --- /dev/null +++ b/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/mailbox/quota/mailing/listeners/QuotaThresholdCrossingListener.java @@ -0,0 +1,71 @@ +/**************************************************************** + * 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.quota.mailing.listeners; + +import org.apache.james.core.User; +import org.apache.james.eventsourcing.EventSourcingSystem; +import org.apache.james.mailbox.Event; +import org.apache.james.mailbox.MailboxListener; +import org.apache.james.mailbox.quota.mailing.commands.DetectThresholdCrossing; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class QuotaThresholdCrossingListener implements MailboxListener { + private static final Logger LOGGER = LoggerFactory.getLogger(QuotaThresholdCrossingListener.class); + + private final EventSourcingSystem eventSourcingSystem; + + public QuotaThresholdCrossingListener(EventSourcingSystem eventSourcingSystem) { + this.eventSourcingSystem = eventSourcingSystem; + } + + @Override + public ListenerType getType() { + return ListenerType.ONCE; + } + + @Override + public ExecutionMode getExecutionMode() { + return ExecutionMode.SYNCHRONOUS; + } + + @Override + public void event(Event event) { + try { + if (event instanceof QuotaUsageUpdatedEvent) { + handleEvent(getUser(event), (QuotaUsageUpdatedEvent) event); + } + } catch (Exception e) { + LOGGER.error("Can not re-emmit quota threshold events", e); + } + } + + private void handleEvent(User user, QuotaUsageUpdatedEvent event) { + eventSourcingSystem.dispatch( + new DetectThresholdCrossing(user, event.getCountQuota(), event.getSizeQuota(), event.getInstant())); + } + + private User getUser(Event event) { + return User.fromUsername( + event.getSession() + .getUser() + .getUserName()); + } +} http://git-wip-us.apache.org/repos/asf/james-project/blob/0e437d25/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/mailbox/quota/mailing/subscribers/QuotaThresholdMailer.java ---------------------------------------------------------------------- diff --git a/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/mailbox/quota/mailing/subscribers/QuotaThresholdMailer.java b/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/mailbox/quota/mailing/subscribers/QuotaThresholdMailer.java new file mode 100644 index 0000000..2b5cc94 --- /dev/null +++ b/mailbox/plugin/quota-mailing/src/main/java/org/apache/james/mailbox/quota/mailing/subscribers/QuotaThresholdMailer.java @@ -0,0 +1,79 @@ +/**************************************************************** + * 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.quota.mailing.subscribers; + +import java.util.Optional; + +import javax.mail.MessagingException; + +import org.apache.james.core.MailAddress; +import org.apache.james.core.User; +import org.apache.james.core.builder.MimeMessageBuilder; +import org.apache.james.eventsourcing.Event; +import org.apache.james.eventsourcing.Subscriber; +import org.apache.james.mailbox.quota.mailing.events.QuotaThresholdChangedEvent; +import org.apache.james.user.api.UsersRepository; +import org.apache.james.user.api.UsersRepositoryException; +import org.apache.mailet.MailetContext; + +import com.github.fge.lambdas.Throwing; +import com.google.common.collect.ImmutableList; + +public class QuotaThresholdMailer implements Subscriber { + private final MailetContext mailetContext; + private final UsersRepository usersRepository; + + public QuotaThresholdMailer(MailetContext mailetContext, UsersRepository usersRepository) { + this.mailetContext = mailetContext; + this.usersRepository = usersRepository; + } + + @Override + public void handle(Event event) { + if (event instanceof QuotaThresholdChangedEvent) { + handleEvent((QuotaThresholdChangedEvent) event); + } + } + + private void handleEvent(QuotaThresholdChangedEvent event) { + Optional<QuotaThresholdNotice> maybeNotice = QuotaThresholdNotice.builder() + .countQuota(event.getCountQuota()) + .sizeQuota(event.getSizeQuota()) + .countThreshold(event.getCountHistoryEvolution()) + .sizeThreshold(event.getSizeHistoryEvolution()) + .build(); + + maybeNotice.ifPresent(Throwing.consumer(notice -> sendNotice(notice, event.getAggregateId().getUser()))); + } + + private void sendNotice(QuotaThresholdNotice notice, User user) throws UsersRepositoryException, MessagingException { + MailAddress sender = mailetContext.getPostmaster(); + MailAddress recipient = usersRepository.getMailAddressFor(user); + + mailetContext.sendMail(sender, ImmutableList.of(recipient), + MimeMessageBuilder.mimeMessageBuilder() + .addFrom(sender.asString()) + .addToRecipient(recipient.asString()) + .setSubject("Warning: Your email usage just exceeded a configured threshold") + .setText(notice.generateReport()) + .build()); + } + +} --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
