You are viewing a plain text version of this content. The canonical link for it is here.
Posted to server-dev@james.apache.org by bt...@apache.org on 2019/01/22 10:09:59 UTC
[4/6] james-project git commit: MAILBOX-374 Bind RabbitMQEventBus
module in RabbitMQ related product
MAILBOX-374 Bind RabbitMQEventBus module in RabbitMQ related product
Project: http://git-wip-us.apache.org/repos/asf/james-project/repo
Commit: http://git-wip-us.apache.org/repos/asf/james-project/commit/255f1be5
Tree: http://git-wip-us.apache.org/repos/asf/james-project/tree/255f1be5
Diff: http://git-wip-us.apache.org/repos/asf/james-project/diff/255f1be5
Branch: refs/heads/master
Commit: 255f1be500f100a1f9e094e9f378249728341158
Parents: 04e8f7b
Author: Benoit Tellier <bt...@linagora.com>
Authored: Thu Jan 17 10:22:10 2019 +0700
Committer: Benoit Tellier <bt...@linagora.com>
Committed: Tue Jan 22 17:07:15 2019 +0700
----------------------------------------------------------------------
.../james/mailbox/events/EventDeadLetters.java | 2 +-
.../james/mailbox/events/RabbitMQEventBus.java | 4 +-
.../james/event/json/EventSerializer.scala | 3 +-
pom.xml | 5 ++
.../guice/cassandra-rabbitmq-guice/pom.xml | 4 ++
.../james/CassandraRabbitMQJamesServerMain.java | 3 +-
.../modules/event/RabbitMQEventBusModule.java | 75 ++++++++++++++++++++
.../CassandraRabbitMQSwiftJmapTestRule.java | 11 +++
.../modules/mailbox/DefaultEventModule.java | 4 ++
9 files changed, 107 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/james-project/blob/255f1be5/mailbox/api/src/main/java/org/apache/james/mailbox/events/EventDeadLetters.java
----------------------------------------------------------------------
diff --git a/mailbox/api/src/main/java/org/apache/james/mailbox/events/EventDeadLetters.java b/mailbox/api/src/main/java/org/apache/james/mailbox/events/EventDeadLetters.java
index 238533b..5c0057c 100644
--- a/mailbox/api/src/main/java/org/apache/james/mailbox/events/EventDeadLetters.java
+++ b/mailbox/api/src/main/java/org/apache/james/mailbox/events/EventDeadLetters.java
@@ -24,7 +24,7 @@ import org.apache.james.mailbox.Event;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
-interface EventDeadLetters {
+public interface EventDeadLetters {
Mono<Void> store(Group registeredGroup, Event failDeliveredEvent);
Mono<Void> remove(Group registeredGroup, Event.EventId failDeliveredEventId);
http://git-wip-us.apache.org/repos/asf/james-project/blob/255f1be5/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/RabbitMQEventBus.java
----------------------------------------------------------------------
diff --git a/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/RabbitMQEventBus.java b/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/RabbitMQEventBus.java
index 0ab082c..bad0714 100644
--- a/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/RabbitMQEventBus.java
+++ b/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/RabbitMQEventBus.java
@@ -23,6 +23,7 @@ import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
import javax.annotation.PreDestroy;
+import javax.inject.Inject;
import org.apache.james.backend.rabbitmq.RabbitMQConnectionFactory;
import org.apache.james.event.json.EventSerializer;
@@ -36,7 +37,7 @@ import reactor.rabbitmq.RabbitFlux;
import reactor.rabbitmq.Sender;
import reactor.rabbitmq.SenderOptions;
-class RabbitMQEventBus implements EventBus {
+public class RabbitMQEventBus implements EventBus {
static final String MAILBOX_EVENT = "mailboxEvent";
static final String MAILBOX_EVENT_EXCHANGE_NAME = MAILBOX_EVENT + "-exchange";
static final String EVENT_BUS_ID = "eventBusId";
@@ -55,6 +56,7 @@ class RabbitMQEventBus implements EventBus {
private EventDispatcher eventDispatcher;
private Sender sender;
+ @Inject
RabbitMQEventBus(RabbitMQConnectionFactory rabbitMQConnectionFactory, EventSerializer eventSerializer,
RetryBackoffConfiguration retryBackoff,
RoutingKeyConverter routingKeyConverter,
http://git-wip-us.apache.org/repos/asf/james-project/blob/255f1be5/mailbox/event/json/src/main/scala/org/apache/james/event/json/EventSerializer.scala
----------------------------------------------------------------------
diff --git a/mailbox/event/json/src/main/scala/org/apache/james/event/json/EventSerializer.scala b/mailbox/event/json/src/main/scala/org/apache/james/event/json/EventSerializer.scala
index 03c62eb..592d366 100644
--- a/mailbox/event/json/src/main/scala/org/apache/james/event/json/EventSerializer.scala
+++ b/mailbox/event/json/src/main/scala/org/apache/james/event/json/EventSerializer.scala
@@ -22,6 +22,7 @@ package org.apache.james.event.json
import java.time.Instant
import java.util.{Optional, TreeMap => JavaTreeMap}
+import javax.inject.Inject
import julienrf.json.derived
import org.apache.james.core.quota.{QuotaCount, QuotaSize, QuotaValue}
import org.apache.james.core.{Domain, User}
@@ -330,7 +331,7 @@ class JsonSerialize(mailboxIdFactory: MailboxId.Factory, messageIdFactory: Messa
.map(event => event.toJava)
}
-class EventSerializer(mailboxIdFactory: MailboxId.Factory, messageIdFactory: MessageId.Factory) {
+class EventSerializer @Inject() (mailboxIdFactory: MailboxId.Factory, messageIdFactory: MessageId.Factory) {
private val jsonSerialize = new JsonSerialize(mailboxIdFactory, messageIdFactory)
def toJson(event: JavaEvent): String = jsonSerialize.toJson(event)
http://git-wip-us.apache.org/repos/asf/james-project/blob/255f1be5/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 0119f10..1bc0311 100644
--- a/pom.xml
+++ b/pom.xml
@@ -772,6 +772,11 @@
</dependency>
<dependency>
<groupId>${james.groupId}</groupId>
+ <artifactId>apache-james-mailbox-event-rabbitmq</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>${james.groupId}</groupId>
<artifactId>apache-james-mailbox-jpa</artifactId>
<version>${project.version}</version>
</dependency>
http://git-wip-us.apache.org/repos/asf/james-project/blob/255f1be5/server/container/guice/cassandra-rabbitmq-guice/pom.xml
----------------------------------------------------------------------
diff --git a/server/container/guice/cassandra-rabbitmq-guice/pom.xml b/server/container/guice/cassandra-rabbitmq-guice/pom.xml
index 716b626..ced0a3d 100644
--- a/server/container/guice/cassandra-rabbitmq-guice/pom.xml
+++ b/server/container/guice/cassandra-rabbitmq-guice/pom.xml
@@ -65,6 +65,10 @@
</dependency>
<dependency>
<groupId>${james.groupId}</groupId>
+ <artifactId>apache-james-mailbox-event-rabbitmq</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>${james.groupId}</groupId>
<artifactId>apache-james-mailbox-elasticsearch</artifactId>
<type>test-jar</type>
<scope>test</scope>
http://git-wip-us.apache.org/repos/asf/james-project/blob/255f1be5/server/container/guice/cassandra-rabbitmq-guice/src/main/java/org/apache/james/CassandraRabbitMQJamesServerMain.java
----------------------------------------------------------------------
diff --git a/server/container/guice/cassandra-rabbitmq-guice/src/main/java/org/apache/james/CassandraRabbitMQJamesServerMain.java b/server/container/guice/cassandra-rabbitmq-guice/src/main/java/org/apache/james/CassandraRabbitMQJamesServerMain.java
index 8bcf444..8597644 100644
--- a/server/container/guice/cassandra-rabbitmq-guice/src/main/java/org/apache/james/CassandraRabbitMQJamesServerMain.java
+++ b/server/container/guice/cassandra-rabbitmq-guice/src/main/java/org/apache/james/CassandraRabbitMQJamesServerMain.java
@@ -22,6 +22,7 @@ package org.apache.james;
import static org.apache.james.CassandraJamesServerMain.ALL_BUT_JMX_CASSANDRA_MODULE;
import org.apache.james.modules.blobstore.BlobStoreChoosingModule;
+import org.apache.james.modules.event.RabbitMQEventBusModule;
import org.apache.james.modules.rabbitmq.RabbitMQModule;
import org.apache.james.modules.server.JMXServerModule;
import org.apache.james.server.core.configuration.Configuration;
@@ -33,7 +34,7 @@ public class CassandraRabbitMQJamesServerMain {
public static final Module MODULES =
Modules
.override(Modules.combine(ALL_BUT_JMX_CASSANDRA_MODULE))
- .with(new RabbitMQModule(), new BlobStoreChoosingModule());
+ .with(new RabbitMQModule(), new BlobStoreChoosingModule(), new RabbitMQEventBusModule());
public static void main(String[] args) throws Exception {
Configuration configuration = Configuration.builder()
http://git-wip-us.apache.org/repos/asf/james-project/blob/255f1be5/server/container/guice/cassandra-rabbitmq-guice/src/main/java/org/apache/james/modules/event/RabbitMQEventBusModule.java
----------------------------------------------------------------------
diff --git a/server/container/guice/cassandra-rabbitmq-guice/src/main/java/org/apache/james/modules/event/RabbitMQEventBusModule.java b/server/container/guice/cassandra-rabbitmq-guice/src/main/java/org/apache/james/modules/event/RabbitMQEventBusModule.java
new file mode 100644
index 0000000..df4927e
--- /dev/null
+++ b/server/container/guice/cassandra-rabbitmq-guice/src/main/java/org/apache/james/modules/event/RabbitMQEventBusModule.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.modules.event;
+
+import java.util.List;
+
+import org.apache.james.event.json.EventSerializer;
+import org.apache.james.lifecycle.api.Configurable;
+import org.apache.james.mailbox.events.EventBus;
+import org.apache.james.mailbox.events.MailboxIdRegistrationKey;
+import org.apache.james.mailbox.events.RabbitMQEventBus;
+import org.apache.james.mailbox.events.RegistrationKey;
+import org.apache.james.mailbox.events.RetryBackoffConfiguration;
+import org.apache.james.utils.ConfigurationPerformer;
+import org.parboiled.common.ImmutableList;
+
+import com.google.inject.AbstractModule;
+import com.google.inject.Inject;
+import com.google.inject.Scopes;
+import com.google.inject.multibindings.Multibinder;
+
+public class RabbitMQEventBusModule extends AbstractModule {
+
+ @Override
+ protected void configure() {
+ bind(EventSerializer.class).in(Scopes.SINGLETON);
+
+ bind(RabbitMQEventBus.class).in(Scopes.SINGLETON);
+ bind(EventBus.class).to(RabbitMQEventBus.class);
+
+ Multibinder.newSetBinder(binder(), RegistrationKey.Factory.class)
+ .addBinding().to(MailboxIdRegistrationKey.Factory.class);
+
+ Multibinder.newSetBinder(binder(), ConfigurationPerformer.class)
+ .addBinding().to(RabbitMQEventBusInitializer.class);
+
+ bind(RetryBackoffConfiguration.class).toInstance(RetryBackoffConfiguration.DEFAULT);
+ }
+
+ static class RabbitMQEventBusInitializer implements ConfigurationPerformer {
+ private final RabbitMQEventBus rabbitMQEventBus;
+
+ @Inject
+ RabbitMQEventBusInitializer(RabbitMQEventBus rabbitMQEventBus) {
+ this.rabbitMQEventBus = rabbitMQEventBus;
+ }
+
+ @Override
+ public void initModule() {
+ rabbitMQEventBus.start();
+ }
+
+ @Override
+ public List<Class<? extends Configurable>> forClasses() {
+ return ImmutableList.of();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/james-project/blob/255f1be5/server/container/guice/cassandra-rabbitmq-guice/src/test/java/org/apache/james/CassandraRabbitMQSwiftJmapTestRule.java
----------------------------------------------------------------------
diff --git a/server/container/guice/cassandra-rabbitmq-guice/src/test/java/org/apache/james/CassandraRabbitMQSwiftJmapTestRule.java b/server/container/guice/cassandra-rabbitmq-guice/src/test/java/org/apache/james/CassandraRabbitMQSwiftJmapTestRule.java
index 1f988ad..8c7d7a1 100644
--- a/server/container/guice/cassandra-rabbitmq-guice/src/test/java/org/apache/james/CassandraRabbitMQSwiftJmapTestRule.java
+++ b/server/container/guice/cassandra-rabbitmq-guice/src/test/java/org/apache/james/CassandraRabbitMQSwiftJmapTestRule.java
@@ -39,6 +39,7 @@ import com.google.inject.Module;
public class CassandraRabbitMQSwiftJmapTestRule implements TestRule {
private static final int LIMIT_TO_10_MESSAGES = 10;
+ public static final int TWO_SECONDS = 2000;
private final TemporaryFolder temporaryFolder;
public static CassandraRabbitMQSwiftJmapTestRule defaultTestRule() {
@@ -80,6 +81,16 @@ public class CassandraRabbitMQSwiftJmapTestRule implements TestRule {
}
public void await() {
+ awaitProcessingStart();
guiceModuleTestRule.await();
}
+
+ private void awaitProcessingStart() {
+ // As the RabbitMQEventBus is asynchronous we have otherwise no guaranties that the processing to be awaiting for did start
+ try {
+ Thread.sleep(TWO_SECONDS);
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/james-project/blob/255f1be5/server/container/guice/mailbox/src/main/java/org/apache/james/modules/mailbox/DefaultEventModule.java
----------------------------------------------------------------------
diff --git a/server/container/guice/mailbox/src/main/java/org/apache/james/modules/mailbox/DefaultEventModule.java b/server/container/guice/mailbox/src/main/java/org/apache/james/modules/mailbox/DefaultEventModule.java
index 2cd7252..45dc122 100644
--- a/server/container/guice/mailbox/src/main/java/org/apache/james/modules/mailbox/DefaultEventModule.java
+++ b/server/container/guice/mailbox/src/main/java/org/apache/james/modules/mailbox/DefaultEventModule.java
@@ -27,7 +27,9 @@ import org.apache.commons.configuration.ConfigurationException;
import org.apache.james.lifecycle.api.Configurable;
import org.apache.james.mailbox.MailboxListener;
import org.apache.james.mailbox.events.EventBus;
+import org.apache.james.mailbox.events.EventDeadLetters;
import org.apache.james.mailbox.events.InVMEventBus;
+import org.apache.james.mailbox.events.MemoryEventDeadLetters;
import org.apache.james.mailbox.events.RetryBackoffConfiguration;
import org.apache.james.mailbox.events.delivery.EventDelivery;
import org.apache.james.mailbox.events.delivery.InVmEventDelivery;
@@ -49,6 +51,8 @@ public class DefaultEventModule extends AbstractModule {
bind(MailboxListenersLoaderImpl.class).in(Scopes.SINGLETON);
bind(InVmEventDelivery.class).in(Scopes.SINGLETON);
bind(InVMEventBus.class).in(Scopes.SINGLETON);
+ bind(MemoryEventDeadLetters.class).in(Scopes.SINGLETON);
+ bind(EventDeadLetters.class).to(MemoryEventDeadLetters.class);
bind(MailboxListenersLoader.class).to(MailboxListenersLoaderImpl.class);
bind(EventDelivery.class).to(InVmEventDelivery.class);
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org