fix remaining event bus usage

Signed-off-by: Abhishek Kumar <abhishek.mrt22@gmail.com>
apache · yadvr · Jun 10, 2024 · Jan 16, 2024 · Feb 12, 2024 · Feb 13, 2024
commit d4bd0865cf13f37252400327b9fad0edd7d592c5
diff --git a/engine/components-api/src/main/java/com/cloud/event/UsageEventUtils.java b/engine/components-api/src/main/java/com/cloud/event/UsageEventUtils.java
@@ -25,14 +25,13 @@
 import javax.annotation.PostConstruct;
 import javax.inject.Inject;
 
-import org.apache.commons.collections.MapUtils;
-import org.apache.log4j.Logger;
-import org.springframework.beans.factory.NoSuchBeanDefinitionException;
-
 import org.apache.cloudstack.framework.config.dao.ConfigurationDao;
 import org.apache.cloudstack.framework.events.Event;
 import org.apache.cloudstack.framework.events.EventBus;
-import org.apache.cloudstack.framework.events.EventBusException;
+import org.apache.cloudstack.framework.events.EventDistributor;
+import org.apache.commons.collections.MapUtils;
+import org.apache.log4j.Logger;
+import org.springframework.beans.factory.NoSuchBeanDefinitionException;
 
 import com.cloud.dc.DataCenterVO;
 import com.cloud.dc.dao.DataCenterDao;
