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
*/
Subscription nextNode = null;
boolean refTransferDone = false;
while(!refTransferDone){
Subscription node = _lastSubscriptionNode.get();
nextNode = findNextNode(node);
if(_lastSubscriptionNode.compareAndSet(node, nextNode)){
refTransferDone = true;
}
}
// 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.isAcquired() || entry.isDeleted()) && loops != 0)
{
if (nextNode == null)
{
loops--;
nextNode = _subscriptionList.getHead();
}
else
{
// if subscription at end, and active, offer
Subscription sub = nextNode;
deliverToSubscription(sub, entry);
}
nextNode = findNextNode(nextNode);
}