Google Git
Sign in
apache / qpid-broker-j / 020d4e7d3bcfca7008ae69e02d49faad7afb50af / . / broker-plugins / amqp-0-8-protocol / src / main / java / org / apache / qpid / server / protocol / v0_8 / AMQChannel.java
blob: ccfb99e7c14ede4e9d4d1ac3b628b207ab3a7233 [file] [log] [blame]
/*
*
* 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.qpid.server.protocol.v0_8;
import static org.apache.qpid.transport.util.Functions.hex;
import java.security.AccessControlContext;
import java.security.AccessControlException;
import java.security.AccessController;
import java.security.PrivilegedAction;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.EnumSet;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import javax.security.auth.Subject;
import com.google.common.util.concurrent.ListenableFuture;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.qpid.AMQConnectionException;
import org.apache.qpid.bytebuffer.QpidByteBuffer;
import org.apache.qpid.common.AMQPFilterTypes;
import org.apache.qpid.exchange.ExchangeDefaults;
import org.apache.qpid.framing.*;
import org.apache.qpid.protocol.AMQConstant;
import org.apache.qpid.server.connection.SessionPrincipal;
import org.apache.qpid.server.consumer.ConsumerImpl;
import org.apache.qpid.server.consumer.ConsumerTarget;
import org.apache.qpid.server.filter.AMQInvalidArgumentException;
import org.apache.qpid.server.filter.ArrivalTimeFilter;
import org.apache.qpid.server.filter.FilterManager;
import org.apache.qpid.server.filter.FilterManagerFactory;
import org.apache.qpid.server.filter.Filterable;
import org.apache.qpid.server.filter.MessageFilter;
import org.apache.qpid.server.flow.FlowCreditManager;
import org.apache.qpid.server.logging.EventLogger;
import org.apache.qpid.server.logging.EventLoggerProvider;
import org.apache.qpid.server.logging.LogMessage;
import org.apache.qpid.server.logging.LogSubject;
import org.apache.qpid.server.logging.messages.ChannelMessages;
import org.apache.qpid.server.logging.messages.ExchangeMessages;
import org.apache.qpid.server.logging.subjects.ChannelLogSubject;
import org.apache.qpid.server.message.InstanceProperties;
import org.apache.qpid.server.message.MessageDestination;
import org.apache.qpid.server.message.MessageInstance;
import org.apache.qpid.server.message.MessageReference;
import org.apache.qpid.server.message.MessageSource;
import org.apache.qpid.server.message.ServerMessage;
import org.apache.qpid.server.model.Broker;
import org.apache.qpid.server.model.ConfigurationChangeListener;
import org.apache.qpid.server.model.ConfiguredObject;
import org.apache.qpid.server.model.Connection;
import org.apache.qpid.server.model.Consumer;
import org.apache.qpid.server.model.Exchange;
import org.apache.qpid.server.model.ExclusivityPolicy;
import org.apache.qpid.server.model.LifetimePolicy;
import org.apache.qpid.server.model.NamedAddressSpace;
import org.apache.qpid.server.model.NoFactoryForTypeException;
import org.apache.qpid.server.model.Queue;
import org.apache.qpid.server.model.Session;
import org.apache.qpid.server.model.State;
import org.apache.qpid.server.model.UnknownConfiguredObjectException;
import org.apache.qpid.server.protocol.AMQSessionModel;
import org.apache.qpid.server.protocol.CapacityChecker;
import org.apache.qpid.server.protocol.ConsumerListener;
import org.apache.qpid.server.protocol.PublishAuthorisationCache;
import org.apache.qpid.server.queue.QueueArgumentsConverter;
import org.apache.qpid.server.security.SecurityToken;
import org.apache.qpid.server.store.MessageHandle;
import org.apache.qpid.server.store.MessageStore;
import org.apache.qpid.server.store.StoredMessage;
import org.apache.qpid.server.store.TransactionLogResource;
import org.apache.qpid.server.txn.AsyncAutoCommitTransaction;
import org.apache.qpid.server.txn.LocalTransaction;
import org.apache.qpid.server.txn.LocalTransaction.ActivityTimeAccessor;
import org.apache.qpid.server.txn.ServerTransaction;
import org.apache.qpid.server.util.Action;
import org.apache.qpid.server.util.ServerScopedRuntimeException;
import org.apache.qpid.server.virtualhost.ExchangeExistsException;
import org.apache.qpid.server.virtualhost.ExchangeIsAlternateException;
import org.apache.qpid.server.virtualhost.QueueExistsException;
import org.apache.qpid.server.virtualhost.RequiredExchangeException;
import org.apache.qpid.server.virtualhost.ReservedExchangeNameException;
import org.apache.qpid.transport.network.Ticker;
public class AMQChannel
implements AMQSessionModel<AMQChannel>,
AsyncAutoCommitTransaction.FutureRecorder,
ServerChannelMethodProcessor,
EventLoggerProvider
{
public static final int DEFAULT_PREFETCH = 4096;
private static final Logger _logger = LoggerFactory.getLogger(AMQChannel.class);
private final DefaultQueueAssociationClearingTask
_defaultQueueAssociationClearingTask = new DefaultQueueAssociationClearingTask();
private final int _channelId;
private final Pre0_10CreditManager _creditManager;
private final FlowCreditManager _noAckCreditManager;
private final AccessControlContext _accessControllerContext;
private final SecurityToken _token;
private final PublishAuthorisationCache _publishAuthCahe;
/**
* The delivery tag is unique per channel. This is pre-incremented before putting into the deliver frame so that
* value of this represents the <b>last</b> tag sent out
*/
private long _deliveryTag = 0;
/** A channel has a default queue (the last declared) that is used when no queue name is explicitly set */
private volatile Queue<?> _defaultQueue;
/** This tag is unique per subscription to a queue. The server returns this in response to a basic.consume request. */
private int _consumerTag;
/**
* The current message - which may be partial in the sense that not all frames have been received yet - which has
* been received by this channel. As the frames are received the message gets updated and once all frames have been
* received the message can then be routed.
*/
private IncomingMessage _currentMessage;
/** Maps from consumer tag to subscription instance. Allows us to unsubscribe from a queue. */
private final Map<AMQShortString, ConsumerTarget_0_8> _tag2SubscriptionTargetMap = new HashMap<AMQShortString, ConsumerTarget_0_8>();
private final List<ConsumerTarget_0_8> _consumersWithPendingWork = new ArrayList<>();
private final MessageStore _messageStore;
private final LinkedList<AsyncCommand> _unfinishedCommandsQueue = new LinkedList<AsyncCommand>();
private UnacknowledgedMessageMap _unacknowledgedMessageMap = new UnacknowledgedMessageMapImpl(DEFAULT_PREFETCH);
private final AtomicBoolean _suspended = new AtomicBoolean(false);
private ServerTransaction _transaction;
private final AtomicLong _txnStarts = new AtomicLong(0);
private final AtomicLong _txnCommits = new AtomicLong(0);
private final AtomicLong _txnRejects = new AtomicLong(0);
private final AtomicLong _txnCount = new AtomicLong(0);
private final AMQPConnection_0_8 _connection;
private AtomicBoolean _closing = new AtomicBoolean(false);
private final Set<Object> _blockingEntities = Collections.synchronizedSet(new HashSet<Object>());
private final AtomicBoolean _blocking = new AtomicBoolean(false);
private LogSubject _logSubject;
private volatile boolean _rollingBack;
private List<MessageInstance> _resendList = new ArrayList<MessageInstance>();
private static final
AMQShortString IMMEDIATE_DELIVERY_REPLY_TEXT = new AMQShortString("Immediate delivery is not possible.");
private final ClientDeliveryMethod _clientDeliveryMethod;
private final UUID _id = UUID.randomUUID();
private final List<Action<? super AMQChannel>> _taskList =
new CopyOnWriteArrayList<Action<? super AMQChannel>>();
private final CapacityCheckAction _capacityCheckAction = new CapacityCheckAction();
private final ImmediateAction _immediateAction = new ImmediateAction();
private final Subject _subject;
private final CopyOnWriteArrayList<Consumer<?>> _consumers = new CopyOnWriteArrayList<Consumer<?>>();
private final ConfigurationChangeListener _consumerClosedListener = new ConsumerClosedListener();
private final CopyOnWriteArrayList<ConsumerListener> _consumerListeners = new CopyOnWriteArrayList<ConsumerListener>();
private Session<?> _modelObject;
private long _blockTime;
private long _blockingTimeout;
private boolean _confirmOnPublish;
private long _confirmedMessageCounter;
private volatile long _uncommittedMessageSize;
private final List<StoredMessage<MessageMetaData>> _uncommittedMessages = new ArrayList<>();
private long _maxUncommittedInMemorySize;
private boolean _wireBlockingState;
/** Flag recording if this channel has already written operational logging for prefetch size */
private boolean _prefetchLoggedForChannel = false;
/**
* Handles special case where consumer is polling for messages using qos/flow. Avoids the per-message
* production of channel flow and prefetch operational logging.
*/
private boolean _logChannelFlowMessages = true;
private final CachedFrame _txCommitOkFrame;
public AMQChannel(AMQPConnection_0_8 connection, int channelId, final MessageStore messageStore)
{
_creditManager = new Pre0_10CreditManager(0l,0l, connection);
_noAckCreditManager = new NoAckCreditManager(connection);
_connection = connection;
_channelId = channelId;
_subject = new Subject(false, connection.getSubject().getPrincipals(),
connection.getSubject().getPublicCredentials(),
connection.getSubject().getPrivateCredentials());
_subject.getPrincipals().add(new SessionPrincipal(this));
_accessControllerContext = connection.getAccessControlContextFromSubject(_subject);
_token = (_connection.getAddressSpace() instanceof ConfiguredObject)
? ((ConfiguredObject)_connection.getAddressSpace()).newToken(_subject)
:_connection.getBroker().newToken(_subject);
_maxUncommittedInMemorySize = connection.getContextProvider().getContextValue(Long.class, Connection.MAX_UNCOMMITTED_IN_MEMORY_SIZE);
_logSubject = new ChannelLogSubject(this);
_publishAuthCahe = new PublishAuthorisationCache(_token,
connection.getContextValue(Long.class, Session.PRODUCER_AUTH_CACHE_TIMEOUT),
connection.getContextValue(Integer.class, Session.PRODUCER_AUTH_CACHE_SIZE));
_messageStore = messageStore;
_blockingTimeout = connection.getBroker().getContextValue(Long.class,
Broker.CHANNEL_FLOW_CONTROL_ENFORCEMENT_TIMEOUT);
// by default the session is non-transactional
_transaction = new AsyncAutoCommitTransaction(_messageStore, this);
MethodRegistry methodRegistry = _connection.getMethodRegistry();
AMQMethodBody responseBody = methodRegistry.createTxCommitOkBody();
_txCommitOkFrame = new CachedFrame(responseBody.generateFrame(_channelId));
_clientDeliveryMethod = connection.createDeliveryMethod(_channelId);
AccessController.doPrivileged((new PrivilegedAction<Object>()
{
@Override
public Object run()
{
message(ChannelMessages.CREATE());
return null;
}
}),_accessControllerContext);
}
@Override
public void doTimeoutAction(String reason)
{
_connection.sendConnectionCloseAsync(AMQConstant.RESOURCE_ERROR, reason);
}
private void message(final LogMessage message)
{
getEventLogger().message(message);
}
public AccessControlContext getAccessControllerContext()
{
return _accessControllerContext;
}
private boolean performGet(final MessageSource queue,
final boolean acks)
throws MessageSource.ExistingConsumerPreventsExclusive,
MessageSource.ExistingExclusiveConsumer, MessageSource.ConsumerAccessRefused
{
final FlowCreditManager singleMessageCredit = new MessageOnlyCreditManager(1L);
final GetDeliveryMethod getDeliveryMethod =
new GetDeliveryMethod(singleMessageCredit, queue);
final RecordDeliveryMethod getRecordMethod = new RecordDeliveryMethod()
{
public void recordMessageDelivery(final ConsumerImpl sub,
final MessageInstance entry,
final long deliveryTag)
{
addUnacknowledgedMessage(entry, deliveryTag, null);
}
};
ConsumerTarget_0_8 target;
EnumSet<ConsumerImpl.Option> options = EnumSet.of(ConsumerImpl.Option.TRANSIENT, ConsumerImpl.Option.ACQUIRES,
ConsumerImpl.Option.SEES_REQUEUES);
if (acks)
{
target = ConsumerTarget_0_8.createAckTarget(this,
AMQShortString.EMPTY_STRING, null,
singleMessageCredit, getDeliveryMethod, getRecordMethod);
}
else
{
target = ConsumerTarget_0_8.createGetNoAckTarget(this,
AMQShortString.EMPTY_STRING, null,
singleMessageCredit, getDeliveryMethod, getRecordMethod);
}
ConsumerImpl sub = queue.addConsumer(target, null, AMQMessage.class, "", options, null);
sub.flush();
sub.close();
return getDeliveryMethod.hasDeliveredMessage();
}
/** Sets this channel to be part of a local transaction */
public void setLocalTransactional()
{
_transaction = new LocalTransaction(_messageStore, new ActivityTimeAccessor()
{
@Override
public long getActivityTime()
{
return _connection.getLastReadTime();
}
});
_txnStarts.incrementAndGet();
}
public boolean isTransactional()
{
return _transaction.isTransactional();
}
public void receivedComplete()
{
sync();
}
private void incrementOutstandingTxnsIfNecessary()
{
if(isTransactional())
{
//There can currently only be at most one outstanding transaction
//due to only having LocalTransaction support. Set value to 1 if 0.
_txnCount.compareAndSet(0,1);
}
}
private void decrementOutstandingTxnsIfNecessary()
{
if(isTransactional())
{
//There can currently only be at most one outstanding transaction
//due to only having LocalTransaction support. Set value to 0 if 1.
_txnCount.compareAndSet(1,0);
}
}
public Long getTxnCommits()
{
return _txnCommits.get();
}
public Long getTxnRejects()
{
return _txnRejects.get();
}
public Long getTxnCount()
{
return _txnCount.get();
}
public Long getTxnStart()
{
return _txnStarts.get();
}
public int getChannelId()
{
return _channelId;
}
public void setPublishFrame(MessagePublishInfo info, final MessageDestination e)
{
_currentMessage = new IncomingMessage(info);
_currentMessage.setMessageDestination(e);
}
public void publishContentHeader(ContentHeaderBody contentHeaderBody)
{
if (_logger.isDebugEnabled())
{
_logger.debug("Content header received on channel " + _channelId);
}
_currentMessage.setContentHeaderBody(contentHeaderBody);
deliverCurrentMessageIfComplete();
}
private void deliverCurrentMessageIfComplete()
{
// check and deliver if header says body length is zero
if (_currentMessage.allContentReceived())
{
MessagePublishInfo info = _currentMessage.getMessagePublishInfo();
String routingKey = AMQShortString.toString(info.getRoutingKey());
try
{
final MessageDestination destination = _currentMessage.getDestination();
ContentHeaderBody contentHeader = _currentMessage.getContentHeader();
_connection.checkAuthorizedMessagePrincipal(AMQShortString.toString(contentHeader.getProperties().getUserId()));
_publishAuthCahe.authorisePublish(destination, routingKey, info.isImmediate(), _connection.getLastReadTime());
if (_confirmOnPublish)
{
_confirmedMessageCounter++;
}
Runnable finallyAction = null;
long bodySize = _currentMessage.getSize();
long timestamp = contentHeader.getProperties().getTimestamp();
try
{
final MessagePublishInfo messagePublishInfo = _currentMessage.getMessagePublishInfo();
final MessageMetaData messageMetaData =
new MessageMetaData(messagePublishInfo,
contentHeader,
getConnection().getLastReadTime());
final MessageHandle<MessageMetaData> handle = _messageStore.addMessage(messageMetaData);
int bodyCount = _currentMessage.getBodyCount();
if (bodyCount > 0)
{
for (int i = 0; i < bodyCount; i++)
{
ContentBody contentChunk = _currentMessage.getContentChunk(i);
handle.addContent(contentChunk.getPayload());
contentChunk.dispose();
}
}
final StoredMessage<MessageMetaData> storedMessage = handle.allContentAdded();
final AMQMessage amqMessage = createAMQMessage(storedMessage);
MessageReference reference = amqMessage.newReference();
try
{
_currentMessage = null;
final boolean immediate = messagePublishInfo.isImmediate();
final InstanceProperties instanceProperties =
new InstanceProperties()
{
@Override
public Object getProperty(final Property prop)
{
switch (prop)
{
case EXPIRATION:
return amqMessage.getExpiration();
case IMMEDIATE:
return immediate;
case PERSISTENT:
return amqMessage.isPersistent();
case MANDATORY:
return messagePublishInfo.isMandatory();
case REDELIVERED:
return false;
}
return null;
}
};
int enqueues = destination.send(amqMessage,
amqMessage.getInitialRoutingAddress(),
instanceProperties, _transaction,
immediate ? _immediateAction : _capacityCheckAction
);
if (enqueues == 0)
{
finallyAction = handleUnroutableMessage(amqMessage);
}
else
{
if (_confirmOnPublish)
{
BasicAckBody responseBody = _connection.getMethodRegistry()
.createBasicAckBody(_confirmedMessageCounter, false);
_connection.writeFrame(responseBody.generateFrame(_channelId));
}
incrementUncommittedMessageSize(storedMessage);
incrementOutstandingTxnsIfNecessary();
}
}
finally
{
reference.release();
if (finallyAction != null)
{
finallyAction.run();
}
}
}
finally
{
_connection.registerMessageReceived(bodySize, timestamp);
_currentMessage = null;
}
}
catch (AccessControlException e)
{
_connection.sendConnectionClose(AMQConstant.ACCESS_REFUSED, e.getMessage(), getChannelId());
}
}
}
private void incrementUncommittedMessageSize(final StoredMessage<MessageMetaData> handle)
{
if (isTransactional())
{
_uncommittedMessageSize += handle.getMetaData().getContentSize();
if (_uncommittedMessageSize > getMaxUncommittedInMemorySize())
{
handle.flowToDisk();
if(!_uncommittedMessages.isEmpty() || _uncommittedMessageSize == handle.getMetaData().getContentSize())
{
messageWithSubject(ChannelMessages.LARGE_TRANSACTION_WARN(_uncommittedMessageSize));
}
if(!_uncommittedMessages.isEmpty())
{
for (StoredMessage<MessageMetaData> uncommittedHandle : _uncommittedMessages)
{
uncommittedHandle.flowToDisk();
}
_uncommittedMessages.clear();
}
}
else
{
_uncommittedMessages.add(handle);
}
}
}
/**
* Either throws a {@link AMQConnectionException} or returns the message
*
* Pre-requisite: the current message is judged to have no destination queues.
*
* @throws AMQConnectionException if the message is mandatory close-on-no-route
* @see AMQPConnection_0_8Impl#isCloseWhenNoRoute()
*/
private Runnable handleUnroutableMessage(AMQMessage message)
{
boolean mandatory = message.isMandatory();
String exchangeName = AMQShortString.toString(message.getMessagePublishInfo().getExchange());
String routingKey = AMQShortString.toString(message.getMessagePublishInfo().getRoutingKey());
final String description = String.format(
"[Exchange: %s, Routing key: %s]",
exchangeName,
routingKey);
boolean closeOnNoRoute = _connection.isCloseWhenNoRoute();
Runnable returnVal = null;
if(_logger.isDebugEnabled())
{
_logger.debug(String.format(
"Unroutable message %s, mandatory=%s, transactionalSession=%s, closeOnNoRoute=%s",
description, mandatory, isTransactional(), closeOnNoRoute));
}
if (mandatory && isTransactional() && !_confirmOnPublish && _connection.isCloseWhenNoRoute())
{
returnVal = new Runnable()
{
@Override
public void run()
{
_connection.sendConnectionClose(AMQConstant.NO_ROUTE,
"No route for message " + description, _channelId);
}
};
}
else
{
if (mandatory || message.isImmediate())
{
if(_confirmOnPublish)
{
_connection.writeFrame(new AMQFrame(_channelId, new BasicNackBody(_confirmedMessageCounter, false, false)));
}
_transaction.addPostTransactionAction(new WriteReturnAction(AMQConstant.NO_ROUTE,
"No Route for message "
+ description,
message));
}
else
{
message(ExchangeMessages.DISCARDMSG(exchangeName, routingKey));
}
}
return returnVal;
}
public void publishContentBody(ContentBody contentBody)
{
if (_logger.isDebugEnabled())
{
_logger.debug(debugIdentity() + " content body received on channel " + _channelId);
}
try
{
long currentSize = _currentMessage.addContentBodyFrame(contentBody);
if(currentSize > _currentMessage.getSize())
{
_connection.sendConnectionClose(AMQConstant.FRAME_ERROR,
"More message data received than content header defined",
_channelId);
}
else
{
deliverCurrentMessageIfComplete();
}
}
catch (RuntimeException e)
{
// we want to make sure we don't keep a reference to the message in the
// event of an error
_currentMessage = null;
throw e;
}
}
public long getNextDeliveryTag()
{
return ++_deliveryTag;
}
public int getNextConsumerTag()
{
return ++_consumerTag;
}
public ConsumerTarget getSubscription(AMQShortString tag)
{
return _tag2SubscriptionTargetMap.get(tag);
}
/**
* Subscribe to a queue. We register all subscriptions in the channel so that if the channel is closed we can clean
* up all subscriptions, even if the client does not explicitly unsubscribe from all queues.
*
*
* @param tag the tag chosen by the client (if null, server will generate one)
* @param sources the queues to subscribe to
* @param acks Are acks enabled for this subscriber
* @param arguments Filters to apply to this subscriber
*
* @param exclusive Flag requesting exclusive access to the queue
* @return the consumer tag. This is returned to the subscriber and used in subsequent unsubscribe requests
*/
public AMQShortString consumeFromSource(AMQShortString tag, Collection<MessageSource> sources, boolean acks,
FieldTable arguments, boolean exclusive, boolean noLocal)
throws MessageSource.ExistingConsumerPreventsExclusive,
MessageSource.ExistingExclusiveConsumer,
AMQInvalidArgumentException,
MessageSource.ConsumerAccessRefused, ConsumerTagInUseException
{
if (tag == null)
{
tag = new AMQShortString("sgen_" + getNextConsumerTag());
}
if (_tag2SubscriptionTargetMap.containsKey(tag))
{
throw new ConsumerTagInUseException("Consumer already exists with same tag: " + tag);
}
ConsumerTarget_0_8 target;
EnumSet<ConsumerImpl.Option> options = EnumSet.noneOf(ConsumerImpl.Option.class);
final boolean multiQueue = sources.size()>1;
if(arguments != null && Boolean.TRUE.equals(arguments.get(AMQPFilterTypes.NO_CONSUME.getValue())))
{
target = ConsumerTarget_0_8.createBrowserTarget(this, tag, arguments, _noAckCreditManager, multiQueue);
}
else if(acks)
{
target = ConsumerTarget_0_8.createAckTarget(this, tag, arguments, _creditManager, multiQueue);
options.add(ConsumerImpl.Option.ACQUIRES);
options.add(ConsumerImpl.Option.SEES_REQUEUES);
}
else
{
target = ConsumerTarget_0_8.createNoAckTarget(this, tag, arguments, _noAckCreditManager, multiQueue);
options.add(ConsumerImpl.Option.ACQUIRES);
options.add(ConsumerImpl.Option.SEES_REQUEUES);
}
if(exclusive)
{
options.add(ConsumerImpl.Option.EXCLUSIVE);
}
// So to keep things straight we put before the call and catch all exceptions from the register and tidy up.
// We add before we register as the Async Delivery process may AutoClose the subscriber
// so calling _cT2QM.remove before we have done put which was after the register succeeded.
// So to keep things straight we put before the call and catch all exceptions from the register and tidy up.
_tag2SubscriptionTargetMap.put(tag, target);
try
{
FilterManager filterManager = FilterManagerFactory.createManager(FieldTable.convertToMap(arguments));
if(noLocal)
{
if(filterManager == null)
{
filterManager = new FilterManager();
}
MessageFilter filter = new NoLocalFilter();
filterManager.add(filter.getName(), filter);
}
if(arguments != null && arguments.containsKey(AMQPFilterTypes.REPLAY_PERIOD.toString()))
{
Object value = arguments.get(AMQPFilterTypes.REPLAY_PERIOD.toString());
final long period;
if(value instanceof Number)
{
period = ((Number)value).longValue();
}
else if(value instanceof String)
{
try
{
period = Long.parseLong(value.toString());
}
catch (NumberFormatException e)
{
throw new AMQInvalidArgumentException("Cannot parse value " + value + " as a number for filter " + AMQPFilterTypes.REPLAY_PERIOD.toString());
}
}
else
{
throw new AMQInvalidArgumentException("Cannot parse value " + value + " as a number for filter " + AMQPFilterTypes.REPLAY_PERIOD.toString());
}
final long startingFrom = System.currentTimeMillis() - (1000l * period);
if(filterManager == null)
{
filterManager = new FilterManager();
}
MessageFilter filter = new ArrivalTimeFilter(startingFrom, period==0);
filterManager.add(filter.getName(), filter);
}
Integer priority = null;
if(arguments != null && arguments.containsKey("x-priority"))
{
Object value = arguments.get("x-priority");
if(value instanceof Number)
{
priority = ((Number)value).intValue();
}
else if(value instanceof String || value instanceof AMQShortString)
{
try
{
priority = Integer.parseInt(value.toString());
}
catch (NumberFormatException e)
{
// use default vlaue
}
}
}
for(MessageSource source : sources)
{
ConsumerImpl sub =
source.addConsumer(target,
filterManager,
AMQMessage.class,
AMQShortString.toString(tag),
options, priority);
if (sub instanceof Consumer<?>)
{
final Consumer<?> modelConsumer = (Consumer<?>) sub;
consumerAdded(modelConsumer);
modelConsumer.addChangeListener(_consumerClosedListener);
_consumers.add(modelConsumer);
}
}
}
catch (AccessControlException
| MessageSource.ExistingExclusiveConsumer
| MessageSource.ExistingConsumerPreventsExclusive
| AMQInvalidArgumentException
| MessageSource.ConsumerAccessRefused e)
{
_tag2SubscriptionTargetMap.remove(tag);
throw e;
}
return tag;
}
/**
* Unsubscribe a consumer from a queue.
* @param consumerTag
* @return true if the consumerTag had a mapped queue that could be unregistered.
*/
public boolean unsubscribeConsumer(AMQShortString consumerTag)
{
if (_logger.isDebugEnabled())
{
_logger.debug("Unsubscribing consumer '{}' on channel {}", consumerTag, this);
}
ConsumerTarget_0_8 target = _tag2SubscriptionTargetMap.remove(consumerTag);
Collection<ConsumerImpl> subs = target == null ? null : target.getConsumers();
if (subs != null)
{
for(ConsumerImpl sub : subs)
{
sub.close();
if (sub instanceof Consumer<?>)
{
_consumers.remove(sub);
}
}
return true;
}
else
{
_logger.warn("Attempt to unsubscribe consumer with tag '" + consumerTag + "' which is not registered.");
}
return false;
}
@Override
public void close()
{
close(null, null);
}
public void close(AMQConstant cause, String message)
{
if(!_closing.compareAndSet(false, true))
{
//Channel is already closing
return;
}
try
{
unsubscribeAllConsumers();
setDefaultQueue(null);
if(_modelObject != null)
{
_modelObject.delete();
}
for (Action<? super AMQChannel> task : _taskList)
{
task.performAction(this);
}
_transaction.rollback();
requeue();
}
finally
{
LogMessage operationalLogMessage = cause == null ?
ChannelMessages.CLOSE() :
ChannelMessages.CLOSE_FORCED(cause.getCode(), message);
messageWithSubject(operationalLogMessage);
}
}
private void messageWithSubject(final LogMessage operationalLogMessage)
{
getEventLogger().message(_logSubject, operationalLogMessage);
}
private void unsubscribeAllConsumers()
{
if (_logger.isDebugEnabled())
{
if (!_tag2SubscriptionTargetMap.isEmpty())
{
_logger.debug("Unsubscribing all consumers on channel " + toString());
}
else
{
_logger.debug("No consumers to unsubscribe on channel " + toString());
}
}
Set<AMQShortString> subscriptionTags = new HashSet<>(_tag2SubscriptionTargetMap.keySet());
for (AMQShortString tag : subscriptionTags)
{
unsubscribeConsumer(tag);
}
}
/**
* Add a message to the channel-based list of unacknowledged messages
*
* @param entry the record of the message on the queue that was delivered
* @param deliveryTag the delivery tag used when delivering the message (see protocol spec for description of the
* delivery tag)
* @param consumer The consumer that is to acknowledge this message.
*/
public void addUnacknowledgedMessage(MessageInstance entry, long deliveryTag, ConsumerImpl consumer)
{
if (_logger.isDebugEnabled())
{
_logger.debug(debugIdentity() + " Adding unacked message(" + entry.getMessage().toString() + " DT:" + deliveryTag
+ ") for " + consumer + " on " + entry.getOwningResource().getName());
}
_unacknowledgedMessageMap.add(deliveryTag, entry);
}
private final String id = "(" + System.identityHashCode(this) + ")";
public String debugIdentity()
{
return _channelId + id;
}
/**
* Called to attempt re-delivery all outstanding unacknowledged messages on the channel. May result in delivery to
* this same channel or to other subscribers.
*
*/
private void requeue()
{
// we must create a new map since all the messages will get a new delivery tag when they are redelivered
Collection<MessageInstance> messagesToBeDelivered = _unacknowledgedMessageMap.cancelAllMessages();
if (!messagesToBeDelivered.isEmpty())
{
if (_logger.isDebugEnabled())
{
_logger.debug("Requeuing " + messagesToBeDelivered.size() + " unacked messages. for " + toString());
}
}
for (MessageInstance unacked : messagesToBeDelivered)
{
// Mark message redelivered
unacked.setRedelivered();
// Ensure message is released for redelivery
unacked.release(unacked.getAcquiringConsumer());
}
}
/**
* Requeue a single message
*
* @param deliveryTag The message to requeue
*
*/
public void requeue(long deliveryTag)
{
MessageInstance unacked = _unacknowledgedMessageMap.remove(deliveryTag);
if (unacked != null)
{
// Mark message redelivered
unacked.setRedelivered();
// Ensure message is released for redelivery
unacked.release(unacked.getAcquiringConsumer());
}
else
{
_logger.warn("Requested requeue of message:" + deliveryTag + " but no such delivery tag exists."
+ _unacknowledgedMessageMap.size());
}
}
public boolean isMaxDeliveryCountEnabled(final long deliveryTag)
{
final MessageInstance queueEntry = _unacknowledgedMessageMap.get(deliveryTag);
if (queueEntry != null)
{
final int maximumDeliveryCount = queueEntry.getMaximumDeliveryCount();
return maximumDeliveryCount > 0;
}
return false;
}
public boolean isDeliveredTooManyTimes(final long deliveryTag)
{
final MessageInstance queueEntry = _unacknowledgedMessageMap.get(deliveryTag);
if (queueEntry != null)
{
final int maximumDeliveryCount = queueEntry.getMaximumDeliveryCount();
final int numDeliveries = queueEntry.getDeliveryCount();
return maximumDeliveryCount != 0 && numDeliveries >= maximumDeliveryCount;
}
return false;
}
/**
* Called to resend all outstanding unacknowledged messages to this same channel.
*
*/
private void resend()
{
final Map<Long, MessageInstance> msgToRequeue = new LinkedHashMap<Long, MessageInstance>();
final Map<Long, MessageInstance> msgToResend = new LinkedHashMap<Long, MessageInstance>();
if (_logger.isDebugEnabled())
{
_logger.debug("unacked map Size:" + _unacknowledgedMessageMap.size());
}
// Process the Unacked-Map.
// Marking messages who still have a consumer for to be resent
// and those that don't to be requeued.
_unacknowledgedMessageMap.visit(new ExtractResendAndRequeue(_unacknowledgedMessageMap,
msgToRequeue,
msgToResend
));
// Process Messages to Resend
if (_logger.isDebugEnabled())
{
if (!msgToResend.isEmpty())
{
_logger.debug("Preparing (" + msgToResend.size() + ") message to resend.");
}
else
{
_logger.debug("No message to resend.");
}
}
for (Map.Entry<Long, MessageInstance> entry : msgToResend.entrySet())
{
MessageInstance message = entry.getValue();
long deliveryTag = entry.getKey();
//Amend the delivery counter as the client hasn't seen these messages yet.
message.decrementDeliveryCount();
// Without any details from the client about what has been processed we have to mark
// all messages in the unacked map as redelivered.
message.setRedelivered();
if (!message.resend())
{
msgToRequeue.put(deliveryTag, message);
}
} // for all messages
// } else !isSuspend
if (_logger.isDebugEnabled())
{
if (!msgToRequeue.isEmpty())
{
_logger.debug("Preparing (" + msgToRequeue.size() + ") message to requeue");
}
}
// Process Messages to Requeue at the front of the queue
for (Map.Entry<Long, MessageInstance> entry : msgToRequeue.entrySet())
{
MessageInstance message = entry.getValue();
long deliveryTag = entry.getKey();
//Amend the delivery counter as the client hasn't seen these messages yet.
message.decrementDeliveryCount();
_unacknowledgedMessageMap.remove(deliveryTag);
message.setRedelivered();
message.release(message.getAcquiringConsumer());
}
}
/**
* Acknowledge one or more messages.
*
* @param deliveryTag the last delivery tag
* @param multiple if true will acknowledge all messages up to an including the delivery tag. if false only
* acknowledges the single message specified by the delivery tag
*
*/
private void acknowledgeMessage(long deliveryTag, boolean multiple)
{
Collection<MessageInstance> ackedMessages = getAckedMessages(deliveryTag, multiple);
_transaction.dequeue(ackedMessages, new MessageAcknowledgeAction(ackedMessages));
}
private Collection<MessageInstance> getAckedMessages(long deliveryTag, boolean multiple)
{
return _unacknowledgedMessageMap.acknowledge(deliveryTag, multiple);
}
/**
* Used only for testing purposes.
*
* @return the map of unacknowledged messages
*/
public UnacknowledgedMessageMap getUnacknowledgedMessageMap()
{
return _unacknowledgedMessageMap;
}
/**
* Called from the ChannelFlowHandler to suspend this Channel
* @param suspended boolean, should this Channel be suspended
*/
public void setSuspended(boolean suspended)
{
boolean wasSuspended = _suspended.getAndSet(suspended);
if (wasSuspended != suspended)
{
// Log Flow Started before we start the subscriptions
if (!suspended && _logChannelFlowMessages)
{
messageWithSubject(ChannelMessages.FLOW("Started"));
}
// This section takes two different approaches to perform to perform
// the same function. Ensuring that the Subscription has taken note
// of the change in Channel State
// Here we have become unsuspended and so we ask each the queue to
// perform an Async delivery for each of the subscriptions in this
// Channel. The alternative would be to ensure that the subscription
// had received the change in suspension state. That way the logic
// behind deciding to start an async delivery was located with the
// Subscription.
if (wasSuspended)
{
// may need to deliver queued messages
for (ConsumerTarget_0_8 s : getConsumerTargets())
{
for(ConsumerImpl sub : s.getConsumers())
{
sub.externalStateChange();
}
}
}
// Here we have become suspended so we need to ensure that each of
// the Subscriptions has noticed this change so that we can be sure
// they are not still sending messages. Again the code here is a
// very simplistic approach to ensure that the change of suspension
// has been noticed by each of the Subscriptions. Unlike the above
// case we don't actually need to do anything else.
if (!wasSuspended)
{
// may need to deliver queued messages
ensureConsumersNoticedStateChange();
}
// Log Suspension only after we have confirmed all suspensions are
// stopped.
if (suspended && _logChannelFlowMessages)
{
messageWithSubject(ChannelMessages.FLOW("Stopped"));
}
}
}
public boolean isSuspended()
{
return _suspended.get() || _closing.get() || _connection.isClosing();
}
public void commit(final Runnable immediateAction, boolean async)
{
if(async && _transaction instanceof LocalTransaction)
{
((LocalTransaction)_transaction).commitAsync(new Runnable()
{
@Override
public void run()
{
try
{
immediateAction.run();
}
finally
{
_txnCommits.incrementAndGet();
_txnStarts.incrementAndGet();
decrementOutstandingTxnsIfNecessary();
}
}
});
}
else
{
_transaction.commit(immediateAction);
_txnCommits.incrementAndGet();
_txnStarts.incrementAndGet();
decrementOutstandingTxnsIfNecessary();
}
resetUncommittedMessages();
}
private void resetUncommittedMessages()
{
_uncommittedMessageSize = 0l;
_uncommittedMessages.clear();
}
private void rollback(Runnable postRollbackTask)
{
// stop all subscriptions
_rollingBack = true;
boolean requiresSuspend = _suspended.compareAndSet(false,true); // TODO This is probably superfluous owing to the
// message assignment suspended logic in NBC.
// ensure all subscriptions have seen the change to the channel state
ensureConsumersNoticedStateChange();
try
{
_transaction.rollback();
}
finally
{
_rollingBack = false;
_txnRejects.incrementAndGet();
_txnStarts.incrementAndGet();
decrementOutstandingTxnsIfNecessary();
resetUncommittedMessages();
}
postRollbackTask.run();
for(MessageInstance entry : _resendList)
{
ConsumerImpl sub = entry.getAcquiringConsumer();
if (sub == null || sub.isClosed())
{
entry.release(sub);
}
else
{
entry.resend();
}
}
_resendList.clear();
if(requiresSuspend)
{
_suspended.set(false);
for(ConsumerTarget_0_8 target : getConsumerTargets())
{
for(ConsumerImpl sub : target.getConsumers())
{
sub.externalStateChange();
}
}
}
}
public String toString()
{
return "("+ _suspended.get() + ", " + _closing.get() + ", " + _connection.isClosing() + ") "+"["+ _connection.toString()+":"+_channelId+"]";
}
public boolean isClosing()
{
return _closing.get();
}
public AMQPConnection_0_8 getConnection()
{
return _connection;
}
public void setCredit(final long prefetchSize, final int prefetchCount)
{
if (!_prefetchLoggedForChannel)
{
message(ChannelMessages.PREFETCH_SIZE(prefetchSize, prefetchCount));
_prefetchLoggedForChannel = true;
}
if (prefetchCount <= 1 && prefetchSize == 0 )
{
_logChannelFlowMessages = false;
}
_creditManager.setCreditLimits(prefetchSize, prefetchCount);
}
public MessageStore getMessageStore()
{
return _messageStore;
}
public ClientDeliveryMethod getClientDeliveryMethod()
{
return _clientDeliveryMethod;
}
private final RecordDeliveryMethod _recordDeliveryMethod = new RecordDeliveryMethod()
{
public void recordMessageDelivery(final ConsumerImpl sub, final MessageInstance entry, final long deliveryTag)
{
addUnacknowledgedMessage(entry, deliveryTag, sub);
}
};
public RecordDeliveryMethod getRecordDeliveryMethod()
{
return _recordDeliveryMethod;
}
private AMQMessage createAMQMessage(StoredMessage<MessageMetaData> handle)
{
AMQMessage message = new AMQMessage(handle, _connection.getReference());
return message;
}
@Override
public UUID getId()
{
return _id;
}
@Override
public AMQPConnection_0_8 getAMQPConnection()
{
return _connection;
}
public LogSubject getLogSubject()
{
return _logSubject;
}
@Override
public int compareTo(AMQSessionModel o)
{
return getId().compareTo(o.getId());
}
@Override
public void addDeleteTask(final Action<? super AMQChannel> task)
{
_taskList.add(task);
}
@Override
public void removeDeleteTask(final Action<? super AMQChannel> task)
{
_taskList.remove(task);
}
public Subject getSubject()
{
return _subject;
}
public boolean hasCurrentMessage()
{
return _currentMessage != null;
}
public long getMaxUncommittedInMemorySize()
{
return _maxUncommittedInMemorySize;
}
private class NoLocalFilter implements MessageFilter
{
private final Object _connectionReference;
public NoLocalFilter()
{
_connectionReference = getConnectionReference();
}
@Override
public String getName()
{
return AMQPFilterTypes.NO_LOCAL.toString();
}
@Override
public boolean matches(final Filterable message)
{
return message.getConnectionReference() != _connectionReference;
}
@Override
public boolean startAtTail()
{
return false;
}
@Override
public String toString()
{
return "NoLocalFilter[]";
}
}
private class GetDeliveryMethod implements ClientDeliveryMethod
{
private final FlowCreditManager _singleMessageCredit;
private final MessageSource _queue;
private boolean _deliveredMessage;
public GetDeliveryMethod(final FlowCreditManager singleMessageCredit,
final MessageSource queue)
{
_singleMessageCredit = singleMessageCredit;
_queue = queue;
}
@Override
public long deliverToClient(final ConsumerImpl sub, final ServerMessage message,
final InstanceProperties props, final long deliveryTag)
{
_singleMessageCredit.useCreditForMessage(message.getSize());
int queueSize = _queue instanceof Queue ? ((Queue<?>)_queue).getQueueDepthMessages() : 0;
long size = _connection.getProtocolOutputConverter().writeGetOk(message,
props,
AMQChannel.this.getChannelId(),
deliveryTag,
queueSize);
_deliveredMessage = true;
return size;
}
public boolean hasDeliveredMessage()
{
return _deliveredMessage;
}
}
private class ImmediateAction implements Action<MessageInstance>
{
public ImmediateAction()
{
}
public void performAction(MessageInstance entry)
{
TransactionLogResource queue = entry.getOwningResource();
if (!entry.getDeliveredToConsumer() && entry.acquire())
{
ServerTransaction txn = new LocalTransaction(_messageStore);
final AMQMessage message = (AMQMessage) entry.getMessage();
MessageReference ref = message.newReference();
try
{
entry.delete();
txn.dequeue(entry.getEnqueueRecord(),
new ServerTransaction.Action()
{
@Override
public void postCommit()
{
final ProtocolOutputConverter outputConverter =
_connection.getProtocolOutputConverter();
outputConverter.writeReturn(message.getMessagePublishInfo(),
message.getContentHeaderBody(),
message,
_channelId,
AMQConstant.NO_CONSUMERS.getCode(),
IMMEDIATE_DELIVERY_REPLY_TEXT);
}
@Override
public void onRollback()
{
}
}
);
txn.commit();
}
finally
{
ref.release();
}
}
else
{
if(queue instanceof CapacityChecker)
{
((CapacityChecker)queue).checkCapacity(AMQChannel.this);
}
}
}
}
private final class CapacityCheckAction implements Action<MessageInstance>
{
@Override
public void performAction(final MessageInstance entry)
{
TransactionLogResource queue = entry.getOwningResource();
if(queue instanceof CapacityChecker)
{
((CapacityChecker)queue).checkCapacity(AMQChannel.this);
}
}
}
private class MessageAcknowledgeAction implements ServerTransaction.Action
{
private Collection<MessageInstance> _ackedMessages;
public MessageAcknowledgeAction(Collection<MessageInstance> ackedMessages)
{
_ackedMessages = ackedMessages;
}
public void postCommit()
{
try
{
for(MessageInstance entry : _ackedMessages)
{
entry.delete();
}
}
finally
{
_ackedMessages = Collections.emptySet();
}
}
public void onRollback()
{
// explicit rollbacks resend the message after the rollback-ok is sent
if(_rollingBack)
{
for(MessageInstance entry : _ackedMessages)
{
entry.makeAcquisitionStealable();
}
_resendList.addAll(_ackedMessages);
}
else
{
try
{
for(MessageInstance entry : _ackedMessages)
{
entry.release(entry.getAcquiringConsumer());
}
}
finally
{
_ackedMessages = Collections.emptySet();
}
}
}
}
private class WriteReturnAction implements ServerTransaction.Action
{
private final AMQConstant _errorCode;
private final String _description;
private final MessageReference<AMQMessage> _reference;
public WriteReturnAction(AMQConstant errorCode,
String description,
AMQMessage message)
{
_errorCode = errorCode;
_description = description;
_reference = message.newReference();
}
public void postCommit()
{
AMQMessage message = _reference.getMessage();
_connection.getProtocolOutputConverter().writeReturn(message.getMessagePublishInfo(),
message.getContentHeaderBody(),
message,
_channelId,
_errorCode.getCode(),
AMQShortString.validValueOf(_description));
_reference.release();
}
public void onRollback()
{
_reference.release();
}
}
public synchronized void block()
{
if(_blockingEntities.add(this))
{
if(_blocking.compareAndSet(false,true))
{
messageWithSubject(ChannelMessages.FLOW_ENFORCED("** All Queues **"));
getConnection().notifyWork();
}
}
}
public synchronized void unblock()
{
if(_blockingEntities.remove(this))
{
if(_blockingEntities.isEmpty() && _blocking.compareAndSet(true,false))
{
messageWithSubject(ChannelMessages.FLOW_REMOVED());
getConnection().notifyWork();
}
}
}
public synchronized void block(Queue<?> queue)
{
if(_blockingEntities.add(queue))
{
if(_blocking.compareAndSet(false,true))
{
messageWithSubject(ChannelMessages.FLOW_ENFORCED(queue.getName()));
getConnection().notifyWork();
}
}
}
public synchronized void unblock(Queue<?> queue)
{
if(_blockingEntities.remove(queue))
{
if(_blockingEntities.isEmpty() && _blocking.compareAndSet(true,false) && !isClosing())
{
messageWithSubject(ChannelMessages.FLOW_REMOVED());
getConnection().notifyWork();
}
}
}
@Override
public void transportStateChanged()
{
_creditManager.restoreCredit(0, 0);
_noAckCreditManager.restoreCredit(0, 0);
}
@Override
public Object getConnectionReference()
{
return getConnection().getReference();
}
public int getUnacknowledgedMessageCount()
{
return getUnacknowledgedMessageMap().size();
}
private void flow(boolean flow)
{
MethodRegistry methodRegistry = _connection.getMethodRegistry();
AMQMethodBody responseBody = methodRegistry.createChannelFlowBody(flow);
_connection.writeFrame(responseBody.generateFrame(_channelId));
}
@Override
public boolean getBlocking()
{
return _blocking.get();
}
public NamedAddressSpace getAddressSpace()
{
return getConnection().getAddressSpace();
}
private void deadLetter(long deliveryTag)
{
final UnacknowledgedMessageMap unackedMap = getUnacknowledgedMessageMap();
final MessageInstance rejectedQueueEntry = unackedMap.remove(deliveryTag);
if (rejectedQueueEntry == null)
{
_logger.warn("No message found, unable to DLQ delivery tag: " + deliveryTag);
}
else
{
final ServerMessage msg = rejectedQueueEntry.getMessage();
int requeues = 0;
if (rejectedQueueEntry.makeAcquisitionUnstealable(rejectedQueueEntry.getAcquiringConsumer()))
{
requeues = rejectedQueueEntry.routeToAlternate(new Action<MessageInstance>()
{
@Override
public void performAction(final MessageInstance requeueEntry)
{
messageWithSubject(ChannelMessages.DEADLETTERMSG(msg.getMessageNumber(),
requeueEntry.getOwningResource()
.getName()));
}
}, null);
}
if(requeues == 0)
{
final TransactionLogResource owningResource = rejectedQueueEntry.getOwningResource();
if(owningResource instanceof Queue)
{
final Queue<?> queue = (Queue<?>) owningResource;
final Exchange altExchange = queue.getAlternateExchange();
if (altExchange == null)
{
if (_logger.isDebugEnabled())
{
_logger.debug(
"No alternate exchange configured for queue, must discard the message as unable to DLQ: delivery tag: "
+ deliveryTag);
}
messageWithSubject(ChannelMessages.DISCARDMSG_NOALTEXCH(msg.getMessageNumber(),
queue.getName(),
msg.getInitialRoutingAddress()));
}
else
{
if (_logger.isDebugEnabled())
{
_logger.debug(
"Routing process provided no queues to enqueue the message on, must discard message as unable to DLQ: delivery tag: "
+ deliveryTag);
}
messageWithSubject(ChannelMessages.DISCARDMSG_NOROUTE(msg.getMessageNumber(),
altExchange.getName()));
}
}
}
}
}
public void recordFuture(final ListenableFuture<Void> future, final ServerTransaction.Action action)
{
_unfinishedCommandsQueue.add(new AsyncCommand(future, action));
}
public void sync()
{
if(_logger.isDebugEnabled())
{
_logger.debug("sync() called on channel " + debugIdentity());
}
AsyncCommand cmd;
while((cmd = _unfinishedCommandsQueue.poll()) != null)
{
cmd.complete();
}
if(_transaction instanceof LocalTransaction)
{
((LocalTransaction)_transaction).sync();
}
}
private static class AsyncCommand
{
private final ListenableFuture<Void> _future;
private ServerTransaction.Action _action;
public AsyncCommand(final ListenableFuture<Void> future, final ServerTransaction.Action action)
{
_future = future;
_action = action;
}
void complete()
{
boolean interrupted = false;
try
{
while (true)
{
try
{
_future.get();
break;
}
catch (InterruptedException e)
{
interrupted = true;
}
}
}
catch(ExecutionException e)
{
if(e.getCause() instanceof RuntimeException)
{
throw (RuntimeException)e.getCause();
}
else if(e.getCause() instanceof Error)
{
throw (Error) e.getCause();
}
else
{
throw new ServerScopedRuntimeException(e.getCause());
}
}
if(interrupted)
{
Thread.currentThread().interrupt();
}
_action.postCommit();
_action = null;
}
}
@Override
public int getConsumerCount()
{
return _tag2SubscriptionTargetMap.size();
}
@Override
public Collection<Consumer<?>> getConsumers()
{
return Collections.unmodifiableCollection(_consumers);
}
private class ConsumerClosedListener implements ConfigurationChangeListener
{
@Override
public void stateChanged(final ConfiguredObject object, final State oldState, final State newState)
{
if(newState == State.DELETED)
{
consumerRemoved((Consumer<?>)object);
}
}
@Override
public void childAdded(final ConfiguredObject object, final ConfiguredObject child)
{
}
@Override
public void childRemoved(final ConfiguredObject object, final ConfiguredObject child)
{
}
@Override
public void attributeSet(final ConfiguredObject object,
final String attributeName,
final Object oldAttributeValue,
final Object newAttributeValue)
{
}
@Override
public void bulkChangeStart(final ConfiguredObject<?> object)
{
}
@Override
public void bulkChangeEnd(final ConfiguredObject<?> object)
{
}
}
private void consumerAdded(final Consumer<?> consumer)
{
for(ConsumerListener l : _consumerListeners)
{
l.consumerAdded(consumer);
}
}
private void consumerRemoved(final Consumer<?> consumer)
{
for(ConsumerListener l : _consumerListeners)
{
l.consumerRemoved(consumer);
}
}
@Override
public void addConsumerListener(ConsumerListener listener)
{
_consumerListeners.add(listener);
}
@Override
public void removeConsumerListener(ConsumerListener listener)
{
_consumerListeners.remove(listener);
}
@Override
public void setModelObject(final Session<?> session)
{
_modelObject = session;
}
@Override
public Session<?> getModelObject()
{
return _modelObject;
}
@Override
public long getTransactionStartTime()
{
ServerTransaction serverTransaction = _transaction;
if (serverTransaction.isTransactional())
{
return serverTransaction.getTransactionStartTime();
}
else
{
return 0L;
}
}
@Override
public long getTransactionUpdateTime()
{
ServerTransaction serverTransaction = _transaction;
if (serverTransaction.isTransactional())
{
return serverTransaction.getTransactionUpdateTime();
}
else
{
return 0L;
}
}
@Override
public void receiveAccessRequest(final AMQShortString realm,
final boolean exclusive,
final boolean passive,
final boolean active, final boolean write, final boolean read)
{
if(_logger.isDebugEnabled())
{
_logger.debug("RECV[" + _channelId + "] AccessRequest[" +" realm: " + realm +
" exclusive: " + exclusive +
" passive: " + passive +
" active: " + active +
" write: " + write + " read: " + read + " ]");
}
MethodRegistry methodRegistry = _connection.getMethodRegistry();
if (ProtocolVersion.v0_91.equals(_connection.getProtocolVersion()))
{
_connection.sendConnectionClose(AMQConstant.COMMAND_INVALID,
"AccessRequest not present in AMQP versions other than 0-8, 0-9",
_channelId);
}
else
{
// We don't implement access control class, but to keep clients happy that expect it
// always use the "0" ticket.
AccessRequestOkBody response = methodRegistry.createAccessRequestOkBody(0);
sync();
_connection.writeFrame(response.generateFrame(_channelId));
}
}
@Override
public void receiveBasicAck(final long deliveryTag, final boolean multiple)
{
if(_logger.isDebugEnabled())
{
_logger.debug("RECV[" + _channelId + "] BasicAck[" +" deliveryTag: " + deliveryTag + " multiple: " + multiple + " ]");
}
acknowledgeMessage(deliveryTag, multiple);
}
@Override
public void receiveBasicCancel(final AMQShortString consumerTag, final boolean nowait)
{
if(_logger.isDebugEnabled())
{
_logger.debug("RECV[" + _channelId + "] BasicCancel[" +" consumerTag: " + consumerTag + " noWait: " + nowait + " ]");
}
unsubscribeConsumer(consumerTag);
if (!nowait)
{
MethodRegistry methodRegistry = _connection.getMethodRegistry();
BasicCancelOkBody cancelOkBody = methodRegistry.createBasicCancelOkBody(consumerTag);
sync();
_connection.writeFrame(cancelOkBody.generateFrame(_channelId));
}
}
@Override
public void receiveBasicConsume(final AMQShortString queue,
final AMQShortString consumerTag,
final boolean noLocal,
final boolean noAck,
final boolean exclusive, final boolean nowait, final FieldTable arguments)
{
if(_logger.isDebugEnabled())
{
_logger.debug("RECV[" + _channelId + "] BasicConsume[" +" queue: " + queue +
" consumerTag: " + consumerTag +
" noLocal: " + noLocal +
" noAck: " + noAck +
" exclusive: " + exclusive + " nowait: " + nowait + " arguments: " + arguments + " ]");
}
AMQShortString consumerTag1 = consumerTag;
NamedAddressSpace vHost = _connection.getAddressSpace();
sync();
String queueName = AMQShortString.toString(queue);
MessageSource queue1 = queueName == null ? getDefaultQueue() : vHost.getAttainedMessageSource(queueName);
final Collection<MessageSource> sources = new HashSet<>();
if (arguments != null && arguments.get("x-multiqueue") instanceof Collection)
{
for (Object object : (Collection<Object>) arguments.get("x-multiqueue"))
{
String sourceName = String.valueOf(object);
sourceName = sourceName.trim();
if (sourceName.length() != 0)
{
MessageSource source = vHost.getAttainedMessageSource(sourceName);
if (source == null)
{
sources.clear();
break;
}
else
{
sources.add(source);
}
}
}
queueName = arguments.get("x-multiqueue").toString();
}
else if (queue1 != null)
{
sources.add(queue1);
}
if (sources.isEmpty())
{
if (_logger.isDebugEnabled())
{
_logger.debug("No queue for '" + queueName + "'");
}
if (queueName != null)
{
closeChannel(AMQConstant.NOT_FOUND, "No such queue, '" + queueName + "'");
}
else
{
_connection.sendConnectionClose(AMQConstant.NOT_ALLOWED,
"No queue name provided, no default queue defined.", _channelId);
}
}
else
{
try
{
consumerTag1 = consumeFromSource(consumerTag1,
sources,
!noAck,
arguments,
exclusive,
noLocal);
if (!nowait)
{
MethodRegistry methodRegistry = _connection.getMethodRegistry();
AMQMethodBody responseBody = methodRegistry.createBasicConsumeOkBody(consumerTag1);
_connection.writeFrame(responseBody.generateFrame(_channelId));
}
}
catch (ConsumerTagInUseException cte)
{
_connection.sendConnectionClose(AMQConstant.NOT_ALLOWED,
"Non-unique consumer tag, '" + consumerTag1
+ "'", _channelId);
}
catch (AMQInvalidArgumentException ise)
{
_connection.sendConnectionClose(AMQConstant.ARGUMENT_INVALID, ise.getMessage(), _channelId);
}
catch (Queue.ExistingExclusiveConsumer e)
{
_connection.sendConnectionClose(AMQConstant.ACCESS_REFUSED,
"Cannot subscribe to queue '"
+ queue1.getName()
+ "' as it already has an existing exclusive consumer", _channelId);
}
catch (Queue.ExistingConsumerPreventsExclusive e)
{
_connection.sendConnectionClose(AMQConstant.ACCESS_REFUSED,
"Cannot subscribe to queue '"
+ queue1.getName()
+ "' exclusively as it already has a consumer", _channelId);
}
catch (AccessControlException e)
{
_connection.sendConnectionClose(AMQConstant.ACCESS_REFUSED, "Cannot subscribe to queue '"
+ queue1.getName()
+ "' permission denied", _channelId);
}
catch (MessageSource.ConsumerAccessRefused consumerAccessRefused)
{
_connection.sendConnectionClose(AMQConstant.ACCESS_REFUSED,
"Cannot subscribe to queue '"
+ queue1.getName()
+ "' as it already has an incompatible exclusivity policy", _channelId);
}
}
}
@Override
public void receiveBasicGet(final AMQShortString queueName, final boolean noAck)
{
if(_logger.isDebugEnabled())
{
_logger.debug("RECV[" + _channelId + "] BasicGet[" +" queue: " + queueName + " noAck: " + noAck + " ]");
}
NamedAddressSpace vHost = _connection.getAddressSpace();
sync();
MessageSource queue = queueName == null ? getDefaultQueue() : vHost.getAttainedMessageSource(queueName.toString());
if (queue == null)
{
if (_logger.isDebugEnabled())
{
_logger.debug("No queue for '" + queueName + "'");
}
if (queueName != null)
{
_connection.sendConnectionClose(AMQConstant.NOT_FOUND, "No such queue, '" + queueName + "'", _channelId);
}
else
{
_connection.sendConnectionClose(AMQConstant.NOT_ALLOWED,
"No queue name provided, no default queue defined.", _channelId);
}
}
else
{
try
{
if (!performGet(queue, !noAck))
{
MethodRegistry methodRegistry = _connection.getMethodRegistry();
BasicGetEmptyBody responseBody = methodRegistry.createBasicGetEmptyBody(null);
_connection.writeFrame(responseBody.generateFrame(_channelId));
}
}
catch (AccessControlException e)
{
_connection.sendConnectionClose(AMQConstant.ACCESS_REFUSED, e.getMessage(), _channelId);
}
catch (MessageSource.ExistingExclusiveConsumer e)
{
_connection.sendConnectionClose(AMQConstant.NOT_ALLOWED, "Queue has an exclusive consumer", _channelId);
}
catch (MessageSource.ExistingConsumerPreventsExclusive e)
{
_connection.sendConnectionClose(AMQConstant.INTERNAL_ERROR,
"The GET request has been evaluated as an exclusive consumer, " +
"this is likely due to a programming error in the Qpid broker", _channelId);
}
catch (MessageSource.ConsumerAccessRefused consumerAccessRefused)
{
_connection.sendConnectionClose(AMQConstant.NOT_ALLOWED,
"Queue has an incompatible exclusivity policy", _channelId);
}
}
}
@Override
public void receiveBasicPublish(final AMQShortString exchangeName,
final AMQShortString routingKey,
final boolean mandatory,
final boolean immediate)
{
if(_logger.isDebugEnabled())
{
_logger.debug("RECV[" + _channelId + "] BasicPublish[" +" exchange: " + exchangeName +
" routingKey: " + routingKey +
" mandatory: " + mandatory +
" immediate: " + immediate + " ]");
}
NamedAddressSpace vHost = _connection.getAddressSpace();
if(blockingTimeoutExceeded())
{
message(ChannelMessages.FLOW_CONTROL_IGNORED());
closeChannel(AMQConstant.MESSAGE_TOO_LARGE,
"Channel flow control was requested, but not enforced by sender");
}
else
{
MessageDestination destination;
if (isDefaultExchange(exchangeName))
{
destination = vHost.getDefaultDestination();
}
else
{
destination = vHost.getAttainedMessageDestination(exchangeName.toString());
}
// if the exchange does not exist we raise a channel exception
if (destination == null)
{
closeChannel(AMQConstant.NOT_FOUND, "Unknown exchange name: '" + exchangeName + "'");
}
else
{
MessagePublishInfo info = new MessagePublishInfo(exchangeName,
immediate,
mandatory,
routingKey);
try
{
setPublishFrame(info, destination);
}
catch (AccessControlException e)
{
_connection.sendConnectionClose(AMQConstant.ACCESS_REFUSED, e.getMessage(), getChannelId());
}
}
}
}
@Override
public EventLogger getEventLogger()
{
return getConnection().getEventLogger();
}
private boolean blockingTimeoutExceeded()
{
return _wireBlockingState && (System.currentTimeMillis() - _blockTime) > _blockingTimeout;
}
@Override
public void receiveBasicQos(final long prefetchSize, final int prefetchCount, final boolean global)
{
if(_logger.isDebugEnabled())
{
_logger.debug("RECV[" + _channelId + "] BasicQos[" +" prefetchSize: " + prefetchSize + " prefetchCount: " + prefetchCount + " global: " + global + " ]");
}
sync();
setCredit(prefetchSize, prefetchCount);
MethodRegistry methodRegistry = _connection.getMethodRegistry();
AMQMethodBody responseBody = methodRegistry.createBasicQosOkBody();
_connection.writeFrame(responseBody.generateFrame(getChannelId()));
}
@Override
public void receiveBasicRecover(final boolean requeue, final boolean sync)
{
if(_logger.isDebugEnabled())
{
_logger.debug("RECV[" + _channelId + "] BasicRecover[" + " requeue: " + requeue + " sync: " + sync + " ]");
}
resend();
if (sync)
{
MethodRegistry methodRegistry = _connection.getMethodRegistry();
AMQMethodBody recoverOk = methodRegistry.createBasicRecoverSyncOkBody();
sync();
_connection.writeFrame(recoverOk.generateFrame(getChannelId()));
}
}
@Override
public void receiveBasicReject(final long deliveryTag, final boolean requeue)
{
if(_logger.isDebugEnabled())
{
_logger.debug("RECV[" + _channelId + "] BasicReject[" +" deliveryTag: " + deliveryTag + " requeue: " + requeue + " ]");
}
MessageInstance message = getUnacknowledgedMessageMap().get(deliveryTag);
if (message == null)
{
_logger.warn("Dropping reject request as message is null for tag:" + deliveryTag);
}
else
{
if (message.getMessage() == null)
{
_logger.warn("Message has already been purged, unable to Reject.");
}
else
{
if (_logger.isDebugEnabled())
{
_logger.debug("Rejecting: DT:" + deliveryTag
+ "-" + message.getMessage() +
": Requeue:" + requeue
+
" on channel:" + debugIdentity());
}
if (requeue)
{
message.decrementDeliveryCount();
requeue(deliveryTag);
}
else
{
// Since the JMS client abuses the reject flag for requeing after rollback, we won't set reject here
// as it would prevent redelivery
// message.reject();
final boolean maxDeliveryCountEnabled = isMaxDeliveryCountEnabled(deliveryTag);
if (_logger.isDebugEnabled())
{
_logger.debug("maxDeliveryCountEnabled: "
+ maxDeliveryCountEnabled
+ " deliveryTag "
+ deliveryTag);
}
if (maxDeliveryCountEnabled)
{
final boolean deliveredTooManyTimes = isDeliveredTooManyTimes(deliveryTag);
if (_logger.isDebugEnabled())
{
_logger.debug("deliveredTooManyTimes: "
+ deliveredTooManyTimes
+ " deliveryTag "
+ deliveryTag);
}
if (deliveredTooManyTimes)
{
deadLetter(deliveryTag);
}
else
{
//this requeue represents a message rejected because of a recover/rollback that we
//are not ready to DLQ. We rely on the reject command to resend from the unacked map
//and therefore need to increment the delivery counter so we cancel out the effect
//of the AMQChannel#resend() decrement.
message.incrementDeliveryCount();
}
}
else
{
requeue(deliveryTag);
}
}
}
}
}
@Override
public void receiveChannelClose(final int replyCode,
final AMQShortString replyText,
final int classId,
final int methodId)
{
if(_logger.isDebugEnabled())
{
_logger.debug("RECV[" + _channelId + "] ChannelClose[" +" replyCode: " + replyCode + " replyText: " + replyText + " classId: " + classId + " methodId: " + methodId + " ]");
}
sync();
_connection.closeChannel(this);
_connection.writeFrame(new AMQFrame(getChannelId(),
_connection.getMethodRegistry().createChannelCloseOkBody()));
}
@Override
public void receiveChannelCloseOk()
{
if(_logger.isDebugEnabled())
{
_logger.debug("RECV[" + _channelId + "] ChannelCloseOk");
}
_connection.closeChannelOk(getChannelId());
}
@Override
public void receiveMessageContent(final QpidByteBuffer data)
{
if(_logger.isDebugEnabled())
{
_logger.debug("RECV[" + _channelId + "] MessageContent[" +" data: " + hex(data,_connection.getBinaryDataLimit()) + " ] ");
}
if(hasCurrentMessage())
{
publishContentBody(new ContentBody(data));
}
else
{
_connection.sendConnectionClose(AMQConstant.COMMAND_INVALID,
"Attempt to send a content header without first sending a publish frame",
_channelId);
}
}
@Override
public void receiveMessageHeader(final BasicContentHeaderProperties properties, final long bodySize)
{
if(_logger.isDebugEnabled())
{
_logger.debug("RECV[" + _channelId + "] MessageHeader[ properties: {" + properties + "} bodySize: " + bodySize + " ]");
}
if(hasCurrentMessage())
{
if(bodySize > _connection.getMaxMessageSize())
{
closeChannel(AMQConstant.MESSAGE_TOO_LARGE,
"Message size of " + bodySize + " greater than allowed maximum of " + _connection.getMaxMessageSize());
}
publishContentHeader(new ContentHeaderBody(properties, bodySize));
}
else
{
_connection.sendConnectionClose(AMQConstant.COMMAND_INVALID,
"Attempt to send a content header without first sending a publish frame",
_channelId);
}
}
@Override
public boolean ignoreAllButCloseOk()
{
return _connection.ignoreAllButCloseOk() || _connection.channelAwaitingClosure(_channelId);
}
@Override
public void receiveBasicNack(final long deliveryTag, final boolean multiple, final boolean requeue)
{
if(_logger.isDebugEnabled())
{
_logger.debug("RECV[" + _channelId + "] BasicNack[" +" deliveryTag: " + deliveryTag + " multiple: " + multiple + " requeue: " + requeue + " ]");
}
Map<Long, MessageInstance> nackedMessageMap = new LinkedHashMap<>();
_unacknowledgedMessageMap.collect(deliveryTag, multiple, nackedMessageMap);
for(MessageInstance message : nackedMessageMap.values())
{
if (message == null)
{
_logger.warn("Ignoring nack request as message is null for tag:" + deliveryTag);
}
else
{
if (message.getMessage() == null)
{
_logger.warn("Message has already been purged, unable to nack.");
}
else
{
if (_logger.isDebugEnabled())
{
_logger.debug("Nack-ing: DT:" + deliveryTag
+ "-" + message.getMessage() +
": Requeue:" + requeue
+
" on channel:" + debugIdentity());
}
if (requeue)
{
message.decrementDeliveryCount();
requeue(deliveryTag);
}
else
{
message.reject();
final boolean maxDeliveryCountEnabled = isMaxDeliveryCountEnabled(deliveryTag);
if (_logger.isDebugEnabled())
{
_logger.debug("maxDeliveryCountEnabled: "
+ maxDeliveryCountEnabled
+ " deliveryTag "
+ deliveryTag);
}
if (maxDeliveryCountEnabled)
{
final boolean deliveredTooManyTimes = isDeliveredTooManyTimes(deliveryTag);
if (_logger.isDebugEnabled())
{
_logger.debug("deliveredTooManyTimes: "
+ deliveredTooManyTimes
+ " deliveryTag "
+ deliveryTag);
}
if (deliveredTooManyTimes)
{
deadLetter(deliveryTag);
}
else
{
message.incrementDeliveryCount();
}
}
else
{
requeue(deliveryTag);
}
}
}
}
}
}
@Override
public void receiveChannelFlow(final boolean active)
{
if(_logger.isDebugEnabled())
{
_logger.debug("RECV[" + _channelId + "] ChannelFlow[" +" active: " + active + " ]");
}
sync();
setSuspended(!active);
MethodRegistry methodRegistry = _connection.getMethodRegistry();
AMQMethodBody responseBody = methodRegistry.createChannelFlowOkBody(active);
_connection.writeFrame(responseBody.generateFrame(getChannelId()));
}
@Override
public void receiveChannelFlowOk(final boolean active)
{
if(_logger.isDebugEnabled())
{
_logger.debug("RECV[" + _channelId + "] ChannelFlowOk[" +" active: " + active + " ]");
}
// TODO - should we do anything here?
}
@Override
public void receiveExchangeBound(final AMQShortString exchangeName,
final AMQShortString routingKey,
final AMQShortString queueName)
{
if(_logger.isDebugEnabled())
{
_logger.debug("RECV[" + _channelId + "] ExchangeBound[" +" exchange: " + exchangeName + " routingKey: " +
routingKey + " queue: " + queueName + " ]");
}
NamedAddressSpace virtualHost = _connection.getAddressSpace();
MethodRegistry methodRegistry = _connection.getMethodRegistry();
sync();
int replyCode;
String replyText;
if (isDefaultExchange(exchangeName))
{
if (routingKey == null)
{
if (queueName == null)
{
replyCode = virtualHost.hasMessageSources()
? ExchangeBoundOkBody.OK
: ExchangeBoundOkBody.NO_BINDINGS;
replyText = null;
}
else
{
MessageSource queue = virtualHost.getAttainedMessageSource(queueName.toString());
if (queue == null)
{
replyCode = ExchangeBoundOkBody.QUEUE_NOT_FOUND;
replyText = "Queue '" + queueName + "' not found";
}
else
{
replyCode = ExchangeBoundOkBody.OK;
replyText = null;
}
}
}
else
{
if (queueName == null)
{
replyCode = virtualHost.getAttainedMessageDestination(routingKey.toString()) instanceof Queue
? ExchangeBoundOkBody.OK
: ExchangeBoundOkBody.NO_QUEUE_BOUND_WITH_RK;
replyText = null;
}
else
{
MessageDestination destination = virtualHost.getAttainedMessageDestination(queueName.toString());
Queue<?> queue = destination instanceof Queue ? (Queue) destination : null;
if (queue == null)
{
replyCode = ExchangeBoundOkBody.QUEUE_NOT_FOUND;
replyText = "Queue '" + queueName + "' not found";
}
else
{
replyCode = queueName.equals(routingKey)
? ExchangeBoundOkBody.OK
: ExchangeBoundOkBody.SPECIFIC_QUEUE_NOT_BOUND_WITH_RK;
replyText = null;
}
}
}
}
else
{
Exchange<?> exchange = getExchange(exchangeName.toString());
if (exchange == null)
{
replyCode = ExchangeBoundOkBody.EXCHANGE_NOT_FOUND;
replyText = "Exchange '" + exchangeName + "' not found";
}
else if (routingKey == null)
{
if (queueName == null)
{
if (exchange.hasBindings())
{
replyCode = ExchangeBoundOkBody.OK;
replyText = null;
}
else
{
replyCode = ExchangeBoundOkBody.NO_BINDINGS;
replyText = null;
}
}
else
{
Queue<?> queue = getQueue(queueName.toString());
if (queue == null)
{
replyCode = ExchangeBoundOkBody.QUEUE_NOT_FOUND;
replyText = "Queue '" + queueName + "' not found";
}
else
{
if (exchange.isBound(queue))
{
replyCode = ExchangeBoundOkBody.OK;
replyText = null;
}
else
{
replyCode = ExchangeBoundOkBody.QUEUE_NOT_BOUND;
replyText = "Queue '"
+ queueName
+ "' not bound to exchange '"
+ exchangeName
+ "'";
}
}
}
}
else if (queueName != null)
{
Queue<?> queue = getQueue(queueName.toString());
if (queue == null)
{
replyCode = ExchangeBoundOkBody.QUEUE_NOT_FOUND;
replyText = "Queue '" + queueName + "' not found";
}
else
{
String bindingKey = routingKey == null ? null : routingKey.toString();
if (exchange.isBound(bindingKey, queue))
{
replyCode = ExchangeBoundOkBody.OK;
replyText = null;
}
else
{
replyCode = ExchangeBoundOkBody.SPECIFIC_QUEUE_NOT_BOUND_WITH_RK;
replyText = "Queue '" + queueName + "' not bound with routing key '" +
routingKey + "' to exchange '" + exchangeName + "'";
}
}
}
else
{
if (exchange.isBound(routingKey == null ? "" : routingKey.toString()))
{
replyCode = ExchangeBoundOkBody.OK;
replyText = null;
}
else
{
replyCode = ExchangeBoundOkBody.NO_QUEUE_BOUND_WITH_RK;
replyText =
"No queue bound with routing key '" + routingKey + "' to exchange '" + exchangeName + "'";
}
}
}
ExchangeBoundOkBody exchangeBoundOkBody =
methodRegistry.createExchangeBoundOkBody(replyCode, AMQShortString.validValueOf(replyText));
_connection.writeFrame(exchangeBoundOkBody.generateFrame(getChannelId()));
}
@Override
public void receiveExchangeDeclare(final AMQShortString exchangeName,
final AMQShortString type,
final boolean passive,
final boolean durable,
final boolean autoDelete,
final boolean internal,
final boolean nowait,
final FieldTable arguments)
{
if(_logger.isDebugEnabled())
{
_logger.debug("RECV[" + _channelId + "] ExchangeDeclare[" +" exchange: " + exchangeName +
" type: " + type +
" passive: " + passive +
" durable: " + durable +
" autoDelete: " + autoDelete +
" internal: " + internal + " nowait: " + nowait + " arguments: " + arguments + " ]");
}
final MethodRegistry methodRegistry = _connection.getMethodRegistry();
final AMQMethodBody declareOkBody = methodRegistry.createExchangeDeclareOkBody();
Exchange<?> exchange;
NamedAddressSpace virtualHost = _connection.getAddressSpace();
if (isDefaultExchange(exchangeName))
{
if (!new AMQShortString(ExchangeDefaults.DIRECT_EXCHANGE_CLASS).equals(type))
{
_connection.sendConnectionClose(AMQConstant.NOT_ALLOWED, "Attempt to redeclare default exchange: "
+ " of type "
+ ExchangeDefaults.DIRECT_EXCHANGE_CLASS
+ " to " + type + ".", getChannelId());
}
else if (!nowait)
{
sync();
_connection.writeFrame(declareOkBody.generateFrame(getChannelId()));
}
}
else
{
if (passive)
{
exchange = getExchange(exchangeName.toString());
if (exchange == null)
{
closeChannel(AMQConstant.NOT_FOUND, "Unknown exchange: '" + exchangeName + "'");
}
else if (!(type == null || type.length() == 0) && !exchange.getType().equals(type.toString()))
{
_connection.sendConnectionClose(AMQConstant.NOT_ALLOWED, "Attempt to redeclare exchange: '"
+ exchangeName
+ "' of type "
+ exchange.getType()
+ " to "
+ type
+ ".", getChannelId());
}
else if (!nowait)
{
sync();
_connection.writeFrame(declareOkBody.generateFrame(getChannelId()));
}
}
else
{
String name = exchangeName.toString();
String typeString = type == null ? null : type.toString();
try
{
Map<String, Object> attributes = new HashMap<String, Object>();
if (arguments != null)
{
attributes.putAll(FieldTable.convertToMap(arguments));
}
attributes.put(Exchange.NAME, name);
attributes.put(Exchange.TYPE, typeString);
attributes.put(Exchange.DURABLE, durable);
attributes.put(Exchange.LIFETIME_POLICY,
autoDelete ? LifetimePolicy.DELETE_ON_NO_LINKS : LifetimePolicy.PERMANENT);
if (!attributes.containsKey(Exchange.ALTERNATE_EXCHANGE))
{
attributes.put(Exchange.ALTERNATE_EXCHANGE, null);
}
exchange = virtualHost.createMessageDestination(Exchange.class, attributes);
if (!nowait)
{
sync();
_connection.writeFrame(declareOkBody.generateFrame(getChannelId()));
}
}
catch (ReservedExchangeNameException e)
{
Exchange existing = getExchange(name);
if (existing == null || !existing.getType().equals(typeString))
{
_connection.sendConnectionClose(AMQConstant.NOT_ALLOWED,
"Attempt to declare exchange: '" + exchangeName +
"' which begins with reserved prefix.", getChannelId());
}
else if(!nowait)
{
sync();
_connection.writeFrame(declareOkBody.generateFrame(getChannelId()));
}
}
catch (ExchangeExistsException e)
{
exchange = e.getExistingExchange();
if (!exchange.getType().equals(typeString))
{
_connection.sendConnectionClose(AMQConstant.NOT_ALLOWED, "Attempt to redeclare exchange: '"
+ exchangeName + "' of type "
+ exchange.getType()
+ " to " + type + ".", getChannelId());
}
else
{
if (!nowait)
{
sync();
_connection.writeFrame(declareOkBody.generateFrame(getChannelId()));
}
}
}
catch (NoFactoryForTypeException e)
{
_connection.sendConnectionClose(AMQConstant.COMMAND_INVALID, "Unknown exchange type '"
+ e.getType()
+ "' for exchange '"
+ exchangeName
+ "'", getChannelId());
}
catch (AccessControlException e)
{
_connection.sendConnectionClose(AMQConstant.ACCESS_REFUSED, e.getMessage(), getChannelId());
}
catch (UnknownConfiguredObjectException e)
{
// note - since 0-8/9/9-1 can't set the alt. exchange this exception should never occur
final String message = "Unknown alternate exchange "
+ (e.getName() != null
? "name: '" + e.getName() + "'"
: "id: " + e.getId());
_connection.sendConnectionClose(AMQConstant.NOT_FOUND, message, getChannelId());
}
catch (IllegalArgumentException e)
{
_connection.sendConnectionClose(AMQConstant.COMMAND_INVALID, "Error creating exchange '"
+ exchangeName
+ "': "
+ e.getMessage(), getChannelId());
}
}
}
}
@Override
public void receiveExchangeDelete(final AMQShortString exchangeStr, final boolean ifUnused, final boolean nowait)
{
if(_logger.isDebugEnabled())
{
_logger.debug("RECV[" + _channelId + "] ExchangeDelete[" +" exchange: " + exchangeStr + " ifUnused: " + ifUnused + " nowait: " + nowait + " ]");
}
NamedAddressSpace virtualHost = _connection.getAddressSpace();
sync();
if (isDefaultExchange(exchangeStr))
{
_connection.sendConnectionClose(AMQConstant.NOT_ALLOWED,
"Default Exchange cannot be deleted", getChannelId());
}
else
{
final String exchangeName = exchangeStr.toString();
final Exchange<?> exchange = getExchange(exchangeName);
if (exchange == null)
{
closeChannel(AMQConstant.NOT_FOUND, "No such exchange: '" + exchangeStr + "'");
}
else
{
if (ifUnused && exchange.hasBindings())
{
closeChannel(AMQConstant.IN_USE, "Exchange has bindings");
}
else
{
try
{
exchange.delete();
if (!nowait)
{
ExchangeDeleteOkBody responseBody = _connection.getMethodRegistry().createExchangeDeleteOkBody();
_connection.writeFrame(responseBody.generateFrame(getChannelId()));
}
}
catch (ExchangeIsAlternateException e)
{
closeChannel(AMQConstant.NOT_ALLOWED, "Exchange in use as an alternate exchange");
}
catch (RequiredExchangeException e)
{
closeChannel(AMQConstant.NOT_ALLOWED, "Exchange '" + exchangeStr + "' cannot be deleted");
}
catch (AccessControlException e)
{
_connection.sendConnectionClose(AMQConstant.ACCESS_REFUSED, e.getMessage(), getChannelId());
}
}
}
}
}
@Override
public void receiveQueueBind(final AMQShortString queueName,
final AMQShortString exchange,
AMQShortString routingKey,
final boolean nowait,
final FieldTable argumentsTable)
{
if(_logger.isDebugEnabled())
{
_logger.debug("RECV[" + _channelId + "] QueueBind[" +" queue: " + queueName +
" exchange: " + exchange +
" bindingKey: " + routingKey +
" nowait: " + nowait + " arguments: " + argumentsTable + " ]");
}
NamedAddressSpace virtualHost = _connection.getAddressSpace();
Queue<?> queue;
if (queueName == null)
{
queue = getDefaultQueue();
if (queue != null)
{
if (routingKey == null)
{
routingKey = AMQShortString.valueOf(queue.getName());
}
}
}
else
{
queue = getQueue(queueName.toString());
routingKey = routingKey == null ? AMQShortString.EMPTY_STRING : routingKey;
}
if (queue == null)
{
String message = queueName == null
? "No default queue defined on channel and queue was null"
: "Queue " + queueName + " does not exist.";
closeChannel(AMQConstant.NOT_FOUND, message);
}
else if (isDefaultExchange(exchange))
{
_connection.sendConnectionClose(AMQConstant.NOT_ALLOWED,
"Cannot bind the queue '" + queueName + "' to the default exchange", getChannelId());
}
else
{
final String exchangeName = exchange.toString();
final Exchange<?> exch = getExchange(exchangeName);
if (exch == null)
{
closeChannel(AMQConstant.NOT_FOUND,
"Exchange '" + exchangeName + "' does not exist.");
}
else
{
try
{
Map<String, Object> arguments = FieldTable.convertToMap(argumentsTable);
String bindingKey = String.valueOf(routingKey);
if (!exch.isBound(bindingKey, arguments, queue))
{
if (!exch.addBinding(bindingKey, queue, arguments)
&& ExchangeDefaults.TOPIC_EXCHANGE_CLASS.equals(
exch.getType()))
{
exch.replaceBinding(bindingKey, queue, arguments);
}
}
if (_logger.isDebugEnabled())
{
_logger.debug("Binding queue "
+ queue
+ " to exchange "
+ exch
+ " with routing key "
+ routingKey);
}
if (!nowait)
{
sync();
MethodRegistry methodRegistry = _connection.getMethodRegistry();
AMQMethodBody responseBody = methodRegistry.createQueueBindOkBody();
_connection.writeFrame(responseBody.generateFrame(getChannelId()));
}
}
catch (AccessControlException e)
{
_connection.sendConnectionClose(AMQConstant.ACCESS_REFUSED, e.getMessage(), getChannelId());
}
}
}
}
@Override
public void receiveQueueDeclare(final AMQShortString queueStr,
final boolean passive,
final boolean durable,
final boolean exclusive,
final boolean autoDelete,
final boolean nowait,
final FieldTable arguments)
{
if(_logger.isDebugEnabled())
{
_logger.debug("RECV[" + _channelId + "] QueueDeclare[" +" queue: " + queueStr +
" passive: " + passive +
" durable: " + durable +
" exclusive: " + exclusive +
" autoDelete: " + autoDelete + " nowait: " + nowait + " arguments: " + arguments + " ]");
}
NamedAddressSpace virtualHost = _connection.getAddressSpace();
final AMQShortString queueName;
// if we aren't given a queue name, we create one which we return to the client
if ((queueStr == null) || (queueStr.length() == 0))
{
queueName = new AMQShortString("tmp_" + UUID.randomUUID());
}
else
{
queueName = queueStr;
}
Queue<?> queue;
//TODO: do we need to check that the queue already exists with exactly the same "configuration"?
if (passive)
{
queue = getQueue(queueName.toString());
if (queue == null)
{
closeChannel(AMQConstant.NOT_FOUND,
"Queue: '"
+ queueName
+ "' not found on VirtualHost '"
+ virtualHost.getName()
+ "'.");
}
else
{
if (!queue.verifySessionAccess(this))
{
_connection.sendConnectionClose(AMQConstant.NOT_ALLOWED, "Queue '"
+ queue.getName()
+ "' is exclusive, but not created on this Connection.", getChannelId());
}
else
{
//set this as the default queue on the channel:
setDefaultQueue(queue);
if (!nowait)
{
sync();
MethodRegistry methodRegistry = _connection.getMethodRegistry();
QueueDeclareOkBody responseBody =
methodRegistry.createQueueDeclareOkBody(queueName,
queue.getQueueDepthMessages(),
queue.getConsumerCount());
_connection.writeFrame(responseBody.generateFrame(getChannelId()));
if (_logger.isDebugEnabled())
{
_logger.debug("Queue " + queueName + " declared successfully");
}
}
}
}
}
else
{
try
{
Map<String, Object> attributes =
QueueArgumentsConverter.convertWireArgsToModel(FieldTable.convertToMap(arguments));
final String queueNameString = AMQShortString.toString(queueName);
attributes.put(Queue.NAME, queueNameString);
attributes.put(Queue.DURABLE, durable);
LifetimePolicy lifetimePolicy;
ExclusivityPolicy exclusivityPolicy;
if (exclusive)
{
lifetimePolicy = autoDelete
? LifetimePolicy.DELETE_ON_NO_OUTBOUND_LINKS
: durable ? LifetimePolicy.PERMANENT : LifetimePolicy.DELETE_ON_CONNECTION_CLOSE;
exclusivityPolicy = durable ? ExclusivityPolicy.CONTAINER : ExclusivityPolicy.CONNECTION;
}
else
{
lifetimePolicy = autoDelete ? LifetimePolicy.DELETE_ON_NO_OUTBOUND_LINKS : LifetimePolicy.PERMANENT;
exclusivityPolicy = ExclusivityPolicy.NONE;
}
if(!attributes.containsKey(Queue.EXCLUSIVE))
{
attributes.put(Queue.EXCLUSIVE, exclusivityPolicy);
}
if(!attributes.containsKey(Queue.LIFETIME_POLICY))
{
attributes.put(Queue.LIFETIME_POLICY, lifetimePolicy);
}
queue = virtualHost.createMessageSource(Queue.class, attributes);
setDefaultQueue(queue);
if (!nowait)
{
sync();
MethodRegistry methodRegistry = _connection.getMethodRegistry();
QueueDeclareOkBody responseBody =
methodRegistry.createQueueDeclareOkBody(queueName,
queue.getQueueDepthMessages(),
queue.getConsumerCount());
_connection.writeFrame(responseBody.generateFrame(getChannelId()));
if (_logger.isDebugEnabled())
{
_logger.debug("Queue " + queueName + " declared successfully");
}
}
}
catch (QueueExistsException qe)
{
queue = qe.getExistingQueue();
if (!queue.verifySessionAccess(this))
{
_connection.sendConnectionClose(AMQConstant.NOT_ALLOWED, "Queue '"
+ queue.getName()
+ "' is exclusive, but not created on this Connection.", getChannelId());
}
else if (queue.isExclusive() != exclusive)
{
closeChannel(AMQConstant.ALREADY_EXISTS,
"Cannot re-declare queue '"
+ queue.getName()
+ "' with different exclusivity (was: "
+ queue.isExclusive()
+ " requested "
+ exclusive
+ ")");
}
else if ((autoDelete
&& queue.getLifetimePolicy() == LifetimePolicy.PERMANENT)
|| (!autoDelete && queue.getLifetimePolicy() != ((exclusive
&& !durable)
? LifetimePolicy.DELETE_ON_CONNECTION_CLOSE
: LifetimePolicy.PERMANENT)))
{
closeChannel(AMQConstant.ALREADY_EXISTS,
"Cannot re-declare queue '"
+ queue.getName()
+ "' with different lifetime policy (was: "
+ queue.getLifetimePolicy()
+ " requested autodelete: "
+ autoDelete
+ ")");
}
else
{
setDefaultQueue(queue);
if (!nowait)
{
sync();
MethodRegistry methodRegistry = _connection.getMethodRegistry();
QueueDeclareOkBody responseBody =
methodRegistry.createQueueDeclareOkBody(queueName,
queue.getQueueDepthMessages(),
queue.getConsumerCount());
_connection.writeFrame(responseBody.generateFrame(getChannelId()));
if (_logger.isDebugEnabled())
{
_logger.debug("Queue " + queueName + " declared successfully");
}
}
}
}
catch (AccessControlException e)
{
_connection.sendConnectionClose(AMQConstant.ACCESS_REFUSED, e.getMessage(), getChannelId());
}
}
}
@Override
public void receiveQueueDelete(final AMQShortString queueName,
final boolean ifUnused,
final boolean ifEmpty,
final boolean nowait)
{
if(_logger.isDebugEnabled())
{
_logger.debug("RECV[" + _channelId + "] QueueDelete[" +" queue: " + queueName + " ifUnused: " + ifUnused + " ifEmpty: " + ifEmpty + " nowait: " + nowait + " ]");
}
NamedAddressSpace virtualHost = _connection.getAddressSpace();
sync();
Queue<?> queue;
if (queueName == null)
{
//get the default queue on the channel:
queue = getDefaultQueue();
}
else
{
queue = getQueue(queueName.toString());
}
if (queue == null)
{
closeChannel(AMQConstant.NOT_FOUND, "Queue '" + queueName + "' does not exist.");
}
else
{
if (ifEmpty && !queue.isEmpty())
{
closeChannel(AMQConstant.IN_USE, "Queue: '" + queueName + "' is not empty.");
}
else if (ifUnused && !queue.isUnused())
{
// TODO - Error code
closeChannel(AMQConstant.IN_USE, "Queue: '" + queueName + "' is still used.");
}
else
{
if (!queue.verifySessionAccess(this))
{
_connection.sendConnectionClose(AMQConstant.NOT_ALLOWED, "Queue '"
+ queue.getName()
+ "' is exclusive, but not created on this Connection.", getChannelId());
}
else
{
try
{
int purged = queue.deleteAndReturnCount();
if (!nowait || _connection.isSendQueueDeleteOkRegardless())
{
MethodRegistry methodRegistry = _connection.getMethodRegistry();
QueueDeleteOkBody responseBody = methodRegistry.createQueueDeleteOkBody(purged);
_connection.writeFrame(responseBody.generateFrame(getChannelId()));
}
}
catch (AccessControlException e)
{
_connection.sendConnectionClose(AMQConstant.ACCESS_REFUSED, e.getMessage(), getChannelId());
}
}
}
}
}
@Override
public void receiveQueuePurge(final AMQShortString queueName, final boolean nowait)
{
if(_logger.isDebugEnabled())
{
_logger.debug("RECV[" + _channelId + "] QueuePurge[" +" queue: " + queueName + " nowait: " + nowait + " ]");
}
NamedAddressSpace virtualHost = _connection.getAddressSpace();
Queue<?> queue = null;
if (queueName == null && (queue = getDefaultQueue()) == null)
{
_connection.sendConnectionClose(AMQConstant.NOT_ALLOWED, "No queue specified.", getChannelId());
}
else if ((queueName != null) && (queue = getQueue(queueName.toString())) == null)
{
closeChannel(AMQConstant.NOT_FOUND, "Queue '" + queueName + "' does not exist.");
}
else if (!queue.verifySessionAccess(this))
{
_connection.sendConnectionClose(AMQConstant.NOT_ALLOWED,
"Queue is exclusive, but not created on this Connection.", getChannelId());
}
else
{
try
{
long purged = queue.clearQueue();
if (!nowait)
{
sync();
MethodRegistry methodRegistry = _connection.getMethodRegistry();
AMQMethodBody responseBody = methodRegistry.createQueuePurgeOkBody(purged);
_connection.writeFrame(responseBody.generateFrame(getChannelId()));
}
}
catch (AccessControlException e)
{
_connection.sendConnectionClose(AMQConstant.ACCESS_REFUSED, e.getMessage(), getChannelId());
}
}
}
@Override
public void receiveQueueUnbind(final AMQShortString queueName,
final AMQShortString exchange,
final AMQShortString bindingKey,
final FieldTable arguments)
{
if(_logger.isDebugEnabled())
{
_logger.debug("RECV[" + _channelId + "] QueueUnbind[" +" queue: " + queueName +
" exchange: " + exchange +
" bindingKey: " + bindingKey +
" arguments: " + arguments + " ]");
}
NamedAddressSpace virtualHost = _connection.getAddressSpace();
final boolean useDefaultQueue = queueName == null;
final Queue<?> queue = useDefaultQueue
? getDefaultQueue()
: getQueue(queueName.toString());
if (queue == null)
{
String message = useDefaultQueue
? "No default queue defined on channel and queue was null"
: "Queue '" + queueName + "' does not exist.";
closeChannel(AMQConstant.NOT_FOUND, message);
}
else if (isDefaultExchange(exchange))
{
_connection.sendConnectionClose(AMQConstant.NOT_ALLOWED, "Cannot unbind the queue '"
+ queue.getName()
+ "' from the default exchange", getChannelId());
}
else
{
final Exchange<?> exch = getExchange(exchange.toString());
if (exch == null)
{
closeChannel(AMQConstant.NOT_FOUND, "Exchange '" + exchange + "' does not exist.");
}
else if (!exch.hasBinding(String.valueOf(bindingKey), queue))
{
closeChannel(AMQConstant.NOT_FOUND, "No such binding");
}
else
{
try
{
exch.deleteBinding(String.valueOf(bindingKey), queue);
final AMQMethodBody responseBody = _connection.getMethodRegistry().createQueueUnbindOkBody();
sync();
_connection.writeFrame(responseBody.generateFrame(getChannelId()));
}
catch (AccessControlException e)
{
_connection.sendConnectionClose(AMQConstant.ACCESS_REFUSED, e.getMessage(), getChannelId());
}
}
}
}
@Override
public void receiveTxSelect()
{
if(_logger.isDebugEnabled())
{
_logger.debug("RECV[" + _channelId + "] TxSelect");
}
setLocalTransactional();
MethodRegistry methodRegistry = _connection.getMethodRegistry();
TxSelectOkBody responseBody = methodRegistry.createTxSelectOkBody();
_connection.writeFrame(responseBody.generateFrame(_channelId));
}
@Override
public void receiveTxCommit()
{
if(_logger.isDebugEnabled())
{
_logger.debug("RECV[" + _channelId + "] TxCommit");
}
if (!isTransactional())
{
closeChannel(AMQConstant.COMMAND_INVALID,
"Fatal error: commit called on non-transactional channel");
}
commit(new Runnable()
{
@Override
public void run()
{
_connection.writeFrame(_txCommitOkFrame);
}
}, true);
}
@Override
public void receiveTxRollback()
{
if(_logger.isDebugEnabled())
{
_logger.debug("RECV[" + _channelId + "] TxRollback");
}
if (!isTransactional())
{
closeChannel(AMQConstant.COMMAND_INVALID,
"Fatal error: rollback called on non-transactional channel");
}
final MethodRegistry methodRegistry = _connection.getMethodRegistry();
final AMQMethodBody responseBody = methodRegistry.createTxRollbackOkBody();
Runnable task = new Runnable()
{
public void run()
{
_connection.writeFrame(responseBody.generateFrame(_channelId));
}
};
rollback(task);
//Now resend all the unacknowledged messages back to the original subscribers.
//(Must be done after the TxnRollback-ok response).
// Why, are we not allowed to send messages back to client before the ok method?
resend();
}
@Override
public void receiveConfirmSelect(final boolean nowait)
{
if(_logger.isDebugEnabled())
{
_logger.debug("RECV[" + _channelId + "] ConfirmSelect [ nowait: " + nowait + " ]");
}
_confirmOnPublish = true;
if(!nowait)
{
_connection.writeFrame(new AMQFrame(_channelId, ConfirmSelectOkBody.INSTANCE));
}
}
private void closeChannel(final AMQConstant cause, final String message)
{
_connection.closeChannelAndWriteFrame(this, cause, message);
}
private boolean isDefaultExchange(final AMQShortString exchangeName)
{
return exchangeName == null || AMQShortString.EMPTY_STRING.equals(exchangeName);
}
private void setDefaultQueue(Queue<?> queue)
{
Queue<?> currentDefaultQueue = _defaultQueue;
if (queue != currentDefaultQueue)
{
if (currentDefaultQueue != null)
{
currentDefaultQueue.removeDeleteTask(_defaultQueueAssociationClearingTask);
}
if (queue != null)
{
queue.addDeleteTask(_defaultQueueAssociationClearingTask);
}
}
_defaultQueue = queue;
}
private Queue<?> getDefaultQueue()
{
return _defaultQueue;
}
private class DefaultQueueAssociationClearingTask implements Action<Queue<?>>
{
@Override
public void performAction(final Queue<?> queue)
{
if ( queue == _defaultQueue)
{
_defaultQueue = null;
}
}
}
@Override
public boolean processPending()
{
if (!getAMQPConnection().isIOThread())
{
return false;
}
boolean desiredBlockingState = _blocking.get();
if (desiredBlockingState != _wireBlockingState)
{
_wireBlockingState = desiredBlockingState;
flow(!desiredBlockingState);
_blockTime = desiredBlockingState ? System.currentTimeMillis() : 0;
}
boolean consumerListNeedsRefreshing;
if(_consumersWithPendingWork.isEmpty())
{
_consumersWithPendingWork.addAll(getConsumerTargets());
consumerListNeedsRefreshing = false;
}
else
{
consumerListNeedsRefreshing = true;
}
// QPID-7447: prevent unnecessary allocation of empty iterator
Iterator<ConsumerTarget_0_8> iter = _consumersWithPendingWork.isEmpty() ? Collections.<ConsumerTarget_0_8>emptyIterator() : _consumersWithPendingWork.iterator();
boolean consumerHasMoreWork = false;
while(iter.hasNext())
{
final ConsumerTarget_0_8 target = iter.next();
iter.remove();
if(target.hasPendingWork())
{
consumerHasMoreWork = true;
target.processPending();
break;
}
}
return consumerHasMoreWork || consumerListNeedsRefreshing;
}
@Override
public void addTicker(final Ticker ticker)
{
getConnection().getAggregateTicker().addTicker(ticker);
// trigger a wakeup to ensure the ticker will be taken into account
getAMQPConnection().notifyWork();
}
@Override
public void removeTicker(final Ticker ticker)
{
getConnection().getAggregateTicker().removeTicker(ticker);
}
@Override
public void notifyConsumerTargetCurrentStates()
{
for(ConsumerTarget_0_8 consumerTarget : getConsumerTargets())
{
if(!consumerTarget.isPullOnly())
{
consumerTarget.notifyCurrentState();
}
}
}
@Override
public void ensureConsumersNoticedStateChange()
{
for (ConsumerTarget_0_8 consumerTarget : getConsumerTargets())
{
try
{
consumerTarget.getSendLock();
}
finally
{
consumerTarget.releaseSendLock();
}
}
}
private Collection<ConsumerTarget_0_8> getConsumerTargets()
{
return _tag2SubscriptionTargetMap.values();
}
private Exchange<?> getExchange(String name)
{
MessageDestination destination = getAddressSpace().getAttainedMessageDestination(name);
return destination instanceof Exchange ? (Exchange<?>) destination : null;
}
private Queue<?> getQueue(String name)
{
MessageSource source = getAddressSpace().getAttainedMessageSource(name);
return source instanceof Queue ? (Queue<?>) source : null;
}
public void dispose()
{
_txCommitOkFrame.dispose();
}
}