Refactor some event bus handlers so that they do not tightly bind to …

…RabbitMQEventBus (#2656)

Co-authored-by: hung phan <hphan@linagora.com>

event-bus/api/src/main/java/org/apache/james/events/EventBus.java

@@ -100,4 +100,13 @@ default Registration register(EventListener.ReactiveGroupEventListener groupList
     default Collection<Group> listRegisteredGroups() {
         return ImmutableList.of();
     }
+
+    default void start() {
+    }
+
+    default void restart() {
+    }
+
+    default void stop() {
+    }
 }

event-bus/distributed/src/main/java/org/apache/james/events/EventBusReconnectionHandler.java

@@ -19,8 +19,6 @@
 
 package org.apache.james.events;
 
-import jakarta.inject.Inject;
-
 import org.apache.james.backends.rabbitmq.SimpleConnectionPool;
 import org.reactivestreams.Publisher;
 
@@ -29,10 +27,9 @@
 import reactor.core.publisher.Mono;
 
 public class EventBusReconnectionHandler implements SimpleConnectionPool.ReconnectionHandler {
-    private final RabbitMQEventBus rabbitMQEventBus;
+    private final EventBus rabbitMQEventBus;
 
-    @Inject
-    public EventBusReconnectionHandler(RabbitMQEventBus rabbitMQEventBus) {
+    public EventBusReconnectionHandler(EventBus rabbitMQEventBus) {
         this.rabbitMQEventBus = rabbitMQEventBus;
     }
 

event-bus/distributed/src/main/java/org/apache/james/events/GroupRegistrationHandler.java

@@ -63,10 +63,6 @@
 class GroupRegistrationHandler {
     private static final Logger LOGGER = LoggerFactory.getLogger(GroupRegistrationHandler.class);
 
-    public static class GroupRegistrationHandlerGroup extends Group {
-
-    }
-
     static final Group GROUP = new GroupRegistrationHandlerGroup();
 
     private final NamingStrategy namingStrategy;

event-bus/distributed/src/main/java/org/apache/james/events/GroupRegistrationHandlerGroup.java

@@ -0,0 +1,23 @@
+/****************************************************************
+ * 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.events;
+
+public class GroupRegistrationHandlerGroup extends Group {
+}

event-bus/distributed/src/main/java/org/apache/james/events/KeyReconnectionHandler.java

@@ -26,8 +26,6 @@
 import static org.apache.james.backends.rabbitmq.Constants.evaluateDurable;
 import static org.apache.james.backends.rabbitmq.Constants.evaluateExclusive;
 
-import jakarta.inject.Inject;
-
 import org.apache.james.backends.rabbitmq.QueueArguments;
 import org.apache.james.backends.rabbitmq.RabbitMQConfiguration;
 import org.apache.james.backends.rabbitmq.SimpleConnectionPool;
@@ -47,7 +45,6 @@ public class KeyReconnectionHandler implements SimpleConnectionPool.Reconnection
     private final EventBusId eventBusId;
     private final RabbitMQConfiguration configuration;
 
-    @Inject
     public KeyReconnectionHandler(NamingStrategy namingStrategy, EventBusId eventBusId, RabbitMQConfiguration configuration) {
         this.namingStrategy = namingStrategy;
         this.eventBusId = eventBusId;

event-bus/distributed/src/main/java/org/apache/james/events/RabbitEventBusConsumerHealthCheck.java

@@ -36,11 +36,11 @@
 public class RabbitEventBusConsumerHealthCheck implements HealthCheck {
     public static final String COMPONENT = "EventbusConsumers";
 
-    private final RabbitMQEventBus eventBus;
+    private final EventBus eventBus;
     private final NamingStrategy namingStrategy;
     private final SimpleConnectionPool connectionPool;
 
-    public RabbitEventBusConsumerHealthCheck(RabbitMQEventBus eventBus, NamingStrategy namingStrategy,
+    public RabbitEventBusConsumerHealthCheck(EventBus eventBus, NamingStrategy namingStrategy,
                                              SimpleConnectionPool connectionPool) {
         this.eventBus = eventBus;
         this.namingStrategy = namingStrategy;
@@ -65,7 +65,7 @@ public Mono<Result> check() {
     private Result check(Channel channel) {
         Stream<Group> groups = Stream.concat(
             eventBus.listRegisteredGroups().stream(),
-            Stream.of(new GroupRegistrationHandler.GroupRegistrationHandlerGroup()));
+            Stream.of(new GroupRegistrationHandlerGroup()));
 
         Optional<String> queueWithoutConsumers = groups
             .map(namingStrategy::workQueue)

event-bus/distributed/src/main/java/org/apache/james/events/RabbitMQEventBus.java

@@ -83,6 +83,7 @@ public RabbitMQEventBus(NamingStrategy namingStrategy, Sender sender, ReceiverPr
         this.isStopping = false;
     }
 
+    @Override
     public void start() {
         if (!isRunning && !isStopping) {
 
@@ -97,6 +98,7 @@ public void start() {
         }
     }
 
+    @Override
     public void restart() {
         keyRegistrationHandler.restart();
         groupRegistrationHandler.restart();

server/apps/distributed-app/src/main/java/org/apache/james/CassandraRabbitMQJamesServerMain.java

@@ -51,7 +51,7 @@
 import org.apache.james.modules.data.CassandraUsersRepositoryModule;
 import org.apache.james.modules.data.CassandraVacationModule;
 import org.apache.james.modules.event.JMAPEventBusModule;
-import org.apache.james.modules.event.RabbitMQEventBusModule;
+import org.apache.james.modules.event.MailboxEventBusModule;
 import org.apache.james.modules.eventstore.CassandraEventStoreModule;
 import org.apache.james.modules.mailbox.CassandraDeletedMessageVaultModule;
 import org.apache.james.modules.mailbox.CassandraMailboxModule;
@@ -181,7 +181,7 @@ public class CassandraRabbitMQJamesServerMain implements JamesServerMain {
 
     protected static final Module MODULES = Modules.override(REQUIRE_TASK_MANAGER_MODULE, new DistributedTaskManagerModule())
         .with(new RabbitMQModule(),
-            new RabbitMQEventBusModule(),
+            new MailboxEventBusModule(),
             new DistributedTaskSerializationModule());
 
     public static void main(String[] args) throws Exception {

server/apps/distributed-pop3-app/src/main/java/org/apache/james/DistributedPOP3JamesServerMain.java

@@ -57,7 +57,7 @@
 import org.apache.james.modules.data.CassandraUsersRepositoryModule;
 import org.apache.james.modules.data.CassandraVacationModule;
 import org.apache.james.modules.event.JMAPEventBusModule;
-import org.apache.james.modules.event.RabbitMQEventBusModule;
+import org.apache.james.modules.event.MailboxEventBusModule;
 import org.apache.james.modules.eventstore.CassandraEventStoreModule;
 import org.apache.james.modules.mailbox.CassandraBlobStoreDependenciesModule;
 import org.apache.james.modules.mailbox.CassandraDeletedMessageVaultModule;
@@ -175,7 +175,7 @@ public class DistributedPOP3JamesServerMain implements JamesServerMain {
         .with(new RabbitMQModule(),
             new RabbitMQMailQueueModule(),
             new RabbitMailQueueRoutesModule(),
-            new RabbitMQEventBusModule(),
+            new MailboxEventBusModule(),
             new DistributedTaskSerializationModule());
 
     public static void main(String[] args) throws Exception {

server/apps/postgres-app/src/main/java/org/apache/james/PostgresJamesServerMain.java

@@ -49,7 +49,7 @@
 import org.apache.james.modules.data.PostgresVacationModule;
 import org.apache.james.modules.data.SievePostgresRepositoryModules;
 import org.apache.james.modules.event.JMAPEventBusModule;
-import org.apache.james.modules.event.RabbitMQEventBusModule;
+import org.apache.james.modules.event.MailboxEventBusModule;
 import org.apache.james.modules.events.PostgresDeadLetterModule;
 import org.apache.james.modules.mailbox.DefaultEventModule;
 import org.apache.james.modules.mailbox.PostgresDeletedMessageVaultModule;
@@ -231,7 +231,7 @@ public static List<Module> chooseEventBusModules(PostgresJamesConfiguration conf
                     new ActiveMQQueueModule());
             case RABBITMQ:
                 return List.of(
-                    Modules.override(new DefaultEventModule()).with(new RabbitMQEventBusModule()),
+                    Modules.override(new DefaultEventModule()).with(new MailboxEventBusModule()),
                     new RabbitMQModule(),
                     new RabbitMQMailQueueModule(),
                     new FakeMailQueueViewModule(),

server/container/guice/distributed/src/main/java/org/apache/james/modules/event/RabbitMQEventBusModule.java → server/container/guice/distributed/src/main/java/org/apache/james/modules/event/MailboxEventBusModule.java

@@ -52,7 +52,7 @@
 
 import reactor.rabbitmq.Sender;
 
-public class RabbitMQEventBusModule extends AbstractModule {
+public class MailboxEventBusModule extends AbstractModule {
 
     @Override
     protected void configure() {
@@ -64,10 +64,6 @@ protected void configure() {
         bind(RetryBackoffConfiguration.class).toInstance(RetryBackoffConfiguration.DEFAULT);
         bind(EventBusId.class).toInstance(EventBusId.random());
 
-        Multibinder<SimpleConnectionPool.ReconnectionHandler> reconnectionHandlerMultibinder = Multibinder.newSetBinder(binder(), SimpleConnectionPool.ReconnectionHandler.class);
-        reconnectionHandlerMultibinder.addBinding().to(KeyReconnectionHandler.class);
-        reconnectionHandlerMultibinder.addBinding().to(EventBusReconnectionHandler.class);
-
         Multibinder.newSetBinder(binder(), HealthCheck.class)
             .addBinding().to(RabbitMQMailboxEventBusDeadLetterQueueHealthCheck.class);
     }
@@ -78,6 +74,16 @@ HealthCheck healthCheck(RabbitMQEventBus eventBus, NamingStrategy namingStrategy
         return new RabbitEventBusConsumerHealthCheck(eventBus, namingStrategy, connectionPool);
     }
 
+    @ProvidesIntoSet
+    SimpleConnectionPool.ReconnectionHandler provideReconnectionHandler(RabbitMQEventBus eventBus) {
+        return new EventBusReconnectionHandler(eventBus);
+    }
+
+    @ProvidesIntoSet
+    SimpleConnectionPool.ReconnectionHandler provideReconnectionHandler(NamingStrategy namingStrategy, EventBusId eventBusId, RabbitMQConfiguration configuration) {
+        return new KeyReconnectionHandler(namingStrategy, eventBusId, configuration);
+    }
+
     @ProvidesIntoSet
     InitializationOperation workQueue(RabbitMQEventBus instance) {
         return InitilizationOperationBuilder