incrementQueueSize(message);
_totalMessagesReceived.incrementAndGet();
QueueEntry entry;
Subscription exclusiveSub = _exclusiveSubscriber;
if (exclusiveSub != null)
{
exclusiveSub.getSendLock();
try
{
entry = _entries.add(message);
deliverToSubscription(exclusiveSub, entry);
}
finally
{
exclusiveSub.releaseSendLock();
}
}
else
{
entry = _entries.add(message);
/*
iterate over subscriptions and if any is at the end of the queue and can deliver this message, then deliver the message
*/
SubscriptionList.SubscriptionNode node = _lastSubscriptionNode.get();
SubscriptionList.SubscriptionNode nextNode = node.getNext();
if (nextNode == null)
{
nextNode = _subscriptionList.getHead().getNext();
}
while (nextNode != null)
{
if (_lastSubscriptionNode.compareAndSet(node, nextNode))
{
break;
}
else
{
node = _lastSubscriptionNode.get();
nextNode = node.getNext();
if (nextNode == null)
{
nextNode = _subscriptionList.getHead().getNext();
}
}
}
// always do one extra loop after we believe we've finished
// this catches the case where we *just* miss an update
int loops = 2;
while (entry.isAvailable() && loops != 0)
{
if (nextNode == null)
{
loops--;
nextNode = _subscriptionList.getHead();
}
else
{
// if subscription at end, and active, offer
Subscription sub = nextNode.getSubscription();
deliverToSubscription(sub, entry);
}
nextNode = nextNode.getNext();
}