@@ -49,6 +48,7 @@ public class UsageEventUtils {
     private static final Logger s_logger = Logger.getLogger(UsageEventUtils.class);
     protected static EventBus s_eventBus = null;
     protected static ConfigurationDao s_configDao;
+    private static EventDistributor eventDistributor;
 
     @Inject
     UsageEventDao usageEventDao;
@@ -206,9 +206,9 @@ private static void publishUsageEvent(String usageEventType, Long accountId, Lon
         if( !configValue)
             return;
         try {
-            s_eventBus = ComponentContext.getComponent(EventBus.class);
+            eventDistributor = ComponentContext.getComponent(EventDistributor.class);
         } catch (NoSuchBeanDefinitionException nbe) {
-            return; // no provider is configured to provide events bus, so just return
+            return; // no provider is configured to provide events distributor, so just return
         }
 
         Account account = s_accountDao.findById(accountId);
@@ -237,11 +237,7 @@ private static void publishUsageEvent(String usageEventType, Long accountId, Lon
 
         event.setDescription(eventDescription);
 
-        try {
-            s_eventBus.publish(event);
-        } catch (EventBusException e) {
-            s_logger.warn("Failed to publish usage event on the event bus.");
-        }
+        eventDistributor.publish(event);
     }
 
     static final String Name = "management-server";

diff --git a/server/src/main/java/com/cloud/event/AlertGenerator.java b/server/src/main/java/com/cloud/event/AlertGenerator.java
@@ -20,19 +20,19 @@
 import java.text.SimpleDateFormat;
 import java.util.Date;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 
 import javax.annotation.PostConstruct;
 import javax.inject.Inject;
 
+import org.apache.cloudstack.framework.config.dao.ConfigurationDao;
+import org.apache.cloudstack.framework.events.EventBusException;
+import org.apache.cloudstack.framework.events.EventDistributor;
 import org.apache.log4j.Logger;
 import org.springframework.beans.factory.NoSuchBeanDefinitionException;
 import org.springframework.stereotype.Component;
 
-import org.apache.cloudstack.framework.config.dao.ConfigurationDao;
-import org.apache.cloudstack.framework.events.EventBus;
-import org.apache.cloudstack.framework.events.EventBusException;
-
 import com.cloud.configuration.Config;
 import com.cloud.dc.DataCenterVO;
 import com.cloud.dc.HostPodVO;
@@ -47,8 +47,8 @@ public class AlertGenerator {
     private static final Logger s_logger = Logger.getLogger(AlertGenerator.class);
     private static DataCenterDao s_dcDao;
     private static HostPodDao s_podDao;
-    protected static EventBus s_eventBus = null;
     protected static ConfigurationDao s_configDao;
+    protected static EventDistributor eventDistributor;
 
     @Inject
     DataCenterDao dcDao;
@@ -75,9 +75,9 @@ public static void publishAlertOnEventBus(String alertType, long dataCenterId, L
         if(!configValue)
             return;
         try {
-            s_eventBus = ComponentContext.getComponent(EventBus.class);
+            eventDistributor = ComponentContext.getComponent(EventDistributor.class);
         } catch (NoSuchBeanDefinitionException nbe) {
-            return; // no provider is configured to provide events bus, so just return
+            return; // no provider is configured to provide events distributor, so just return
         }
 
         org.apache.cloudstack.framework.events.Event event =
@@ -106,10 +106,11 @@ public static void publishAlertOnEventBus(String alertType, long dataCenterId, L
 
         event.setDescription(eventDescription);
 
-        try {
-            s_eventBus.publish(event);
-        } catch (EventBusException e) {
-            s_logger.warn("Failed to publish alert on the event bus.");
+
+        List<EventBusException> exceptions = eventDistributor.publish(event);
+        for (EventBusException ex : exceptions) {
+            String errMsg = "Failed to publish event.";
+            s_logger.warn(errMsg, ex);
         }
     }
 }
diff --git a/server/src/main/java/com/cloud/vm/UserVmStateListener.java b/server/src/main/java/com/cloud/vm/UserVmStateListener.java
@@ -43,6 +43,7 @@
 
 import org.apache.cloudstack.framework.config.dao.ConfigurationDao;
 import org.apache.cloudstack.framework.events.EventDistributor;
+import org.springframework.beans.factory.NoSuchBeanDefinitionException;
 
 public class UserVmStateListener implements StateListener<State, VirtualMachine.Event, VirtualMachine> {
 
@@ -66,10 +67,6 @@ public UserVmStateListener(UsageEventDao usageEventDao, NetworkDao networkDao, N
         this._configDao = configDao;
     }
 
-    public void setEventDistributor(EventDistributor eventDistributor) {
-        this.eventDistributor = eventDistributor;
-    }
-
     @Override
     public boolean preStateTransitionEvent(State oldState, Event event, State newState, VirtualMachine vo, boolean status, Object opaque) {
         pubishOnEventBus(event.name(), "preStateTransitionEvent", vo, oldState, newState);
@@ -127,8 +124,10 @@ private void pubishOnEventBus(String event, String status, VirtualMachine vo, Vi
         boolean configValue = Boolean.parseBoolean(value);
         if(!configValue)
             return;
-        if (eventDistributor == null) {
-            setEventDistributor(ComponentContext.getComponent(EventDistributor.class));
+        try {
+            eventDistributor = ComponentContext.getComponent(EventDistributor.class);
+        } catch (NoSuchBeanDefinitionException nbe) {
+            return; // no provider is configured to provide events distributor, so just return
         }
 
         String resourceName = getEntityFromClassName(VirtualMachine.class.getName());

diff --git a/server/src/test/java/com/cloud/event/ActionEventUtilsTest.java b/server/src/test/java/com/cloud/event/ActionEventUtilsTest.java
@@ -29,7 +29,7 @@
 import org.apache.cloudstack.context.CallContext;
 import org.apache.cloudstack.framework.config.dao.ConfigurationDao;
 import org.apache.cloudstack.framework.events.Event;
-import org.apache.cloudstack.framework.events.EventBus;
+import org.apache.cloudstack.framework.events.EventDistributor;
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
@@ -97,7 +97,7 @@ public class ActionEventUtilsTest {
     protected ConfigurationDao configDao;
 
     @Mock
-    protected EventBus eventBus;
+    protected EventDistributor eventDistributor;
 
     private AccountVO account;
     private UserVO user;
@@ -149,7 +149,7 @@ public void setupCommonMocks() throws Exception {
         //Some basic mocks.
         Mockito.when(configDao.getValue(Config.PublishActionEvent.key())).thenReturn("true");
         componentContextMocked = Mockito.mockStatic(ComponentContext.class);
-        componentContextMocked.when(() -> ComponentContext.getComponent(EventBus.class)).thenReturn(eventBus);
+        componentContextMocked.when(() -> ComponentContext.getComponent(EventDistributor.class)).thenReturn(eventDistributor);
 
         //Needed for persist to actually set an ID that can be returned from the ActionEventUtils
         //methods.
@@ -166,14 +166,11 @@ public EventVO answer(InvocationOnMock invocation) throws Throwable {
         });
 
         //Needed to record events published on the bus.
-        Mockito.doAnswer(new Answer<Void>() {
-            @Override public Void answer(InvocationOnMock invocation) throws Throwable {
-                Event event = (Event)invocation.getArguments()[0];
-                publishedEvents.add(event);
-                return null;
-            }
-
-        }).when(eventBus).publish(Mockito.any(Event.class));
+        Mockito.doAnswer((Answer<Void>) invocation -> {
+            Event event = (Event)invocation.getArguments()[0];
+            publishedEvents.add(event);
+            return null;
+        }).when(eventDistributor).publish(Mockito.any(Event.class));
 
         account = new AccountVO("testaccount", 1L, "networkdomain", Account.Type.NORMAL, "uuid");
         account.setId(ACCOUNT_ID);

diff --git a/server/src/test/java/com/cloud/template/HypervisorTemplateAdapterTest.java b/server/src/test/java/com/cloud/template/HypervisorTemplateAdapterTest.java
@@ -18,26 +18,26 @@
 
 package com.cloud.template;
 
-import com.cloud.dc.DataCenterVO;
-import com.cloud.dc.dao.DataCenterDao;
-import com.cloud.event.EventTypes;
-import com.cloud.event.UsageEventUtils;
-import com.cloud.event.UsageEventVO;
-import com.cloud.event.dao.UsageEventDao;
-import com.cloud.exception.InvalidParameterValueException;
-import com.cloud.org.Grouping;
-import com.cloud.server.StatsCollector;
-import com.cloud.storage.Storage.ImageFormat;
-import com.cloud.storage.TemplateProfile;
-import com.cloud.storage.VMTemplateStorageResourceAssoc.Status;
-import com.cloud.storage.VMTemplateVO;
-import com.cloud.storage.dao.VMTemplateZoneDao;
-import com.cloud.test.TestAppender;
-import com.cloud.user.AccountVO;
-import com.cloud.user.ResourceLimitService;
-import com.cloud.user.dao.AccountDao;
-import com.cloud.utils.component.ComponentContext;
-import com.cloud.utils.exception.CloudRuntimeException;
+import static org.mockito.Mockito.any;
+import static org.mockito.Mockito.anyLong;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.lang.reflect.Field;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.regex.Pattern;
+
 import org.apache.cloudstack.engine.subsystem.api.storage.DataStore;
 import org.apache.cloudstack.engine.subsystem.api.storage.DataStoreManager;
 import org.apache.cloudstack.engine.subsystem.api.storage.TemplateDataFactory;
@@ -47,8 +47,8 @@
 import org.apache.cloudstack.framework.async.AsyncCallFuture;
 import org.apache.cloudstack.framework.config.dao.ConfigurationDao;
 import org.apache.cloudstack.framework.events.Event;
-import org.apache.cloudstack.framework.events.EventBus;
 import org.apache.cloudstack.framework.events.EventBusException;
+import org.apache.cloudstack.framework.events.EventDistributor;
 import org.apache.cloudstack.framework.messagebus.MessageBus;
 import org.apache.cloudstack.secstorage.heuristics.HeuristicType;
 import org.apache.cloudstack.storage.datastore.db.TemplateDataStoreDao;
@@ -71,30 +71,31 @@
 import org.mockito.junit.MockitoJUnitRunner;
 import org.mockito.stubbing.Answer;
 
-import java.lang.reflect.Field;
-import java.lang.reflect.InvocationTargetException;
-import java.lang.reflect.Method;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.ExecutionException;
-import java.util.regex.Pattern;
-
-import static org.mockito.Mockito.any;
-import static org.mockito.Mockito.anyLong;
-import static org.mockito.Mockito.doAnswer;
-import static org.mockito.Mockito.eq;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
+import com.cloud.dc.DataCenterVO;
+import com.cloud.dc.dao.DataCenterDao;
+import com.cloud.event.EventTypes;
+import com.cloud.event.UsageEventUtils;
+import com.cloud.event.UsageEventVO;
+import com.cloud.event.dao.UsageEventDao;
+import com.cloud.exception.InvalidParameterValueException;
+import com.cloud.org.Grouping;
+import com.cloud.server.StatsCollector;
+import com.cloud.storage.Storage.ImageFormat;
+import com.cloud.storage.TemplateProfile;
+import com.cloud.storage.VMTemplateStorageResourceAssoc.Status;
+import com.cloud.storage.VMTemplateVO;
+import com.cloud.storage.dao.VMTemplateZoneDao;
+import com.cloud.test.TestAppender;
+import com.cloud.user.AccountVO;
+import com.cloud.user.ResourceLimitService;
+import com.cloud.user.dao.AccountDao;
+import com.cloud.utils.component.ComponentContext;
+import com.cloud.utils.exception.CloudRuntimeException;
 
 @RunWith(MockitoJUnitRunner.class)
 public class HypervisorTemplateAdapterTest {
     @Mock
-    EventBus _bus;
+    EventDistributor eventDistributor;
     List<Event> events = new ArrayList<>();
 
     @Mock
@@ -176,16 +177,14 @@ public UsageEventUtils setupUsageUtils() throws EventBusException {
 
         Mockito.when(_usageEventDao.listAll()).thenReturn(usageEvents);
 
-        doAnswer(new Answer<Void>() {
-            @Override public Void answer(InvocationOnMock invocation) throws Throwable {
-                Event event = (Event)invocation.getArguments()[0];
-                events.add(event);
-                return null;
-            }
-        }).when(_bus).publish(any(Event.class));
+        doAnswer((Answer<Void>) invocation -> {
+            Event event = (Event)invocation.getArguments()[0];
+            events.add(event);
+            return null;
+        }).when(eventDistributor).publish(any(Event.class));
 
         componentContextMocked = Mockito.mockStatic(ComponentContext.class);
-        when(ComponentContext.getComponent(eq(EventBus.class))).thenReturn(_bus);
+        when(ComponentContext.getComponent(eq(EventDistributor.class))).thenReturn(eventDistributor);
 
         UsageEventUtils utils = new UsageEventUtils();