Google Git
Sign in
apache / qpid-broker-j / c5c1b8b022a5ed7fa457fd02889151e3ffeba5f0 / . / broker-plugins / amqp-0-8-protocol / src / main / java / org / apache / qpid / server / protocol / v0_8 / AMQChannel.java
blob: 595b726ed35860bf54e21c127e9208efd300ed7d [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.server.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.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import javax.security.auth.Subject;
import com.google.common.base.Function;
import com.google.common.collect.Collections2;
import com.google.common.util.concurrent.ListenableFuture;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.qpid.server.bytebuffer.QpidByteBuffer;
import org.apache.qpid.server.configuration.IllegalConfigurationException;
import org.apache.qpid.server.consumer.ConsumerOption;
import org.apache.qpid.server.consumer.ConsumerTarget;
import org.apache.qpid.server.exchange.ExchangeDefaults;
import org.apache.qpid.server.filter.AMQInvalidArgumentException;
import org.apache.qpid.server.filter.AMQPFilterTypes;
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.logging.EventLoggerProvider;
import org.apache.qpid.server.logging.LogMessage;
import org.apache.qpid.server.logging.messages.ChannelMessages;
import org.apache.qpid.server.logging.messages.ExchangeMessages;
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.MessageInstanceConsumer;
import org.apache.qpid.server.message.MessageReference;
import org.apache.qpid.server.message.MessageSource;
import org.apache.qpid.server.message.RejectType;
import org.apache.qpid.server.message.RoutingResult;
import org.apache.qpid.server.message.ServerMessage;
import org.apache.qpid.server.model.AbstractConfiguredObject;
import org.apache.qpid.server.model.AlternateBinding;
import org.apache.qpid.server.model.Broker;
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.protocol.ErrorCodes;
import org.apache.qpid.server.protocol.ProtocolVersion;
import org.apache.qpid.server.protocol.v0_8.UnacknowledgedMessageMap.Visitor;
import org.apache.qpid.server.protocol.v0_8.transport.*;
import org.apache.qpid.server.queue.QueueArgumentsConverter;
import org.apache.qpid.server.session.AbstractAMQPSession;
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.transport.AMQPConnection;
import org.apache.qpid.server.txn.AsyncAutoCommitTransaction;
import org.apache.qpid.server.txn.AsyncCommand;
import org.apache.qpid.server.txn.LocalTransaction;
import org.apache.qpid.server.txn.ServerTransaction;
import org.apache.qpid.server.util.Action;
import org.apache.qpid.server.virtualhost.MessageDestinationIsAlternateException;
import org.apache.qpid.server.virtualhost.RequiredExchangeException;
import org.apache.qpid.server.virtualhost.ReservedExchangeNameException;
import org.apache.qpid.server.virtualhost.UnknownAlternateBindingException;
public class AMQChannel extends AbstractAMQPSession<AMQChannel, ConsumerTarget_0_8>
implements AsyncAutoCommitTransaction.FutureRecorder,
ServerChannelMethodProcessor,
EventLoggerProvider, CreditRestorer, org.apache.qpid.server.util.Deletable<AMQChannel>
{
public static final int DEFAULT_PREFETCH = 4096;
private static final Logger LOGGER = LoggerFactory.getLogger(AMQChannel.class);
private static final InfiniteCreditCreditManager INFINITE_CREDIT_CREDIT_MANAGER = new InfiniteCreditCreditManager();
private static final Function<MessageConsumerAssociation, MessageInstance>
MESSAGE_INSTANCE_FUNCTION = new Function<MessageConsumerAssociation, MessageInstance>()
{
@Override
public MessageInstance apply(final MessageConsumerAssociation input)
{
return input.getMessageInstance();
}
};
private static final String ALTERNATE_EXCHANGE = "alternateExchange";
private final DefaultQueueAssociationClearingTask
_defaultQueueAssociationClearingTask = new DefaultQueueAssociationClearingTask();
private final int _channelId;
private final Pre0_10CreditManager _creditManager;
/**
* 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 MessageStore _messageStore;
private final java.util.Queue<AsyncCommand> _unfinishedCommandsQueue = new ConcurrentLinkedQueue<>();
private final UnacknowledgedMessageMap _unacknowledgedMessageMap;
private final AtomicBoolean _suspended = new AtomicBoolean(false);
private volatile ServerTransaction _transaction;
private final AMQPConnection_0_8 _connection;
private final AtomicBoolean _closing = new AtomicBoolean(false);
private final Set<Object> _blockingEntities = Collections.synchronizedSet(new HashSet<Object>());
private final AtomicBoolean _blocking = new AtomicBoolean(false);
private volatile boolean _rollingBack;
private List<MessageConsumerAssociation> _resendList = new ArrayList<>();
private static final
AMQShortString IMMEDIATE_DELIVERY_REPLY_TEXT =
AMQShortString.createAMQShortString("Immediate delivery is not possible.");
private final ClientDeliveryMethod _clientDeliveryMethod;
private final ImmediateAction _immediateAction = new ImmediateAction();
private long _blockTime;
private long _blockingTimeout;
private boolean _confirmOnPublish;
private long _confirmedMessageCounter;
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;
private boolean _channelFlow = true;
public AMQChannel(AMQPConnection_0_8 connection, int channelId, final MessageStore messageStore)
{
super(connection, channelId);
_creditManager = new Pre0_10CreditManager(0L, 0L,
connection.getContextValue(Long.class, AMQPConnection_0_8.HIGH_PREFETCH_LIMIT),
connection.getContextValue(Long.class, AMQPConnection_0_8.BATCH_LIMIT));
_unacknowledgedMessageMap = new UnacknowledgedMessageMapImpl(DEFAULT_PREFETCH, this);
_connection = connection;
_channelId = channelId;
_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);
}
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,
MessageSource.QueueDeleted
{
final GetDeliveryMethod getDeliveryMethod = new GetDeliveryMethod(queue);
ConsumerTarget_0_8 target;
EnumSet<ConsumerOption> options = EnumSet.of(ConsumerOption.TRANSIENT, ConsumerOption.ACQUIRES,
ConsumerOption.SEES_REQUEUES);
if (acks)
{
target = ConsumerTarget_0_8.createGetAckTarget(this,
AMQShortString.EMPTY_STRING, null,
INFINITE_CREDIT_CREDIT_MANAGER, getDeliveryMethod);
}
else
{
target = ConsumerTarget_0_8.createGetNoAckTarget(this,
AMQShortString.EMPTY_STRING, null,
INFINITE_CREDIT_CREDIT_MANAGER, getDeliveryMethod);
}
queue.addConsumer(target, null, AMQMessage.class, "", options, null);
target.updateNotifyWorkDesired();
boolean canCallSendNextMessageAgain;
do
{
canCallSendNextMessageAgain = target.sendNextMessage();
}
while (canCallSendNextMessageAgain && !getDeliveryMethod.hasDeliveredMessage());
target.close();
return getDeliveryMethod.hasDeliveredMessage();
}
boolean isTransactional()
{
return _transaction.isTransactional();
}
ServerTransaction getTransaction()
{
return _transaction;
}
public void receivedComplete()
{
AccessController.doPrivileged(new PrivilegedAction<Void>()
{
@Override
public Void run()
{
sync();
return null;
}
}, getAccessControllerContext());
}
private void setPublishFrame(MessagePublishInfo info, final MessageDestination e)
{
_currentMessage = new IncomingMessage(info);
_currentMessage.setMessageDestination(e);
}
private 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());
String exchangeName = AMQShortString.toString(info.getExchange());
try
{
final MessageDestination destination = _currentMessage.getDestination();
ContentHeaderBody contentHeader = _currentMessage.getContentHeader();
_connection.checkAuthorizedMessagePrincipal(AMQShortString.toString(contentHeader.getProperties().getUserId()));
_publishAuthCache.authorisePublish(destination, routingKey, info.isImmediate(), _connection.getLastReadTime());
if (_confirmOnPublish)
{
_confirmedMessageCounter++;
}
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 = new AMQMessage(storedMessage, _connection.getReference());
try (MessageReference reference = amqMessage.newReference())
{
_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;
}
};
final RoutingResult<AMQMessage> result =
destination.route(amqMessage,
amqMessage.getInitialRoutingAddress(),
instanceProperties);
int enqueues = result.send(_transaction, immediate ? _immediateAction : null);
if (enqueues == 0)
{
boolean mandatory = amqMessage.isMandatory();
boolean closeOnNoRoute = _connection.isCloseWhenNoRoute();
if (LOGGER.isDebugEnabled())
{
LOGGER.debug("Unroutable message exchange='{}', routing key='{}', mandatory={},"
+ " transactionalSession={}, closeOnNoRoute={}, confirmOnPublish={}",
exchangeName,
routingKey,
mandatory,
isTransactional(),
closeOnNoRoute,
_confirmOnPublish);
}
int errorCode = ErrorCodes.NO_ROUTE;
String errorMessage = String.format("No route for message with exchange '%s' and routing key '%s'",
exchangeName,
routingKey);
if (result.containsReject(RejectType.LIMIT_EXCEEDED))
{
errorCode = ErrorCodes.RESOURCE_ERROR;
errorMessage = errorMessage + ":" + result.getRejectReason();
}
if (mandatory
&& isTransactional()
&& !_confirmOnPublish
&& _connection.isCloseWhenNoRoute())
{
_connection.sendConnectionClose(errorCode, errorMessage, _channelId);
}
else
{
if (mandatory || amqMessage.isImmediate())
{
if (_confirmOnPublish)
{
_connection.writeFrame(new AMQFrame(_channelId,
new BasicNackBody(_confirmedMessageCounter,
false,
false)));
}
_transaction.addPostTransactionAction(new WriteReturnAction(errorCode,
errorMessage,
amqMessage));
}
else
{
if (_confirmOnPublish)
{
_connection.writeFrame(new AMQFrame(_channelId,
new BasicAckBody(_confirmedMessageCounter,
false)));
}
message(ExchangeMessages.DISCARDMSG(exchangeName, routingKey));
}
}
}
else
{
if (_confirmOnPublish)
{
BasicAckBody responseBody = _connection.getMethodRegistry()
.createBasicAckBody(_confirmedMessageCounter, false);
_connection.writeFrame(responseBody.generateFrame(_channelId));
}
}
}
}
finally
{
_connection.registerMessageReceived(bodySize);
if (isTransactional())
{
_connection.registerTransactedMessageReceived();
}
_currentMessage = null;
}
}
catch (AccessControlException e)
{
_connection.sendConnectionClose(ErrorCodes.ACCESS_REFUSED, e.getMessage(), getChannelId());
}
}
}
private 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(ErrorCodes.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;
}
private int getNextConsumerTag()
{
return ++_consumerTag;
}
/**
* 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
*/
private AMQShortString consumeFromSource(AMQShortString tag, Collection<MessageSource> sources, boolean acks,
FieldTable arguments, boolean exclusive, boolean noLocal)
throws MessageSource.ExistingConsumerPreventsExclusive,
MessageSource.ExistingExclusiveConsumer,
AMQInvalidArgumentException,
MessageSource.ConsumerAccessRefused, ConsumerTagInUseException, MessageSource.QueueDeleted
{
if (tag == null)
{
tag = AMQShortString.createAMQShortString("sgen_" + getNextConsumerTag());
}
if (_tag2SubscriptionTargetMap.containsKey(tag))
{
throw new ConsumerTagInUseException("Consumer already exists with same tag: " + tag);
}
ConsumerTarget_0_8 target;
EnumSet<ConsumerOption> options = EnumSet.noneOf(ConsumerOption.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,
INFINITE_CREDIT_CREDIT_MANAGER, multiQueue);
}
else if(acks)
{
target = ConsumerTarget_0_8.createAckTarget(this, tag, arguments, _creditManager, multiQueue);
options.add(ConsumerOption.ACQUIRES);
options.add(ConsumerOption.SEES_REQUEUES);
}
else
{
target = ConsumerTarget_0_8.createNoAckTarget(this, tag, arguments,
INFINITE_CREDIT_CREDIT_MANAGER, multiQueue);
options.add(ConsumerOption.ACQUIRES);
options.add(ConsumerOption.SEES_REQUEUES);
}
if(exclusive)
{
options.add(ConsumerOption.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)
{
source.addConsumer(target,
filterManager,
AMQMessage.class,
AMQShortString.toString(tag),
options, priority);
}
target.updateNotifyWorkDesired();
}
catch (AccessControlException
| MessageSource.ExistingExclusiveConsumer
| MessageSource.ExistingConsumerPreventsExclusive
| MessageSource.QueueDeleted
| 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.
*/
private boolean unsubscribeConsumer(AMQShortString consumerTag)
{
if (LOGGER.isDebugEnabled())
{
LOGGER.debug("Unsubscribing consumer '{}' on channel {}", consumerTag, this);
}
ConsumerTarget_0_8 target = _tag2SubscriptionTargetMap.remove(consumerTag);
if (target != null)
{
target.close();
return true;
}
else
{
LOGGER.warn("Attempt to unsubscribe consumer with tag '" + consumerTag + "' which is not registered.");
}
return false;
}
@Override
public void close()
{
close(0, null);
}
public void close(int cause, String message)
{
if(!_closing.compareAndSet(false, true))
{
//Channel is already closing
return;
}
try
{
unsubscribeAllConsumers();
setDefaultQueue(null);
for (Action<? super AMQChannel> task : _taskList)
{
task.performAction(this);
}
if (_transaction instanceof LocalTransaction)
{
if (((LocalTransaction) _transaction).hasOutstandingWork())
{
_connection.incrementTransactionRollbackCounter();
}
_connection.decrementTransactionOpenCounter();
_connection.unregisterTransactionTickers(_transaction);
}
_transaction.rollback();
requeue();
}
finally
{
dispose();
LogMessage operationalLogMessage = cause == 0?
ChannelMessages.CLOSE() :
ChannelMessages.CLOSE_FORCED(cause, 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.
* @param usesCredit
*/
public void addUnacknowledgedMessage(MessageInstance entry,
long deliveryTag,
MessageInstanceConsumer consumer,
final boolean usesCredit)
{
if (LOGGER.isDebugEnabled())
{
LOGGER.debug(debugIdentity() + " Adding unacked message(" + entry.getMessage().toString() + " DT:" + deliveryTag
+ ") for " + consumer + " on " + entry.getOwningResource().getName());
}
_unacknowledgedMessageMap.add(deliveryTag, entry, consumer, usesCredit);
}
private final String id = "(" + System.identityHashCode(this) + ")";
private 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()
{
final Map<Long, MessageConsumerAssociation> copy = new LinkedHashMap<>();
_unacknowledgedMessageMap.visit(new Visitor()
{
@Override
public boolean callback(final long deliveryTag, final MessageConsumerAssociation messageConsumerPair)
{
copy.put(deliveryTag, messageConsumerPair);
return false;
}
@Override
public void visitComplete()
{
}
});
if (!copy.isEmpty())
{
if (LOGGER.isDebugEnabled())
{
LOGGER.debug("Requeuing {} unacked messages", copy.size());
}
}
for (Map.Entry<Long, MessageConsumerAssociation> entry : copy.entrySet())
{
MessageInstance unacked = entry.getValue().getMessageInstance();
MessageInstanceConsumer consumer = entry.getValue().getConsumer();
// Mark message redelivered
unacked.setRedelivered();
// here we wish to restore credit
_unacknowledgedMessageMap.remove(entry.getKey(), true);
// Ensure message is released for redelivery
unacked.release(consumer);
}
}
/**
* Requeue a single message
*
* @param deliveryTag The message to requeue
*
*/
private void requeue(long deliveryTag)
{
final MessageConsumerAssociation association = _unacknowledgedMessageMap.remove(deliveryTag, true);
if (association != null)
{
MessageInstance unacked = association.getMessageInstance();
// Mark message redelivered
unacked.setRedelivered();
// Ensure message is released for redelivery
unacked.release(association.getConsumer());
}
else
{
LOGGER.warn("Requested requeue of message: {} but no such delivery tag exists.", deliveryTag);
}
}
private boolean isMaxDeliveryCountEnabled(final long deliveryTag)
{
final MessageInstance queueEntry = _unacknowledgedMessageMap.get(deliveryTag);
if (queueEntry != null)
{
final int maximumDeliveryCount = queueEntry.getMaximumDeliveryCount();
return maximumDeliveryCount > 0;
}
return false;
}
private 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, MessageConsumerAssociation> msgToRequeue = new LinkedHashMap<>();
final Map<Long, MessageConsumerAssociation> msgToResend = new LinkedHashMap<>();
if (LOGGER.isDebugEnabled())
{
LOGGER.debug("Unacknowledged messages: {}", _unacknowledgedMessageMap.size());
}
_unacknowledgedMessageMap.visit(new Visitor()
{
@Override
public boolean callback(final long deliveryTag, final MessageConsumerAssociation association)
{
if (association.getConsumer().isClosed())
{
// consumer has gone
msgToRequeue.put(deliveryTag, association);
}
else
{
// Consumer still exists
msgToResend.put(deliveryTag, association);
}
return false;
}
@Override
public void visitComplete()
{
}
});
for (Map.Entry<Long, MessageConsumerAssociation> entry : msgToResend.entrySet())
{
long deliveryTag = entry.getKey();
MessageInstance message = entry.getValue().getMessageInstance();
MessageInstanceConsumer consumer = entry.getValue().getConsumer();
// 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.makeAcquisitionUnstealable(consumer))
{
message.decrementDeliveryCount();
consumer.getTarget().send(consumer, message, false);
// remove from unacked map - don't want to restore credit though(!)
_unacknowledgedMessageMap.remove(deliveryTag, false);
}
else
{
msgToRequeue.put(deliveryTag, entry.getValue());
}
}
// Process Messages to Requeue at the front of the queue
for (Map.Entry<Long, MessageConsumerAssociation> entry : msgToRequeue.entrySet())
{
long deliveryTag = entry.getKey();
MessageInstance message = entry.getValue().getMessageInstance();
MessageInstanceConsumer consumer = entry.getValue().getConsumer();
//Amend the delivery counter as the client hasn't seen these messages yet.
message.decrementDeliveryCount();
// here we do wish to restore credit
_unacknowledgedMessageMap.remove(deliveryTag, true);
message.setRedelivered();
message.release(consumer);
}
}
private UnacknowledgedMessageMap getUnacknowledgedMessageMap()
{
return _unacknowledgedMessageMap;
}
/**
* Called from the ChannelFlowHandler to suspend this Channel
* @param suspended boolean, should this Channel be suspended
*/
private 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"));
}
// 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(MessageInstanceConsumer sub : s.getConsumers())
{
sub.externalStateChange();
}
}
}
// Log Suspension only after we have confirmed all suspensions are
// stopped.
if (suspended && _logChannelFlowMessages)
{
messageWithSubject(ChannelMessages.FLOW("Stopped"));
}
}
}
private 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
{
_connection.incrementTransactionBeginCounter();
}
}
});
}
else
{
_transaction.commit(immediateAction);
_connection.incrementTransactionBeginCounter();
}
}
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.
try
{
_transaction.rollback();
}
finally
{
_rollingBack = false;
_connection.incrementTransactionRollbackCounter();
_connection.incrementTransactionBeginCounter();
}
postRollbackTask.run();
for(MessageConsumerAssociation association : _resendList)
{
final MessageInstance messageInstance = association.getMessageInstance();
final MessageInstanceConsumer consumer = association.getConsumer();
if (consumer.isClosed())
{
messageInstance.release(consumer);
}
else
{
if (messageInstance.makeAcquisitionUnstealable(consumer)
&& _creditManager.useCreditForMessage(association.getSize()))
{
consumer.getTarget().send(consumer, messageInstance, false);
}
else
{
messageInstance.release(consumer);
}
}
}
_resendList.clear();
if(requiresSuspend)
{
_suspended.set(false);
for(ConsumerTarget_0_8 target : getConsumerTargets())
{
for(MessageInstanceConsumer sub : target.getConsumers())
{
sub.externalStateChange();
}
}
}
}
@Override
public String toString()
{
return "("+ _suspended.get() + ", " + _closing.get() + ", " + _connection.isClosing() + ") "+"["+ _connection.toString()+":"+_channelId+"]";
}
@Override
public boolean isClosing()
{
return _closing.get() || getConnection().isClosing();
}
public AMQPConnection_0_8<?> getConnection()
{
return _connection;
}
private 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;
}
boolean hasCredit = _creditManager.hasCredit();
_creditManager.setCreditLimits(prefetchSize, prefetchCount);
if(hasCredit != _creditManager.hasCredit())
{
updateAllConsumerNotifyWorkDesired();
}
}
public ClientDeliveryMethod getClientDeliveryMethod()
{
return _clientDeliveryMethod;
}
public Subject getSubject()
{
return _subject;
}
private boolean hasCurrentMessage()
{
return _currentMessage != null;
}
public boolean isChannelFlow()
{
return _channelFlow;
}
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 MessageSource _queue;
private boolean _deliveredMessage;
public GetDeliveryMethod(final MessageSource queue)
{
_queue = queue;
}
@Override
public long deliverToClient(final ConsumerTarget_0_8 target, final AMQMessage message,
final InstanceProperties props, final long deliveryTag)
{
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()
{
}
@Override
public void performAction(MessageInstance entry)
{
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,
ErrorCodes.NO_CONSUMERS,
IMMEDIATE_DELIVERY_REPLY_TEXT);
}
@Override
public void onRollback()
{
}
}
);
txn.commit();
}
finally
{
ref.release();
}
}
}
}
private class MessageAcknowledgeAction implements ServerTransaction.Action
{
private Collection<MessageConsumerAssociation> _ackedMessages;
public MessageAcknowledgeAction(Collection<MessageConsumerAssociation> ackedMessages)
{
_ackedMessages = ackedMessages;
}
@Override
public void postCommit()
{
try
{
for(MessageConsumerAssociation association : _ackedMessages)
{
association.getMessageInstance().delete();
}
}
finally
{
_ackedMessages = Collections.emptySet();
}
}
@Override
public void onRollback()
{
// explicit rollbacks resend the message after the rollback-ok is sent
if(_rollingBack)
{
for(MessageConsumerAssociation association : _ackedMessages)
{
association.getMessageInstance().makeAcquisitionStealable();
}
_resendList.addAll(_ackedMessages);
}
else
{
try
{
for(MessageConsumerAssociation association : _ackedMessages)
{
final MessageInstance messageInstance = association.getMessageInstance();
messageInstance.release(association.getConsumer());
}
}
finally
{
_ackedMessages = Collections.emptySet();
}
}
}
}
private class WriteReturnAction implements ServerTransaction.Action
{
private final int _errorCode;
private final String _description;
private final MessageReference<AMQMessage> _reference;
public WriteReturnAction(int errorCode,
String description,
AMQMessage message)
{
_errorCode = errorCode;
_description = description;
_reference = message.newReference();
}
@Override
public void postCommit()
{
AMQMessage message = _reference.getMessage();
_connection.getProtocolOutputConverter().writeReturn(message.getMessagePublishInfo(),
message.getContentHeaderBody(),
message,
_channelId,
_errorCode,
AMQShortString.validValueOf(_description));
_reference.release();
}
@Override
public void onRollback()
{
_reference.release();
}
}
@Override
public synchronized void block()
{
if(_blockingEntities.add(this))
{
if(_blocking.compareAndSet(false,true))
{
messageWithSubject(ChannelMessages.FLOW_ENFORCED("** All Queues **"));
getConnection().notifyWork(this);
}
}
}
@Override
public synchronized void unblock()
{
if(_blockingEntities.remove(this))
{
if(_blockingEntities.isEmpty() && _blocking.compareAndSet(true,false))
{
messageWithSubject(ChannelMessages.FLOW_REMOVED());
getConnection().notifyWork(this);
}
}
}
@Override
public synchronized void block(Queue<?> queue)
{
if(_blockingEntities.add(queue))
{
if(_blocking.compareAndSet(false,true))
{
messageWithSubject(ChannelMessages.FLOW_ENFORCED(queue.getName()));
getConnection().notifyWork(this);
}
}
}
@Override
public synchronized void unblock(Queue<?> queue)
{
if(_blockingEntities.remove(queue))
{
if(_blockingEntities.isEmpty() && _blocking.compareAndSet(true,false) && !isClosing())
{
messageWithSubject(ChannelMessages.FLOW_REMOVED());
getConnection().notifyWork(this);
}
}
}
@Override
public void transportStateChanged()
{
updateAllConsumerNotifyWorkDesired();
_creditManager.restoreCredit(0, 0);
INFINITE_CREDIT_CREDIT_MANAGER.restoreCredit(0, 0);
if (!_consumersWithPendingWork.isEmpty() && !getAMQPConnection().isTransportBlockedForWriting())
{
getAMQPConnection().notifyWork(this);
}
}
void updateAllConsumerNotifyWorkDesired()
{
for(ConsumerTarget_0_8 target : _tag2SubscriptionTargetMap.values())
{
target.updateNotifyWorkDesired();
}
}
@Override
public Object getConnectionReference()
{
return getConnection().getReference();
}
@Override
public int getUnacknowledgedMessageCount()
{
return getUnacknowledgedMessageMap().size();
}
private void sendFlow(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 MessageConsumerAssociation association = unackedMap.remove(deliveryTag, true);
if (association == null)
{
LOGGER.warn("No message found, unable to DLQ delivery tag: " + deliveryTag);
}
else
{
final MessageInstance messageInstance = association.getMessageInstance();
final ServerMessage msg = messageInstance.getMessage();
int requeues = 0;
if (messageInstance.makeAcquisitionUnstealable(association.getConsumer()))
{
requeues = messageInstance.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 = messageInstance.getOwningResource();
if(owningResource instanceof Queue)
{
final Queue<?> queue = (Queue<?>) owningResource;
final MessageDestination alternateBindingDestination = queue.getAlternateBindingDestination();
if (alternateBindingDestination == null)
{
messageWithSubject(ChannelMessages.DISCARDMSG_NOALTEXCH(msg.getMessageNumber(),
queue.getName(),
msg.getInitialRoutingAddress()));
}
else
{
messageWithSubject(ChannelMessages.DISCARDMSG_NOROUTE(msg.getMessageNumber(),
alternateBindingDestination.getName()));
}
}
}
}
}
@Override
public void recordFuture(final ListenableFuture<Void> future, final ServerTransaction.Action action)
{
_unfinishedCommandsQueue.add(new AsyncCommand(future, action));
}
private 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();
}
}
@Override
public long getTransactionStartTimeLong()
{
ServerTransaction serverTransaction = _transaction;
if (serverTransaction.isTransactional())
{
return serverTransaction.getTransactionStartTime();
}
else
{
return 0L;
}
}
@Override
public long getTransactionUpdateTimeLong()
{
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(ErrorCodes.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 + " ]");
}
Collection<MessageConsumerAssociation> ackedMessages = _unacknowledgedMessageMap.acknowledge(deliveryTag, multiple);
if (!ackedMessages.isEmpty())
{
final Collection<MessageInstance> messages =
Collections2.transform(ackedMessages, MESSAGE_INSTANCE_FUNCTION);
_transaction.dequeue(messages, new MessageAcknowledgeAction(ackedMessages));
}
/*
The AMQP 0-9-1 spec requires to raise a channel exception "precondition-failed"
when delivery tag is not valid:
{quote}
The server MUST validate that a non-zero delivery-tag refers to a delivered message, and raise a channel
exception if this is not the case. On a transacted channel, this check MUST be done immediately and not
delayed until a Tx.Commit. Specifically, a client MUST not acknowledge the same message more than once.
{quote}
The current broker behaviour is spec incompliant but it is kept for backward compatibility.
It should close the channel as below:
if (ackedMessages.isEmpty())
closeChannel(ErrorCodes.NOT_ALLOWED, "precondition-failed: Delivery tag '%d' is not valid.");
*/
}
@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(ErrorCodes.NOT_FOUND, "No such queue, '" + queueName + "'");
}
else
{
_connection.sendConnectionClose(ErrorCodes.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(ErrorCodes.NOT_ALLOWED,
"Non-unique consumer tag, '" + consumerTag1
+ "'", _channelId);
}
catch (AMQInvalidArgumentException ise)
{
_connection.sendConnectionClose(ErrorCodes.ARGUMENT_INVALID, ise.getMessage(), _channelId);
}
catch (Queue.ExistingExclusiveConsumer e)
{
_connection.sendConnectionClose(ErrorCodes.ACCESS_REFUSED,
"Cannot subscribe to queue '"
+ queue1.getName()
+ "' as it already has an existing exclusive consumer", _channelId);
}
catch (Queue.ExistingConsumerPreventsExclusive e)
{
_connection.sendConnectionClose(ErrorCodes.ACCESS_REFUSED,
"Cannot subscribe to queue '"
+ queue1.getName()
+ "' exclusively as it already has a consumer", _channelId);
}
catch (AccessControlException e)
{
_connection.sendConnectionClose(ErrorCodes.ACCESS_REFUSED, "Cannot subscribe to queue '"
+ queue1.getName()
+ "' permission denied", _channelId);
}
catch (MessageSource.ConsumerAccessRefused consumerAccessRefused)
{
_connection.sendConnectionClose(ErrorCodes.ACCESS_REFUSED,
"Cannot subscribe to queue '"
+ queue1.getName()
+ "' as it already has an incompatible exclusivity policy", _channelId);
}
catch (MessageSource.QueueDeleted queueDeleted)
{
_connection.sendConnectionClose(ErrorCodes.NOT_FOUND,
"Cannot subscribe to queue '"
+ queue1.getName()
+ "' as it has been deleted", _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(ErrorCodes.NOT_FOUND, "No such queue, '" + queueName + "'", _channelId);
}
else
{
_connection.sendConnectionClose(ErrorCodes.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(ErrorCodes.ACCESS_REFUSED, e.getMessage(), _channelId);
}
catch (MessageSource.ExistingExclusiveConsumer e)
{
_connection.sendConnectionClose(ErrorCodes.NOT_ALLOWED, "Queue has an exclusive consumer", _channelId);
}
catch (MessageSource.ExistingConsumerPreventsExclusive e)
{
_connection.sendConnectionClose(ErrorCodes.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(ErrorCodes.NOT_ALLOWED,
"Queue has an incompatible exclusivity policy", _channelId);
}
catch (MessageSource.QueueDeleted queueDeleted)
{
_connection.sendConnectionClose(ErrorCodes.NOT_FOUND, "Queue has been deleted", _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(ErrorCodes.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(ErrorCodes.NOT_FOUND, "Unknown exchange name: '" + exchangeName + "'");
}
else
{
MessagePublishInfo info = new MessagePublishInfo(exchangeName,
immediate,
mandatory,
routingKey);
try
{
setPublishFrame(info, destination);
}
catch (AccessControlException e)
{
_connection.sendConnectionClose(ErrorCodes.ACCESS_REFUSED, e.getMessage(), getChannelId());
}
}
}
}
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 + " ]");
}
if (requeue)
{
requeue();
}
else
{
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(ErrorCodes.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())
{
properties.dispose();
closeChannel(ErrorCodes.MESSAGE_TOO_LARGE,
"Message size of " + bodySize + " greater than allowed maximum of " + _connection.getMaxMessageSize());
}
else
{
publishContentHeader(new ContentHeaderBody(properties, bodySize));
}
}
else
{
properties.dispose();
_connection.sendConnectionClose(ErrorCodes.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, MessageConsumerAssociation> nackedMessageMap = new LinkedHashMap<>();
_unacknowledgedMessageMap.collect(deliveryTag, multiple, nackedMessageMap);
for(MessageConsumerAssociation unackedMessageConsumerAssociation : nackedMessageMap.values())
{
if (unackedMessageConsumerAssociation == null)
{
LOGGER.warn("Ignoring nack request as message is null for tag:" + deliveryTag);
}
else
{
MessageInstance message = unackedMessageConsumerAssociation.getMessageInstance();
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(unackedMessageConsumerAssociation.getConsumer());
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();
message.release(unackedMessageConsumerAssociation.getConsumer());
}
}
else
{
requeue(deliveryTag);
}
}
}
}
}
}
@Override
public void receiveChannelFlow(final boolean active)
{
if(LOGGER.isDebugEnabled())
{
LOGGER.debug("RECV[" + _channelId + "] ChannelFlow[" +" active: " + active + " ]");
}
sync();
if(_channelFlow != active)
{
_channelFlow = active;
// inform consumer targets
updateAllConsumerNotifyWorkDesired();
}
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 (!AMQShortString.createAMQShortString(ExchangeDefaults.DIRECT_EXCHANGE_CLASS).equals(type))
{
_connection.sendConnectionClose(ErrorCodes.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(ErrorCodes.NOT_FOUND, "Unknown exchange: '" + exchangeName + "'");
}
else if (!(type == null || type.length() == 0) && !exchange.getType().equals(type.toString()))
{
_connection.sendConnectionClose(ErrorCodes.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);
Object alternateExchange = attributes.remove(ALTERNATE_EXCHANGE);
if (alternateExchange != null)
{
String alternateExchangeName = String.valueOf(alternateExchange);
validateAlternateExchangeIsNotQueue(virtualHost, alternateExchangeName);
attributes.put(Exchange.ALTERNATE_BINDING,
Collections.singletonMap(AlternateBinding.DESTINATION, alternateExchangeName));
}
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(ErrorCodes.NOT_ALLOWED,
"Attempt to declare exchange: '" + exchangeName +
"' which begins with reserved prefix.", getChannelId());
}
else if(!nowait)
{
sync();
_connection.writeFrame(declareOkBody.generateFrame(getChannelId()));
}
}
catch (AbstractConfiguredObject.DuplicateNameException e)
{
exchange = (Exchange<?>) e.getExisting();
if (!exchange.getType().equals(typeString))
{
_connection.sendConnectionClose(ErrorCodes.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(ErrorCodes.COMMAND_INVALID, "Unknown exchange type '"
+ e.getType()
+ "' for exchange '"
+ exchangeName
+ "'", getChannelId());
}
catch (AccessControlException e)
{
_connection.sendConnectionClose(ErrorCodes.ACCESS_REFUSED, e.getMessage(), getChannelId());
}
catch (UnknownAlternateBindingException e)
{
final String message = String.format("Unknown alternate exchange '%s'", e.getAlternateBindingName());
_connection.sendConnectionClose(ErrorCodes.NOT_FOUND, message, getChannelId());
}
catch (IllegalArgumentException | IllegalConfigurationException e)
{
_connection.sendConnectionClose(ErrorCodes.COMMAND_INVALID, "Error creating exchange '"
+ exchangeName
+ "': "
+ e.getMessage(), getChannelId());
}
}
}
}
private void validateAlternateExchangeIsNotQueue(final NamedAddressSpace addressSpace, final String alternateExchangeName)
{
MessageDestination alternateMessageDestination = addressSpace.getAttainedMessageDestination(alternateExchangeName, false);
if (alternateMessageDestination != null && !(alternateMessageDestination instanceof Exchange))
{
throw new IllegalConfigurationException(String.format(
"Alternate exchange '%s' is not a destination of type 'exchange'.", alternateExchangeName));
}
}
@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(ErrorCodes.NOT_ALLOWED,
"Default Exchange cannot be deleted", getChannelId());
}
else
{
final String exchangeName = exchangeStr.toString();
final Exchange<?> exchange = getExchange(exchangeName);
if (exchange == null)
{
closeChannel(ErrorCodes.NOT_FOUND, "No such exchange: '" + exchangeStr + "'");
}
else
{
if (ifUnused && exchange.hasBindings())
{
closeChannel(ErrorCodes.IN_USE, "Exchange has bindings");
}
else
{
try
{
exchange.delete();
if (!nowait)
{
ExchangeDeleteOkBody
responseBody = _connection.getMethodRegistry().createExchangeDeleteOkBody();
_connection.writeFrame(responseBody.generateFrame(getChannelId()));
}
}
catch (MessageDestinationIsAlternateException e)
{
closeChannel(ErrorCodes.NOT_ALLOWED, "Exchange in use as an alternate binding destination");
}
catch (RequiredExchangeException e)
{
closeChannel(ErrorCodes.NOT_ALLOWED, "Exchange '" + exchangeStr + "' cannot be deleted");
}
catch (AccessControlException e)
{
_connection.sendConnectionClose(ErrorCodes.ACCESS_REFUSED, e.getMessage(), getChannelId());
}
}
}
}
}
@Override
public void receiveQueueBind(final AMQShortString queueName,
final AMQShortString exchange,
AMQShortString bindingKey,
final boolean nowait,
final FieldTable argumentsTable)
{
if(LOGGER.isDebugEnabled())
{
LOGGER.debug("RECV[" + _channelId + "] QueueBind[" +" queue: " + queueName +
" exchange: " + exchange +
" bindingKey: " + bindingKey +
" nowait: " + nowait + " arguments: " + argumentsTable + " ]");
}
NamedAddressSpace virtualHost = _connection.getAddressSpace();
Queue<?> queue;
if (queueName == null)
{
queue = getDefaultQueue();
if (queue != null)
{
if (bindingKey == null)
{
bindingKey = AMQShortString.valueOf(queue.getName());
}
}
}
else
{
queue = getQueue(queueName.toString());
}
if (queue == null)
{
String message = queueName == null
? "No default queue defined on channel and queue was null"
: "Queue " + queueName + " does not exist.";
closeChannel(ErrorCodes.NOT_FOUND, message);
}
else if (isDefaultExchange(exchange))
{
_connection.sendConnectionClose(ErrorCodes.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(ErrorCodes.NOT_FOUND,
"Exchange '" + exchangeName + "' does not exist.");
}
else
{
try
{
Map<String, Object> arguments = FieldTable.convertToMap(argumentsTable);
String bindingKeyStr = AMQShortString.toString(bindingKey);
if (!exch.isBound(bindingKeyStr, arguments, queue))
{
if (!exch.addBinding(bindingKeyStr, queue, arguments)
&& ExchangeDefaults.TOPIC_EXCHANGE_CLASS.equals(
exch.getType()))
{
exch.replaceBinding(bindingKeyStr, queue, arguments);
}
}
if (LOGGER.isDebugEnabled())
{
LOGGER.debug("Binding queue "
+ queue
+ " to exchange "
+ exch
+ " with routing key "
+ bindingKeyStr);
}
if (!nowait)
{
sync();
MethodRegistry methodRegistry = _connection.getMethodRegistry();
AMQMethodBody responseBody = methodRegistry.createQueueBindOkBody();
_connection.writeFrame(responseBody.generateFrame(getChannelId()));
}
}
catch (AccessControlException e)
{
_connection.sendConnectionClose(ErrorCodes.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 = AMQShortString.createAMQShortString("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(ErrorCodes.NOT_FOUND,
"Queue: '"
+ queueName
+ "' not found on VirtualHost '"
+ virtualHost.getName()
+ "'.");
}
else
{
if (!queue.verifySessionAccess(this))
{
_connection.sendConnectionClose(ErrorCodes.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
{
final String queueNameString = AMQShortString.toString(queueName);
Map<String, Object> wireArguments = FieldTable.convertToMap(arguments);
Object alternateExchange = wireArguments.get(ALTERNATE_EXCHANGE);
if (alternateExchange != null)
{
String alternateExchangeName = String.valueOf(alternateExchange);
validateAlternateExchangeIsNotQueue(virtualHost, alternateExchangeName);
}
Map<String, Object> attributes =
QueueArgumentsConverter.convertWireArgsToModel(queueNameString, wireArguments);
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 (AbstractConfiguredObject.DuplicateNameException qe)
{
queue = (Queue<?>) qe.getExisting();
if (!queue.verifySessionAccess(this))
{
_connection.sendConnectionClose(ErrorCodes.NOT_ALLOWED, "Queue '"
+ queue.getName()
+ "' is exclusive, but not created on this Connection.", getChannelId());
}
else if (queue.isExclusive() != exclusive)
{
closeChannel(ErrorCodes.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(ErrorCodes.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(ErrorCodes.ACCESS_REFUSED, e.getMessage(), getChannelId());
}
catch (UnknownAlternateBindingException e)
{
final String message = String.format("Unknown alternate exchange: '%s'", e.getAlternateBindingName());
_connection.sendConnectionClose(ErrorCodes.NOT_FOUND, message, getChannelId());
}
catch (IllegalArgumentException | IllegalConfigurationException e)
{
String message = String.format("Error creating queue '%s': %s", queueName, e.getMessage());
_connection.sendConnectionClose(ErrorCodes.COMMAND_INVALID, message, 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(ErrorCodes.NOT_FOUND, "Queue '" + queueName + "' does not exist.");
}
else
{
if (ifEmpty && !queue.isEmpty())
{
closeChannel(ErrorCodes.IN_USE, "Queue: '" + queueName + "' is not empty.");
}
else if (ifUnused && !queue.isUnused())
{
// TODO - Error code
closeChannel(ErrorCodes.IN_USE, "Queue: '" + queueName + "' is still used.");
}
else
{
if (!queue.verifySessionAccess(this))
{
_connection.sendConnectionClose(ErrorCodes.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(ErrorCodes.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(ErrorCodes.NOT_ALLOWED, "No queue specified.", getChannelId());
}
else if ((queueName != null) && (queue = getQueue(queueName.toString())) == null)
{
closeChannel(ErrorCodes.NOT_FOUND, "Queue '" + queueName + "' does not exist.");
}
else if (!queue.verifySessionAccess(this))
{
_connection.sendConnectionClose(ErrorCodes.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(ErrorCodes.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(ErrorCodes.NOT_FOUND, message);
}
else if (isDefaultExchange(exchange))
{
_connection.sendConnectionClose(ErrorCodes.NOT_ALLOWED, "Cannot unbind the queue '"
+ queue.getName()
+ "' from the default exchange", getChannelId());
}
else
{
final Exchange<?> exch = getExchange(exchange.toString());
if (exch == null)
{
closeChannel(ErrorCodes.NOT_FOUND, "Exchange '" + exchange + "' does not exist.");
}
else if (!exch.hasBinding(AMQShortString.toString(bindingKey), queue))
{
closeChannel(ErrorCodes.NOT_FOUND, "No such binding");
}
else
{
try
{
exch.deleteBinding(AMQShortString.toString(bindingKey), queue);
final AMQMethodBody responseBody = _connection.getMethodRegistry().createQueueUnbindOkBody();
sync();
_connection.writeFrame(responseBody.generateFrame(getChannelId()));
}
catch (AccessControlException e)
{
_connection.sendConnectionClose(ErrorCodes.ACCESS_REFUSED, e.getMessage(), getChannelId());
}
}
}
}
@Override
public void receiveTxSelect()
{
if(LOGGER.isDebugEnabled())
{
LOGGER.debug("RECV[" + _channelId + "] TxSelect");
}
ServerTransaction txn = _transaction;
if (txn instanceof LocalTransaction)
{
getConnection().unregisterTransactionTickers(_transaction);
}
_transaction = _connection.createLocalTransaction();
long notificationRepeatPeriod = getContextValue(Long.class,
TRANSACTION_TIMEOUT_NOTIFICATION_REPEAT_PERIOD);
getConnection().registerTransactionTickers(_transaction,
message -> _connection.sendConnectionCloseAsync(AMQPConnection.CloseReason.TRANSACTION_TIMEOUT,
message),
notificationRepeatPeriod);
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(ErrorCodes.COMMAND_INVALID,
"Fatal error: commit called on non-transactional channel");
}
else
{
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(ErrorCodes.COMMAND_INVALID,
"Fatal error: rollback called on non-transactional channel");
}
else
{
final MethodRegistry methodRegistry = _connection.getMethodRegistry();
final AMQMethodBody responseBody = methodRegistry.createTxRollbackOkBody();
Runnable task = () -> _connection.writeFrame(responseBody.generateFrame(_channelId));
rollback(task);
// TODO: This is not spec compliant but we currently seem to rely on this behaviour
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(int 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
protected void updateBlockedStateIfNecessary()
{
boolean desiredBlockingState = _blocking.get();
if (desiredBlockingState != _wireBlockingState)
{
_wireBlockingState = desiredBlockingState;
sendFlow(!desiredBlockingState);
_blockTime = desiredBlockingState ? System.currentTimeMillis() : 0;
}
}
@Override
public void restoreCredit(final ConsumerTarget target, final int count, final long size)
{
boolean hasCredit = _creditManager.hasCredit();
_creditManager.restoreCredit(count, size);
if(_creditManager.hasCredit() != hasCredit)
{
if (hasCredit || !_creditManager.isNotBytesLimitedAndHighPrefetch())
{
updateAllConsumerNotifyWorkDesired();
}
}
else if (hasCredit)
{
if (_creditManager.isNotBytesLimitedAndHighPrefetch())
{
if (_creditManager.isCreditOverBatchLimit())
{
updateAllConsumerNotifyWorkDesired();
}
}
else if(_creditManager.isBytesLimited())
{
target.notifyWork();
}
}
}
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();
final IncomingMessage currentMessage = _currentMessage;
if (currentMessage != null)
{
_currentMessage = null;
final ContentHeaderBody contentHeader = currentMessage.getContentHeader();
if (contentHeader != null)
{
contentHeader.dispose();
}
int bodyCount = currentMessage.getBodyCount();
if (bodyCount > 0)
{
for (int i = 0; i < bodyCount; i++)
{
currentMessage.getContentChunk(i).dispose();
}
}
}
}
}