diff --git a/engine/orchestration/src/main/java/com/cloud/agent/manager/DirectAgentAttache.java b/engine/orchestration/src/main/java/com/cloud/agent/manager/DirectAgentAttache.java index 906a06da7ddc..7abfd0eaddc8 100644 --- a/engine/orchestration/src/main/java/com/cloud/agent/manager/DirectAgentAttache.java +++ b/engine/orchestration/src/main/java/com/cloud/agent/manager/DirectAgentAttache.java @@ -22,9 +22,11 @@ import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; import org.apache.cloudstack.framework.config.ConfigKey; import org.apache.cloudstack.managed.context.ManagedContextRunnable; +import org.apache.logging.log4j.ThreadContext; import com.cloud.agent.api.Answer; import com.cloud.agent.api.Command; @@ -36,7 +38,6 @@ import com.cloud.exception.AgentUnavailableException; import com.cloud.host.Status; import com.cloud.resource.ServerResource; -import org.apache.logging.log4j.ThreadContext; public class DirectAgentAttache extends AgentAttache { @@ -44,12 +45,19 @@ public class DirectAgentAttache extends AgentAttache { "Number of times retrying a host ping while waiting for check results", true); protected final ConfigKey _HostPingRetryTimer = new ConfigKey("Advanced", Integer.class, "host.ping.retry.timer", "5", "Interval to wait before retrying a host ping while waiting for check results", true); - ServerResource _resource; - List> _futures = new ArrayList>(); - long _seq = 0; - LinkedList tasks = new LinkedList(); - AtomicInteger _outstandingTaskCount; - AtomicInteger _outstandingCronTaskCount; + // volatile so that isClosed() and PingTask/CronTask can read it without a lock. + // All writes go through _futuresLock to stay atomic with futures management. + private volatile ServerResource _resource; + private final List> _futures = new ArrayList>(); + // Separate lock for futures and _resource state. We intentionally do NOT use + // synchronized(this) here because disconnect() calls resource.disconnected() which + // can be slow (hypervisor roundtrip). A dedicated lock keeps that slow call outside + // the critical section so process() and send() are not blocked by it. + private final Object _futuresLock = new Object(); + private final AtomicLong _seq = new AtomicLong(0); + private final LinkedList tasks = new LinkedList(); + private final AtomicInteger _outstandingTaskCount; + private final AtomicInteger _outstandingCronTaskCount; public DirectAgentAttache(AgentManagerImpl agentMgr, long id, String uuid,String name, ServerResource resource, boolean maintenance) { super(agentMgr, id, uuid, name, maintenance); @@ -62,15 +70,21 @@ public DirectAgentAttache(AgentManagerImpl agentMgr, long id, String uuid,String public void disconnect(Status state) { logger.debug("Processing disconnect [id: {}, uuid: {}, name: {}]", _id, _uuid, _name); - for (ScheduledFuture future : _futures) { - future.cancel(false); - } - - synchronized (this) { - if (_resource != null) { - _resource.disconnected(); - _resource = null; + // Capture the resource reference and null it out atomically with futures cleanup, + // so that process() and send() cannot sneak in a new scheduled task after we clear. + // We call resource.disconnected() outside the lock intentionally - it can be slow + // (calls into the hypervisor driver), and we don't want to hold _futuresLock during that. + ServerResource resource; + synchronized (_futuresLock) { + for (ScheduledFuture future : _futures) { + future.cancel(false); } + _futures.clear(); + resource = _resource; + _resource = null; + } + if (resource != null) { + resource.disconnected(); } } @@ -83,7 +97,7 @@ public boolean equals(Object obj) { } @Override - public synchronized boolean isClosed() { + public boolean isClosed() { return _resource == null; } @@ -96,7 +110,14 @@ public void send(Request req) throws AgentUnavailableException { if (answers != null && answers[0] instanceof StartupAnswer) { StartupAnswer startup = (StartupAnswer)answers[0]; int interval = startup.getPingInterval(); - _futures.add(_agentMgr.getCronJobPool().scheduleAtFixedRate(new PingTask(), interval, interval, TimeUnit.SECONDS)); + synchronized (_futuresLock) { + if (!isClosed()) { + // scheduleWithFixedDelay - next ping starts only after the previous one + // finishes. scheduleAtFixedRate would pile up concurrent pings if the + // hypervisor is slow, eventually exhausting the cron thread pool. + _futures.add(_agentMgr.getCronJobPool().scheduleWithFixedDelay(new PingTask(), interval, interval, TimeUnit.SECONDS)); + } + } } } else { Command[] cmds = req.getCommands(); @@ -105,7 +126,11 @@ public void send(Request req) throws AgentUnavailableException { scheduleFromQueue(); } else { CronCommand cmd = (CronCommand)cmds[0]; - _futures.add(_agentMgr.getCronJobPool().scheduleAtFixedRate(new CronTask(req), cmd.getInterval(), cmd.getInterval(), TimeUnit.SECONDS)); + synchronized (_futuresLock) { + if (!isClosed()) { + _futures.add(_agentMgr.getCronJobPool().scheduleAtFixedRate(new CronTask(req), cmd.getInterval(), cmd.getInterval(), TimeUnit.SECONDS)); + } + } } } } @@ -115,8 +140,17 @@ public void process(Answer[] answers) { if (answers != null && answers[0] instanceof StartupAnswer) { StartupAnswer startup = (StartupAnswer)answers[0]; int interval = startup.getPingInterval(); - logger.info("StartupAnswer received [id: {}, uuid: {}, name: {}, interval: {}]", startup.getHostId(), startup.getHostUuid(), startup.getHostName(), interval); - _futures.add(_agentMgr.getCronJobPool().scheduleAtFixedRate(new PingTask(), interval, interval, TimeUnit.SECONDS)); + logger.info(String.format( + "StartupAnswer received [id: %d, uuid: %s, name: %s, interval: %d]", + startup.getHostId(), startup.getHostUuid(), startup.getHostName(), interval)); + synchronized (_futuresLock) { + if (!isClosed()) { + // scheduleWithFixedDelay - next ping starts only after the previous one + // finishes. scheduleAtFixedRate would pile up concurrent pings if the + // hypervisor is slow, eventually exhausting the cron thread pool. + _futures.add(_agentMgr.getCronJobPool().scheduleWithFixedDelay(new PingTask(), interval, interval, TimeUnit.SECONDS)); + } + } } } @@ -124,11 +158,9 @@ public void process(Answer[] answers) { protected void finalize() throws Throwable { try { assert _resource == null : "Come on now....If you're going to dabble in agent code, you better know how to close out our resources. Ever considered why there's a method called disconnect()?"; - synchronized (this) { - if (_resource != null) { - logger.warn("Lost attache for [id: {}, uuid: {}, name: {}]", _id, _uuid, _name); - disconnect(Status.Alert); - } + if (_resource != null) { + logger.warn(String.format("Lost attache for [id: %d, uuid: %s, name: %s]", _id, _uuid, _name)); + disconnect(Status.Alert); } } finally { super.finalize(); @@ -143,8 +175,19 @@ private synchronized void scheduleFromQueue() { logger.trace("Agent attache [id: {}, uuid: {}, name: {}], task queue size={}, outstanding tasks={}", _id, _uuid, _name, tasks.size(), _outstandingTaskCount.get()); while (!tasks.isEmpty() && _outstandingTaskCount.get() < _agentMgr.getDirectAgentThreadCap()) { + Task task = tasks.removeFirst(); _outstandingTaskCount.incrementAndGet(); - _agentMgr.getDirectAgentPool().execute(tasks.remove()); + try { + _agentMgr.getDirectAgentPool().execute(task); + } catch (RuntimeException e) { + // The thread pool rejected the task (likely full or shutting down). + // Roll back: return the slot and put the task back at the head of the queue + // so it gets a chance to run on the next scheduleFromQueue() call. + _outstandingTaskCount.decrementAndGet(); + tasks.addFirst(task); + logger.warn("Failed to submit direct agent task, will retry on next schedule", e); + break; + } } } @@ -155,12 +198,12 @@ public int hashCode() { protected class PingTask extends ManagedContextRunnable { @Override - protected synchronized void runInContext() { + protected void runInContext() { try { - if (_outstandingCronTaskCount.incrementAndGet() >= _agentMgr.getDirectAgentThreadCap()) { - logger.warn( - "PingTask execution for direct attache [id: {}, uuid: {}, name: {}] has reached maximum outstanding limit({}), bailing out", - _id, _uuid, _name, _agentMgr.getDirectAgentThreadCap()); + if (_outstandingCronTaskCount.incrementAndGet() > _agentMgr.getDirectAgentThreadCap()) { + logger.warn(String.format( + "PingTask execution for direct attache [id: %d, uuid: %s, name: %s] has reached maximum outstanding limit(%d), bailing out", + _id, _uuid, _name, _agentMgr.getDirectAgentThreadCap())); return; } @@ -183,7 +226,7 @@ protected synchronized void runInContext() { ThreadContext.put("logcontextid", cmd.getContextParam("logid")); } logger.debug("Ping from [id: {}, uuid: {}, name: {}]", _id, _uuid, _name); - long seq = _seq++; + long seq = _seq.getAndIncrement(); logger.trace("SeqA {}-{}: {}", _id, seq, new Request(_id, -1, cmd, false).toString()); @@ -200,8 +243,7 @@ protected synchronized void runInContext() { } protected class CronTask extends ManagedContextRunnable { - Request _req; - + private final Request _req; public CronTask(Request req) { _req = req; } @@ -226,10 +268,10 @@ private void bailout() { protected void runInContext() { long seq = _req.getSequence(); try { - if (_outstandingCronTaskCount.incrementAndGet() >= _agentMgr.getDirectAgentThreadCap()) { - logger.warn( - "CronTask execution for direct attache [id: {}, uuid: {}, name: {}] has reached maximum outstanding limit({}), bailing out", - _id, _uuid, _name, _agentMgr.getDirectAgentThreadCap()); + if (_outstandingCronTaskCount.incrementAndGet() > _agentMgr.getDirectAgentThreadCap()) { + logger.warn(String.format( + "CronTask execution for direct attache [id: %d, uuid: %s, name: %s] has reached maximum outstanding limit(%d), bailing out", + _id, _uuid, _name, _agentMgr.getDirectAgentThreadCap())); bailout(); return; } @@ -282,8 +324,7 @@ protected void runInContext() { } protected class Task extends ManagedContextRunnable { - Request _req; - + private final Request _req; public Task(Request req) { _req = req; } diff --git a/engine/orchestration/src/test/java/com/cloud/agent/manager/DirectAgentAttacheTest.java b/engine/orchestration/src/test/java/com/cloud/agent/manager/DirectAgentAttacheTest.java index 65e31c271a42..d24f26249e4a 100644 --- a/engine/orchestration/src/test/java/com/cloud/agent/manager/DirectAgentAttacheTest.java +++ b/engine/orchestration/src/test/java/com/cloud/agent/manager/DirectAgentAttacheTest.java @@ -16,6 +16,15 @@ // under the License. package com.cloud.agent.manager; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.eq; + +import java.util.UUID; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; + import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; @@ -24,10 +33,11 @@ import org.mockito.MockitoAnnotations; import org.mockito.junit.MockitoJUnitRunner; +import com.cloud.agent.api.Answer; +import com.cloud.agent.api.StartupAnswer; +import com.cloud.host.Status; import com.cloud.resource.ServerResource; -import java.util.UUID; - @RunWith(MockitoJUnitRunner.class) public class DirectAgentAttacheTest { @Mock @@ -36,6 +46,12 @@ public class DirectAgentAttacheTest { @Mock private ServerResource _resource; + @Mock + private ScheduledExecutorService _cronJobPool; + + @Mock + private ScheduledFuture _future; + long _id = 0L; String _uuid = UUID.randomUUID().toString(); @@ -55,4 +71,31 @@ public void testPingTask() throws Exception { pt.runInContext(); Mockito.verify(_resource, Mockito.times(1)).getCurrentStatus(_id); } + + @Test + public void testProcessSchedulesPingWhenConnected() { + Mockito.doReturn(_cronJobPool).when(_agentMgr).getCronJobPool(); + Mockito.doReturn(_future).when(_cronJobPool).scheduleWithFixedDelay(any(Runnable.class), anyLong(), anyLong(), any(TimeUnit.class)); + + directAgentAttache.process(new Answer[] {buildStartupAnswer()}); + + Mockito.verify(_cronJobPool, Mockito.times(1)).scheduleWithFixedDelay(any(Runnable.class), eq(60L), eq(60L), eq(TimeUnit.SECONDS)); + } + + @Test + public void testProcessDoesNotSchedulePingAfterDisconnect() { + // Once disconnect() has cleared the resource, a late StartupAnswer must not schedule + // a PingTask - otherwise the future would never be cancelled (the leak this fix targets). + directAgentAttache.disconnect(Status.Disconnected); + + directAgentAttache.process(new Answer[] {buildStartupAnswer()}); + + Mockito.verify(_cronJobPool, Mockito.never()).scheduleWithFixedDelay(any(Runnable.class), anyLong(), anyLong(), any(TimeUnit.class)); + } + + private StartupAnswer buildStartupAnswer() { + StartupAnswer startup = Mockito.mock(StartupAnswer.class); + Mockito.doReturn(60).when(startup).getPingInterval(); + return startup; + } } diff --git a/framework/ipc/src/main/java/org/apache/cloudstack/framework/messagebus/MessageBusBase.java b/framework/ipc/src/main/java/org/apache/cloudstack/framework/messagebus/MessageBusBase.java index 742fd90c33e0..eed368139e50 100644 --- a/framework/ipc/src/main/java/org/apache/cloudstack/framework/messagebus/MessageBusBase.java +++ b/framework/ipc/src/main/java/org/apache/cloudstack/framework/messagebus/MessageBusBase.java @@ -21,23 +21,45 @@ import java.util.ArrayList; import java.util.Arrays; +import java.util.Collection; import java.util.Collections; import java.util.HashMap; +import java.util.LinkedHashSet; import java.util.List; import java.util.Map; - -import org.apache.logging.log4j.Logger; -import org.apache.logging.log4j.LogManager; +import java.util.Set; +import java.util.concurrent.locks.ReadWriteLock; +import java.util.concurrent.locks.ReentrantReadWriteLock; import org.apache.cloudstack.framework.serializer.MessageSerializer; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; import com.cloud.utils.db.TransactionLegacy; import com.cloud.utils.exception.CloudRuntimeException; +/** + * MessageBus implementation based on a hierarchical topic tree. + * + * Locking model: + * Multiple publishers can run concurrently - publish() takes a read lock, + * which only blocks when someone is modifying the subscriber tree (subscribe, + * unsubscribe, clearAll, prune). This means a pod restart with hundreds of + * hosts reconnecting and firing events simultaneously will not serialize + * through a single bottleneck. + * + * Subscriber callbacks are intentionally called OUTSIDE the read lock. + * Holding a lock during external callbacks is a classic source of production + * outages: a slow subscriber (DB call, GC pause, hypervisor roundtrip) would + * block ALL other publishers and eventually starve write lock holders + * (subscribe/unsubscribe). Instead we snapshot the subscriber list under + * the lock and release it before calling anyone. + */ public class MessageBusBase implements MessageBus { - private final Gate _gate; - private final List _pendingActions; + // Fair mode: a queued writer (subscribe/unsubscribe/prune/clearAll) is served ahead of + // newly arriving readers, so a steady stream of publish() read locks cannot starve writers. + private final ReadWriteLock _lock = new ReentrantReadWriteLock(true); private final SubscriptionNode _subscriberRoot; private MessageSerializer _messageSerializer; @@ -45,9 +67,6 @@ public class MessageBusBase implements MessageBus { protected Logger logger = LogManager.getLogger(getClass()); public MessageBusBase() { - _gate = new Gate(); - _pendingActions = new ArrayList(); - _subscriberRoot = new SubscriptionNode(null, "/", null); } @@ -65,82 +84,54 @@ public MessageSerializer getMessageSerializer() { public void subscribe(String subject, MessageSubscriber subscriber) { assert (subject != null); assert (subscriber != null); - if (_gate.enter()) { - if (logger.isTraceEnabled()) { - logger.trace("Enter gate in message bus subscribe"); - } - try { - SubscriptionNode current = locate(subject, null, true); - assert (current != null); - current.addSubscriber(subscriber); - } finally { - _gate.leave(); - } - } else { - synchronized (_pendingActions) { - _pendingActions.add(new ActionRecord(ActionType.Subscribe, subject, subscriber)); - } + _lock.writeLock().lock(); + try { + logger.trace("Acquired write lock in message bus subscribe"); + SubscriptionNode current = locate(subject, null, true); + assert (current != null); + current.addSubscriber(subscriber); + } finally { + _lock.writeLock().unlock(); } } @Override public void unsubscribe(String subject, MessageSubscriber subscriber) { - if (_gate.enter()) { - if (logger.isTraceEnabled()) { - logger.trace("Enter gate in message bus unsubscribe"); - } - try { - if (subject != null) { - SubscriptionNode current = locate(subject, null, false); - if (current != null) - current.removeSubscriber(subscriber, false); - } else { - _subscriberRoot.removeSubscriber(subscriber, true); - } - } finally { - _gate.leave(); - } - } else { - synchronized (_pendingActions) { - _pendingActions.add(new ActionRecord(ActionType.Unsubscribe, subject, subscriber)); + _lock.writeLock().lock(); + try { + logger.trace("Acquired write lock in message bus unsubscribe"); + if (subject != null) { + SubscriptionNode current = locate(subject, null, false); + if (current != null) + current.removeSubscriber(subscriber, false); + } else { + _subscriberRoot.removeSubscriber(subscriber, true); } + } finally { + _lock.writeLock().unlock(); } } @Override public void clearAll() { - if (_gate.enter()) { - if (logger.isTraceEnabled()) { - logger.trace("Enter gate in message bus clearAll"); - } - try { - _subscriberRoot.clearAll(); - doPrune(); - } finally { - _gate.leave(); - } - } else { - synchronized (_pendingActions) { - _pendingActions.add(new ActionRecord(ActionType.ClearAll, null, null)); - } + _lock.writeLock().lock(); + try { + logger.trace("Acquired write lock in message bus clearAll"); + _subscriberRoot.clearAll(); + doPrune(); + } finally { + _lock.writeLock().unlock(); } } @Override public void prune() { - if (_gate.enter()) { - if (logger.isTraceEnabled()) { - logger.trace("Enter gate in message bus prune"); - } - try { - doPrune(); - } finally { - _gate.leave(); - } - } else { - synchronized (_pendingActions) { - _pendingActions.add(new ActionRecord(ActionType.Prune, null, null)); - } + _lock.writeLock().lock(); + try { + logger.trace("Acquired write lock in message bus prune"); + doPrune(); + } finally { + _lock.writeLock().unlock(); } } @@ -163,66 +154,36 @@ private void doPrune() { @Override public void publish(String senderAddress, String subject, PublishScope scope, Object args) { // publish cannot be in DB transaction, which may hold DB lock too long, and we are guarding this here - if (!noDbTxn()){ + if (!noDbTxn()) { String errMsg = "NO EVENT PUBLISH CAN BE WRAPPED WITHIN DB TRANSACTION!"; logger.error(errMsg, new CloudRuntimeException(errMsg)); } - if (_gate.enter(true)) { - if (logger.isTraceEnabled()) { - logger.trace("Enter gate in message bus publish"); - } - try { - List chainFromTop = new ArrayList(); - SubscriptionNode current = locate(subject, chainFromTop, false); - - if (current != null) - current.notifySubscribers(senderAddress, subject, args); - - Collections.reverse(chainFromTop); - for (SubscriptionNode node : chainFromTop) - node.notifySubscribers(senderAddress, subject, args); - } finally { - _gate.leave(); - } + // Collect subscribers under read lock (fast - just tree traversal and list copy), + // then release the lock before calling any callbacks. + // LinkedHashSet deduplicates: a subscriber registered on both "Host" and "Host.123" + // gets exactly one callback when "Host.123" is published. + Set toNotify = new LinkedHashSet<>(); + _lock.readLock().lock(); + try { + logger.trace("Acquired read lock in message bus publish"); + List chainFromTop = new ArrayList<>(); + SubscriptionNode current = locate(subject, chainFromTop, false); + + if (current != null) + current.collectSubscribers(toNotify); + + Collections.reverse(chainFromTop); + for (SubscriptionNode node : chainFromTop) + node.collectSubscribers(toNotify); + } finally { + _lock.readLock().unlock(); } - } - - private void onGateOpen() { - synchronized (_pendingActions) { - ActionRecord record = null; - while (_pendingActions.size() > 0) { - record = _pendingActions.remove(0); - switch (record.getType()) { - case Subscribe: { - SubscriptionNode current = locate(record.getSubject(), null, true); - assert (current != null); - current.addSubscriber(record.getSubscriber()); - } - break; - - case Unsubscribe: - if (record.getSubject() != null) { - SubscriptionNode current = locate(record.getSubject(), null, false); - if (current != null) - current.removeSubscriber(record.getSubscriber(), false); - } else { - _subscriberRoot.removeSubscriber(record.getSubscriber(), true); - } - break; - - case ClearAll: - _subscriberRoot.clearAll(); - break; - - case Prune: - doPrune(); - break; - - default: - assert (false); - break; - } + for (MessageSubscriber subscriber : toNotify) { + try { + subscriber.onPublishMessage(senderAddress, subject, args); + } catch (Throwable t) { + logger.error("Subscriber threw an exception during publish of subject: " + subject + " scope: " + scope + " args: " + args, t); } } } @@ -272,95 +233,6 @@ private boolean noDbTxn() { // // Support inner classes // - private static enum ActionType { - Subscribe, Unsubscribe, ClearAll, Prune - } - - private static class ActionRecord { - private final ActionType _type; - private final String _subject; - private final MessageSubscriber _subscriber; - - public ActionRecord(ActionType type, String subject, MessageSubscriber subscriber) { - _type = type; - _subject = subject; - _subscriber = subscriber; - } - - public ActionType getType() { - return _type; - } - - public String getSubject() { - return _subject; - } - - public MessageSubscriber getSubscriber() { - return _subscriber; - } - } - - private class Gate { - private int _reentranceCount; - private Thread _gateOwner; - - public Gate() { - _reentranceCount = 0; - _gateOwner = null; - } - - public boolean enter() { - return enter(false); - } - - public boolean enter(boolean wait) { - while (true) { - synchronized (this) { - if (_reentranceCount == 0) { - assert (_gateOwner == null); - - _reentranceCount++; - _gateOwner = Thread.currentThread(); - return true; - } else { - if (wait) { - try { - wait(); - } catch (InterruptedException e) { - logger.debug("[ignored] interrupted while guarding re-entrance on message bus."); - } - } else { - break; - } - } - } - } - - return false; - } - - public void leave() { - synchronized (this) { - if (_reentranceCount > 0) { - try { - assert (_gateOwner == Thread.currentThread()); - - onGateOpen(); - } finally { - if (logger.isTraceEnabled()) { - logger.trace("Open gate of message bus"); - } - _reentranceCount--; - assert (_reentranceCount == 0); - _gateOwner = null; - - notifyAll(); - } - } - } - } - } - private static class SubscriptionNode { private final String _nodeKey; private final List _subscribers; @@ -437,10 +309,8 @@ public void prune(List trimNodes) { trimNodes.add(this); } - public void notifySubscribers(String senderAddress, String subject, Object args) { - for (MessageSubscriber subscriber : _subscribers) { - subscriber.onPublishMessage(senderAddress, subject, args); - } + public void collectSubscribers(Collection target) { + target.addAll(_subscribers); } public boolean isTrimmable() { diff --git a/framework/ipc/src/test/java/org/apache/cloudstack/framework/messagebus/MessageBusBaseTest.java b/framework/ipc/src/test/java/org/apache/cloudstack/framework/messagebus/MessageBusBaseTest.java new file mode 100644 index 000000000000..c13e813e3f5f --- /dev/null +++ b/framework/ipc/src/test/java/org/apache/cloudstack/framework/messagebus/MessageBusBaseTest.java @@ -0,0 +1,106 @@ +/* + * 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.cloudstack.framework.messagebus; + +import java.util.ArrayList; +import java.util.List; + +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +import com.cloud.utils.db.TransactionLegacy; + +public class MessageBusBaseTest { + + private MessageBusBase bus; + private TransactionLegacy txn; + + @Before + public void setUp() { + bus = new MessageBusBase(); + // publish() calls noDbTxn() which needs a thread-local transaction to exist. + // open() only registers a lazy transaction; it does not open a DB connection. + txn = TransactionLegacy.open("MessageBusBaseTest"); + } + + @After + public void tearDown() { + if (txn != null) { + txn.close(); + } + } + + /** + * A subscriber registered on both an ancestor ("Host") and a descendant ("Host.123") + * must receive exactly one callback when the descendant topic is published, not one per + * matching node. This pins the LinkedHashSet dedup behavior in publish(). + */ + @Test + public void testSubscriberOnAncestorAndDescendantNotifiedOnce() { + CountingSubscriber subscriber = new CountingSubscriber(); + bus.subscribe("Host", subscriber); + bus.subscribe("Host.123", subscriber); + + bus.publish(null, "Host.123", PublishScope.LOCAL, null); + + Assert.assertEquals(1, subscriber.count); + } + + /** + * Distinct subscribers on ancestor and descendant are both notified, and the more specific + * (descendant) subscriber is notified before the ancestor one. + */ + @Test + public void testDescendantNotifiedBeforeAncestor() { + final List order = new ArrayList<>(); + MessageSubscriber ancestor = (sender, subject, args) -> order.add("ancestor"); + MessageSubscriber descendant = (sender, subject, args) -> order.add("descendant"); + bus.subscribe("Host", ancestor); + bus.subscribe("Host.123", descendant); + + bus.publish(null, "Host.123", PublishScope.LOCAL, null); + + Assert.assertEquals(List.of("descendant", "ancestor"), order); + } + + /** + * A subscriber on a sibling topic ("Host.456") must not be notified for "Host.123". + */ + @Test + public void testSiblingSubscriberNotNotified() { + CountingSubscriber sibling = new CountingSubscriber(); + bus.subscribe("Host.456", sibling); + + bus.publish(null, "Host.123", PublishScope.LOCAL, null); + + Assert.assertEquals(0, sibling.count); + } + + private static class CountingSubscriber implements MessageSubscriber { + int count = 0; + + @Override + public void onPublishMessage(String senderAddress, String subject, Object args) { + count++; + } + } +}