diff options
| author | Alex Rudyy <orudyy@apache.org> | 2015-04-15 09:47:28 +0000 |
|---|---|---|
| committer | Alex Rudyy <orudyy@apache.org> | 2015-04-15 09:47:28 +0000 |
| commit | 0a0baee45ebcff44635907d457c4ff6810b09c87 (patch) | |
| tree | 8bfb0f9eddbc23cff88af69be80ab3ce7d47011c /qpid/java/broker-plugins/amqp-0-8-protocol | |
| parent | 54aa3d7070da16ce55c28ccad3f7d0871479e461 (diff) | |
| download | qpid-python-0a0baee45ebcff44635907d457c4ff6810b09c87.tar.gz | |
QPID-6481: Move java source tree to top level
git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1673693 13f79535-47bb-0310-9956-ffa450edef68
Diffstat (limited to 'qpid/java/broker-plugins/amqp-0-8-protocol')
41 files changed, 0 insertions, 11468 deletions
diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/pom.xml b/qpid/java/broker-plugins/amqp-0-8-protocol/pom.xml deleted file mode 100644 index e09a3ba922..0000000000 --- a/qpid/java/broker-plugins/amqp-0-8-protocol/pom.xml +++ /dev/null @@ -1,77 +0,0 @@ -<?xml version="1.0" encoding="UTF-8"?> -<!-- - 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. ---> -<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd"> - <modelVersion>4.0.0</modelVersion> - - <parent> - <groupId>org.apache.qpid</groupId> - <artifactId>qpid-java-build</artifactId> - <version>0.32-SNAPSHOT</version> - <relativePath>../../pom.xml</relativePath> - </parent> - - <artifactId>qpid-broker-plugins-amqp-0-8-protocol</artifactId> - <name>Qpid AMQP 0-8 Protocol Broker Plug-in</name> - <description>AMQP 0-8, 0-9 and 0-9-1 protocol broker plug-in</description> - - <properties> - <qpid.home>${basedir}/../</qpid.home> <!-- override for broker tests --> - </properties> - - <dependencies> - <dependency> - <groupId>org.apache.qpid</groupId> - <artifactId>qpid-broker-core</artifactId> - <version>${project.version}</version> - <scope>provided</scope> - </dependency> - - <dependency> - <groupId>org.apache.qpid</groupId> - <artifactId>qpid-broker-codegen</artifactId> - <version>${project.version}</version> - <optional>true</optional> - </dependency> - - <dependency> - <groupId>log4j</groupId> - <artifactId>log4j</artifactId> - <version>${log4j-version}</version> - </dependency> - - <!-- test dependencies --> - <dependency> - <groupId>org.apache.qpid</groupId> - <artifactId>qpid-test-utils</artifactId> - <version>${project.version}</version> - <scope>test</scope> - </dependency> - - <dependency> - <groupId>org.apache.qpid</groupId> - <artifactId>qpid-broker-core</artifactId> - <version>${project.version}</version> - <type>test-jar</type> - <scope>test</scope> - </dependency> - </dependencies> - - <build> - </build> - -</project> diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java deleted file mode 100644 index 522408910d..0000000000 --- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java +++ /dev/null @@ -1,3670 +0,0 @@ -/* - * - * 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.nio.ByteBuffer; -import java.security.AccessControlException; -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.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.atomic.AtomicBoolean; -import java.util.concurrent.atomic.AtomicLong; - -import javax.security.auth.Subject; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import org.apache.qpid.AMQConnectionException; -import org.apache.qpid.AMQException; -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.TransactionTimeoutHelper; -import org.apache.qpid.server.TransactionTimeoutHelper.CloseAction; -import org.apache.qpid.server.configuration.BrokerProperties; -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.exchange.ExchangeImpl; -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.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.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.queue.AMQQueue; -import org.apache.qpid.server.queue.QueueArgumentsConverter; -import org.apache.qpid.server.security.SecurityManager; -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.ConnectionScopedRuntimeException; -import org.apache.qpid.server.util.FutureResult; -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.server.virtualhost.VirtualHostImpl; -import org.apache.qpid.transport.TransportException; - -public class AMQChannel - implements AMQSessionModel<AMQChannel, AMQProtocolEngine>, - AsyncAutoCommitTransaction.FutureRecorder, - ServerChannelMethodProcessor -{ - public static final int DEFAULT_PREFETCH = 4096; - - private static final Logger _logger = LoggerFactory.getLogger(AMQChannel.class); - private final DefaultQueueAssociationClearingTask - _defaultQueueAssociationClearingTask = new DefaultQueueAssociationClearingTask(); - - //TODO use Broker property to configure message authorization requirements - private boolean _messageAuthorizationRequired = Boolean.getBoolean(BrokerProperties.PROPERTY_MSG_AUTH); - - private final int _channelId; - - - private final Pre0_10CreditManager _creditManager; - private final FlowCreditManager _noAckCreditManager; - - /** - * 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 AMQQueue<?> _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 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 AMQProtocolEngine _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 TransactionTimeoutHelper _transactionTimeoutHelper; - 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 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; - - public AMQChannel(AMQProtocolEngine 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.getAuthorizedSubject().getPrincipals(), - connection.getAuthorizedSubject().getPublicCredentials(), - connection.getAuthorizedSubject().getPrivateCredentials()); - _subject.getPrincipals().add(new SessionPrincipal(this)); - _maxUncommittedInMemorySize = connection.getVirtualHost().getContextValue(Long.class, Connection.MAX_UNCOMMITTED_IN_MEMORY_SIZE); - _logSubject = new ChannelLogSubject(this); - - _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); - - _clientDeliveryMethod = connection.createDeliveryMethod(_channelId); - - _transactionTimeoutHelper = new TransactionTimeoutHelper(_logSubject, new CloseAction() - { - @Override - public void doTimeoutAction(String reason) - { - try - { - closeConnection(reason); - } - catch (AMQException e) - { - throw new ConnectionScopedRuntimeException(e); - } - } - }, getVirtualHost()); - - Subject.doAs(_subject, new PrivilegedAction<Object>() - { - @Override - public Object run() - { - getVirtualHost().getEventLogger().message(ChannelMessages.CREATE()); - - return null; - } - }); - - } - - private boolean performGet(final AMQQueue 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); - 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.getLastReceivedTime(); - } - }); - _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) - { - String routingKey = info.getRoutingKey() == null ? null : info.getRoutingKey().asString(); - VirtualHostImpl virtualHost = getVirtualHost(); - SecurityManager securityManager = virtualHost.getSecurityManager(); - - securityManager.authorisePublish(info.isImmediate(), routingKey, e.getName(), virtualHost.getName()); - - _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()) - { - if(_confirmOnPublish) - { - _confirmedMessageCounter++; - } - Runnable finallyAction = null; - ContentHeaderBody contentHeader = _currentMessage.getContentHeader(); - - long bodySize = _currentMessage.getSize(); - long timestamp = contentHeader.getProperties().getTimestamp(); - - try - { - - final MessagePublishInfo messagePublishInfo = _currentMessage.getMessagePublishInfo(); - final MessageDestination destination = _currentMessage.getDestination(); - - final MessageMetaData messageMetaData = - new MessageMetaData(messagePublishInfo, - contentHeader, - getConnection().getLastReceivedTime()); - - final MessageHandle<MessageMetaData> handle = _messageStore.addMessage(messageMetaData); - int bodyCount = _currentMessage.getBodyCount(); - if(bodyCount > 0) - { - long bodyLengthReceived = 0; - for(int i = 0 ; i < bodyCount ; i++) - { - ContentBody contentChunk = _currentMessage.getContentChunk(i); - handle.addContent(ByteBuffer.wrap(contentChunk.getPayload())); - bodyLengthReceived += contentChunk.getSize(); - } - } - final StoredMessage<MessageMetaData> storedMessage = handle.allContentAdded(); - - final AMQMessage amqMessage = createAMQMessage(storedMessage); - MessageReference reference = amqMessage.newReference(); - try - { - - _currentMessage = null; - - if(!checkMessageUserId(contentHeader)) - { - if(_confirmOnPublish) - { - _connection.writeFrame(new AMQFrame(_channelId, new BasicNackBody(_confirmedMessageCounter, false, false))); - } - _transaction.addPostTransactionAction(new WriteReturnAction(AMQConstant.ACCESS_REFUSED, "Access Refused", amqMessage)); - } - else - { - 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; - } - } - - } - - 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()) - { - getVirtualHost().getEventLogger() - .message(_logSubject, 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 AMQProtocolEngine#isCloseWhenNoRoute() - */ - private Runnable handleUnroutableMessage(AMQMessage message) - { - boolean mandatory = message.isMandatory(); - - String exchangeName = message.getMessagePublishInfo().getExchange() == null - ? null : message.getMessagePublishInfo().getExchange().asString(); - String routingKey = message.getMessagePublishInfo().getRoutingKey() == null - ? null : message.getMessagePublishInfo().getRoutingKey().asString(); - - 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.closeConnection(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 - { - - getVirtualHost().getEventLogger().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.closeConnection(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 - * - * @throws org.apache.qpid.AMQException if something goes wrong - */ - 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); - - if(arguments != null && Boolean.TRUE.equals(arguments.get(AMQPFilterTypes.NO_CONSUME.getValue()))) - { - target = ConsumerTarget_0_8.createBrowserTarget(this, tag, arguments, _noAckCreditManager); - } - else if(acks) - { - target = ConsumerTarget_0_8.createAckTarget(this, tag, arguments, _creditManager); - options.add(ConsumerImpl.Option.ACQUIRES); - options.add(ConsumerImpl.Option.SEES_REQUEUES); - } - else - { - target = ConsumerTarget_0_8.createNoAckTarget(this, tag, arguments, _noAckCreditManager); - 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(); - } - final Object connectionReference = getConnectionReference(); - MessageFilter filter = new MessageFilter() - { - - @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; - } - }; - 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); - - } - - for(MessageSource source : sources) - { - ConsumerImpl sub = - source.addConsumer(target, - filterManager, - AMQMessage.class, - AMQShortString.toString(tag), - options); - 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) - { - - 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; - } - - LogMessage operationalLogMessage = cause == null ? - ChannelMessages.CLOSE() : - ChannelMessages.CLOSE_FORCED(cause.getCode(), message); - getVirtualHost().getEventLogger().message(_logSubject, operationalLogMessage); - - unsubscribeAllConsumers(); - setDefaultQueue(null); - for (Action<? super AMQChannel> task : _taskList) - { - task.performAction(this); - } - - - _transaction.rollback(); - - try - { - requeue(); - } - catch (TransportException e) - { - _logger.error("Caught TransportException whilst attempting to requeue:" + e); - } - } - - private void unsubscribeAllConsumers() - { - if (_logger.isInfoEnabled()) - { - if (!_tag2SubscriptionTargetMap.isEmpty()) - { - _logger.info("Unsubscribing all consumers on channel " + toString()); - } - else - { - _logger.info("No consumers to unsubscribe on channel " + toString()); - } - } - - for (Map.Entry<AMQShortString, ConsumerTarget_0_8> me : _tag2SubscriptionTargetMap.entrySet()) - { - if (_logger.isInfoEnabled()) - { - _logger.info("Unsubscribing consumer '" + me.getKey() + "' on channel " + toString()); - } - - Collection<ConsumerImpl> subs = me.getValue().getConsumers(); - - if(subs != null) - { - for(ConsumerImpl sub : subs) - { - sub.close(); - } - } - } - - _tag2SubscriptionTargetMap.clear(); - } - - /** - * 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. - * - */ - public 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.isInfoEnabled()) - { - _logger.info("Requeuing " + messagesToBeDelivered.size() + " unacked messages. for " + toString()); - } - - } - - for (MessageInstance unacked : messagesToBeDelivered) - { - // Mark message redelivered - unacked.setRedelivered(); - - // Ensure message is released for redelivery - unacked.release(); - } - - } - - /** - * 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(); - - } - 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. - * - */ - public 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.isInfoEnabled()) - { - if (!msgToRequeue.isEmpty()) - { - _logger.info("Preparing (" + msgToRequeue.size() + ") message to requeue to."); - } - } - - // 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(); - - } - } - - - /** - * 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 - * - */ - public 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) - { - getVirtualHost().getEventLogger().message(_logSubject, 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 : _tag2SubscriptionTargetMap.values()) - { - 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 - for (ConsumerTarget_0_8 s : _tag2SubscriptionTargetMap.values()) - { - try - { - s.getSendLock(); - } - finally - { - s.releaseSendLock(); - } - } - } - - - // Log Suspension only after we have confirmed all suspensions are - // stopped. - if (suspended) - { - getVirtualHost().getEventLogger().message(_logSubject, 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(); - } - - public 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 - for(ConsumerTarget_0_8 sub : _tag2SubscriptionTargetMap.values()) - { - sub.getSendLock(); - sub.releaseSendLock(); - } - - try - { - _transaction.rollback(); - } - finally - { - _rollingBack = false; - - _txnRejects.incrementAndGet(); - _txnStarts.incrementAndGet(); - decrementOutstandingTxnsIfNecessary(); - resetUncommittedMessages(); - } - - postRollbackTask.run(); - - for(MessageInstance entry : _resendList) - { - ConsumerImpl sub = entry.getDeliveredConsumer(); - if(sub == null || sub.isClosed()) - { - entry.release(); - } - else - { - entry.resend(); - } - } - _resendList.clear(); - - if(requiresSuspend) - { - _suspended.set(false); - for(ConsumerTarget_0_8 target : _tag2SubscriptionTargetMap.values()) - { - 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 AMQProtocolEngine getConnection() - { - return _connection; - } - - public FlowCreditManager getCreditManager() - { - return _creditManager; - } - - public void setCredit(final long prefetchSize, final int prefetchCount) - { - getVirtualHost().getEventLogger().message(ChannelMessages.PREFETCH_SIZE(prefetchSize, prefetchCount)); - _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; - } - - private boolean checkMessageUserId(ContentHeaderBody header) - { - AMQShortString userID = header.getProperties().getUserId(); - return (!_messageAuthorizationRequired || _connection.getAuthorizedPrincipal().getName().equals(userID == null? "" : userID.toString())); - - } - - @Override - public UUID getId() - { - return _id; - } - - @Override - public AMQProtocolEngine getConnectionModel() - { - return _connection; - } - - public String getClientID() - { - return String.valueOf(_connection.getContextKey()); - } - - 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 GetDeliveryMethod implements ClientDeliveryMethod - { - - private final FlowCreditManager _singleMessageCredit; - private final AMQQueue _queue; - private boolean _deliveredMessage; - - public GetDeliveryMethod(final FlowCreditManager singleMessageCredit, - final AMQQueue 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()); - long size = _connection.getProtocolOutputConverter().writeGetOk(message, - props, - AMQChannel.this.getChannelId(), - deliveryTag, - _queue.getQueueDepthMessages()); - - _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 final Collection<MessageInstance> _ackedMessages; - - public MessageAcknowledgeAction(Collection<MessageInstance> ackedMessages) - { - _ackedMessages = ackedMessages; - } - - public void postCommit() - { - try - { - for(MessageInstance entry : _ackedMessages) - { - entry.delete(); - } - } - finally - { - _ackedMessages.clear(); - } - - } - - public void onRollback() - { - // explicit rollbacks resend the message after the rollback-ok is sent - if(_rollingBack) - { - for(MessageInstance entry : _ackedMessages) - { - entry.unlockAcquisition(); - } - _resendList.addAll(_ackedMessages); - } - else - { - try - { - for(MessageInstance entry : _ackedMessages) - { - entry.release(); - } - } - finally - { - _ackedMessages.clear(); - } - } - - } - } - - 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)) - { - getVirtualHost().getEventLogger().message(_logSubject, - ChannelMessages.FLOW_ENFORCED("** All Queues **")); - - - getConnection().notifyWork(); - } - } - } - - public synchronized void unblock() - { - if(_blockingEntities.remove(this)) - { - if(_blockingEntities.isEmpty() && _blocking.compareAndSet(true,false)) - { - getVirtualHost().getEventLogger().message(_logSubject, ChannelMessages.FLOW_REMOVED()); - getConnection().notifyWork(); - } - } - } - - - public synchronized void block(AMQQueue queue) - { - if(_blockingEntities.add(queue)) - { - - if(_blocking.compareAndSet(false,true)) - { - getVirtualHost().getEventLogger().message(_logSubject, ChannelMessages.FLOW_ENFORCED(queue.getName())); - getConnection().notifyWork(); - - } - } - } - - public synchronized void unblock(AMQQueue queue) - { - if(_blockingEntities.remove(queue)) - { - if(_blockingEntities.isEmpty() && _blocking.compareAndSet(true,false) && !isClosing()) - { - getVirtualHost().getEventLogger().message(_logSubject, 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 VirtualHostImpl getVirtualHost() - { - return getConnection().getVirtualHost(); - } - - public void checkTransactionStatus(long openWarn, long openClose, long idleWarn, long idleClose) - { - _transactionTimeoutHelper.checkIdleOrOpenTimes(_transaction, openWarn, openClose, idleWarn, idleClose); - } - - /** - * Typically called from the HouseKeepingThread instead of the main receiver thread, - * therefore uses a lock to close the connection in a thread-safe manner. - */ - private void closeConnection(String reason) throws AMQException - { - _connection.closeAsync(AMQConstant.RESOURCE_ERROR, reason); - } - - public 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 = rejectedQueueEntry.routeToAlternate(new Action<MessageInstance>() - { - @Override - public void performAction(final MessageInstance requeueEntry) - { - getVirtualHost().getEventLogger().message(_logSubject, - ChannelMessages.DEADLETTERMSG(msg.getMessageNumber(), - requeueEntry.getOwningResource() - .getName())); - } - }, null); - - if(requeues == 0) - { - - final TransactionLogResource owningResource = rejectedQueueEntry.getOwningResource(); - if(owningResource instanceof AMQQueue) - { - final AMQQueue queue = (AMQQueue) owningResource; - - final Exchange altExchange = queue.getAlternateExchange(); - - if (altExchange == null) - { - _logger.debug("No alternate exchange configured for queue, must discard the message as unable to DLQ: delivery tag: " + deliveryTag); - getVirtualHost().getEventLogger().message(_logSubject, - ChannelMessages.DISCARDMSG_NOALTEXCH(msg.getMessageNumber(), - queue.getName(), - msg.getInitialRoutingAddress())); - - } - else - { - _logger.debug( - "Routing process provided no queues to enqueue the message on, must discard message as unable to DLQ: delivery tag: " - + deliveryTag); - getVirtualHost().getEventLogger().message(_logSubject, - ChannelMessages.DISCARDMSG_NOROUTE(msg.getMessageNumber(), - altExchange.getName())); - } - } - } - - } - } - - public void recordFuture(final FutureResult 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.awaitReadyForCompletion(); - cmd.complete(); - } - if(_transaction instanceof LocalTransaction) - { - ((LocalTransaction)_transaction).sync(); - } - } - - private static class AsyncCommand - { - private final FutureResult _future; - private ServerTransaction.Action _action; - - public AsyncCommand(final FutureResult future, final ServerTransaction.Action action) - { - _future = future; - _action = action; - } - - void awaitReadyForCompletion() - { - _future.waitForCompletion(); - } - - void complete() - { - if(!_future.isComplete()) - { - _future.waitForCompletion(); - } - _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) - { - - } - } - - 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.closeConnection(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; - VirtualHostImpl<?, ?, ?> vHost = _connection.getVirtualHost(); - sync(); - String queueName = queue == null ? null : queue.asString(); - - MessageSource queue1 = queueName == null ? getDefaultQueue() : vHost.getQueue(queueName); - final Collection<MessageSource> sources = new HashSet<>(); - if (queue1 != null) - { - sources.add(queue1); - } - else if (vHost.getContextValue(Boolean.class, "qpid.enableMultiQueueConsumers") - && 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.getMessageSource(sourceName); - if (source == null) - { - sources.clear(); - break; - } - else - { - sources.add(source); - } - } - } - queueName = arguments.get("x-multiqueue").toString(); - } - - 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.closeConnection(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.closeConnection(AMQConstant.NOT_ALLOWED, - "Non-unique consumer tag, '" + consumerTag1 - + "'", _channelId); - } - catch (AMQInvalidArgumentException ise) - { - _connection.closeConnection(AMQConstant.ARGUMENT_INVALID, ise.getMessage(), _channelId); - - - } - catch (AMQQueue.ExistingExclusiveConsumer e) - { - _connection.closeConnection(AMQConstant.ACCESS_REFUSED, - "Cannot subscribe to queue '" - + queue1.getName() - + "' as it already has an existing exclusive consumer", _channelId); - - } - catch (AMQQueue.ExistingConsumerPreventsExclusive e) - { - _connection.closeConnection(AMQConstant.ACCESS_REFUSED, - "Cannot subscribe to queue '" - + queue1.getName() - + "' exclusively as it already has a consumer", _channelId); - - } - catch (AccessControlException e) - { - _connection.closeConnection(AMQConstant.ACCESS_REFUSED, "Cannot subscribe to queue '" - + queue1.getName() - + "' permission denied", _channelId); - - } - catch (MessageSource.ConsumerAccessRefused consumerAccessRefused) - { - _connection.closeConnection(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 + " ]"); - } - - VirtualHostImpl vHost = _connection.getVirtualHost(); - sync(); - AMQQueue queue = queueName == null ? getDefaultQueue() : vHost.getQueue(queueName.toString()); - if (queue == null) - { - _logger.info("No queue for '" + queueName + "'"); - if (queueName != null) - { - _connection.closeConnection(AMQConstant.NOT_FOUND, "No such queue, '" + queueName + "'", _channelId); - - } - else - { - _connection.closeConnection(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.closeConnection(AMQConstant.ACCESS_REFUSED, e.getMessage(), _channelId); - } - catch (MessageSource.ExistingExclusiveConsumer e) - { - _connection.closeConnection(AMQConstant.NOT_ALLOWED, "Queue has an exclusive consumer", _channelId); - } - catch (MessageSource.ExistingConsumerPreventsExclusive e) - { - _connection.closeConnection(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.closeConnection(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 + " ]"); - } - - - - VirtualHostImpl vHost = _connection.getVirtualHost(); - - if(blockingTimeoutExceeded()) - { - getVirtualHost().getEventLogger().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.getMessageDestination(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.closeConnection(AMQConstant.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 + " ]"); - } - - 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 Java 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); - _logger.debug("maxDeliveryCountEnabled: " - + maxDeliveryCountEnabled - + " deliveryTag " - + deliveryTag); - if (maxDeliveryCountEnabled) - { - final boolean deliveredTooManyTimes = isDeliveredTooManyTimes(deliveryTag); - _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 byte[] data) - { - if(_logger.isDebugEnabled()) - { - _logger.debug("RECV[" + _channelId + "] MessageContent[" +" data: " + hex(data,_connection.getBinaryDataLimit()) + " ] "); - } - - if(hasCurrentMessage()) - { - publishContentBody(new ContentBody(data)); - } - else - { - _connection.closeConnection(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.closeConnection(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); - _logger.debug("maxDeliveryCountEnabled: " - + maxDeliveryCountEnabled - + " deliveryTag " - + deliveryTag); - if (maxDeliveryCountEnabled) - { - final boolean deliveredTooManyTimes = isDeliveredTooManyTimes(deliveryTag); - _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 + " ]"); - } - - VirtualHostImpl virtualHost = _connection.getVirtualHost(); - MethodRegistry methodRegistry = _connection.getMethodRegistry(); - - sync(); - - int replyCode; - String replyText; - - if (isDefaultExchange(exchangeName)) - { - if (routingKey == null) - { - if (queueName == null) - { - replyCode = virtualHost.getQueues().isEmpty() - ? ExchangeBoundOkBody.NO_BINDINGS - : ExchangeBoundOkBody.OK; - replyText = null; - - } - else - { - AMQQueue queue = virtualHost.getQueue(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.getQueue(routingKey.toString()) == null - ? ExchangeBoundOkBody.NO_QUEUE_BOUND_WITH_RK - : ExchangeBoundOkBody.OK; - replyText = null; - } - else - { - AMQQueue queue = virtualHost.getQueue(queueName.toString()); - 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 - { - ExchangeImpl exchange = virtualHost.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 - { - - AMQQueue queue = virtualHost.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) - { - AMQQueue queue = virtualHost.getQueue(queueName.toString()); - if (queue == null) - { - replyCode = ExchangeBoundOkBody.QUEUE_NOT_FOUND; - replyText = "Queue '" + queueName + "' not found"; - } - else - { - String bindingKey = routingKey == null ? null : routingKey.asString(); - 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.asString())) - { - - 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 + " ]"); - } - - ExchangeImpl exchange; - VirtualHostImpl<?, ?, ?> virtualHost = _connection.getVirtualHost(); - if (isDefaultExchange(exchangeName)) - { - if (!new AMQShortString(ExchangeDefaults.DIRECT_EXCHANGE_CLASS).equals(type)) - { - _connection.closeConnection(AMQConstant.NOT_ALLOWED, "Attempt to redeclare default exchange: " - + " of type " - + ExchangeDefaults.DIRECT_EXCHANGE_CLASS - + " to " + type + ".", getChannelId()); - } - else if (!nowait) - { - MethodRegistry methodRegistry = _connection.getMethodRegistry(); - AMQMethodBody responseBody = methodRegistry.createExchangeDeclareOkBody(); - sync(); - _connection.writeFrame(responseBody.generateFrame(getChannelId())); - } - - } - else - { - if (passive) - { - exchange = virtualHost.getExchange(exchangeName.toString()); - if (exchange == null) - { - closeChannel(AMQConstant.NOT_FOUND, "Unknown exchange: '" + exchangeName + "'"); - } - else if (!(type == null || type.length() == 0) && !exchange.getType().equals(type.asString())) - { - - _connection.closeConnection(AMQConstant.NOT_ALLOWED, "Attempt to redeclare exchange: '" - + exchangeName - + "' of type " - + exchange.getType() - + " to " - + type - + ".", getChannelId()); - } - else if (!nowait) - { - MethodRegistry methodRegistry = _connection.getMethodRegistry(); - AMQMethodBody responseBody = methodRegistry.createExchangeDeclareOkBody(); - sync(); - _connection.writeFrame(responseBody.generateFrame(getChannelId())); - } - - } - else - { - try - { - String name = exchangeName == null ? null : exchangeName.intern().toString(); - String typeString = type == null ? null : type.intern().toString(); - - 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.createExchange(attributes); - - if (!nowait) - { - sync(); - MethodRegistry methodRegistry = _connection.getMethodRegistry(); - AMQMethodBody responseBody = methodRegistry.createExchangeDeclareOkBody(); - _connection.writeFrame(responseBody.generateFrame( - getChannelId())); - } - - } - catch (ReservedExchangeNameException e) - { - _connection.closeConnection(AMQConstant.NOT_ALLOWED, - "Attempt to declare exchange: '" + exchangeName + - "' which begins with reserved prefix.", getChannelId()); - - - } - catch (ExchangeExistsException e) - { - exchange = e.getExistingExchange(); - if (!new AMQShortString(exchange.getType()).equals(type)) - { - _connection.closeConnection(AMQConstant.NOT_ALLOWED, "Attempt to redeclare exchange: '" - + exchangeName + "' of type " - + exchange.getType() - + " to " + type + ".", getChannelId()); - - } - else - { - if (!nowait) - { - sync(); - MethodRegistry methodRegistry = _connection.getMethodRegistry(); - AMQMethodBody responseBody = methodRegistry.createExchangeDeclareOkBody(); - _connection.writeFrame(responseBody.generateFrame( - getChannelId())); - } - } - } - catch (NoFactoryForTypeException e) - { - _connection.closeConnection(AMQConstant.COMMAND_INVALID, "Unknown exchange type '" - + e.getType() - + "' for exchange '" - + exchangeName - + "'", getChannelId()); - - } - catch (AccessControlException e) - { - _connection.closeConnection(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.closeConnection(AMQConstant.NOT_FOUND, message, getChannelId()); - - } - catch (IllegalArgumentException e) - { - _connection.closeConnection(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 + " ]"); - } - - - VirtualHostImpl virtualHost = _connection.getVirtualHost(); - sync(); - try - { - - if (isDefaultExchange(exchangeStr)) - { - _connection.closeConnection(AMQConstant.NOT_ALLOWED, - "Default Exchange cannot be deleted", getChannelId()); - - } - - else - { - final String exchangeName = exchangeStr.toString(); - - final ExchangeImpl exchange = virtualHost.getExchange(exchangeName); - if (exchange == null) - { - closeChannel(AMQConstant.NOT_FOUND, "No such exchange: '" + exchangeStr + "'"); - } - else - { - virtualHost.removeExchange(exchange, !ifUnused); - - 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.closeConnection(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 + " ]"); - } - - VirtualHostImpl virtualHost = _connection.getVirtualHost(); - AMQQueue<?> queue; - if (queueName == null) - { - - queue = getDefaultQueue(); - - if (queue != null) - { - if (routingKey == null) - { - routingKey = AMQShortString.valueOf(queue.getName()); - } - else - { - routingKey = routingKey.intern(); - } - } - } - else - { - queue = virtualHost.getQueue(queueName.toString()); - routingKey = routingKey == null ? AMQShortString.EMPTY_STRING : routingKey.intern(); - } - - 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.closeConnection(AMQConstant.NOT_ALLOWED, - "Cannot bind the queue '" + queueName + "' to the default exchange", getChannelId()); - - } - else - { - - final String exchangeName = exchange.toString(); - - final ExchangeImpl exch = virtualHost.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.isInfoEnabled()) - { - _logger.info("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.closeConnection(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 + " ]"); - } - - VirtualHostImpl virtualHost = _connection.getVirtualHost(); - - 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.intern(); - } - - AMQQueue<?> queue; - - //TODO: do we need to check that the queue already exists with exactly the same "configuration"? - - - if (passive) - { - queue = virtualHost.getQueue(queueName.toString()); - if (queue == null) - { - closeChannel(AMQConstant.NOT_FOUND, - "Queue: '" - + queueName - + "' not found on VirtualHost '" - + virtualHost.getName() - + "'."); - } - else - { - if (!queue.verifySessionAccess(this)) - { - _connection.closeConnection(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())); - - _logger.info("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; - } - - attributes.put(Queue.EXCLUSIVE, exclusivityPolicy); - attributes.put(Queue.LIFETIME_POLICY, lifetimePolicy); - - - queue = virtualHost.createQueue(attributes); - - setDefaultQueue(queue); - - if (!nowait) - { - sync(); - MethodRegistry methodRegistry = _connection.getMethodRegistry(); - QueueDeclareOkBody responseBody = - methodRegistry.createQueueDeclareOkBody(queueName, - queue.getQueueDepthMessages(), - queue.getConsumerCount()); - _connection.writeFrame(responseBody.generateFrame(getChannelId())); - - _logger.info("Queue " + queueName + " declared successfully"); - } - } - catch (QueueExistsException qe) - { - - queue = qe.getExistingQueue(); - - if (!queue.verifySessionAccess(this)) - { - _connection.closeConnection(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.DELETE_ON_NO_OUTBOUND_LINKS) - || (!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())); - - _logger.info("Queue " + queueName + " declared successfully"); - } - } - } - catch (AccessControlException e) - { - _connection.closeConnection(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 + " ]"); - } - - VirtualHostImpl virtualHost = _connection.getVirtualHost(); - sync(); - AMQQueue queue; - if (queueName == null) - { - - //get the default queue on the channel: - queue = getDefaultQueue(); - } - else - { - queue = virtualHost.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.closeConnection(AMQConstant.NOT_ALLOWED, "Queue '" - + queue.getName() - + "' is exclusive, but not created on this Connection.", getChannelId()); - - } - else - { - try - { - int purged = virtualHost.removeQueue(queue); - - MethodRegistry methodRegistry = _connection.getMethodRegistry(); - QueueDeleteOkBody responseBody = methodRegistry.createQueueDeleteOkBody(purged); - _connection.writeFrame(responseBody.generateFrame(getChannelId())); - } - catch (AccessControlException e) - { - _connection.closeConnection(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 + " ]"); - } - - VirtualHostImpl virtualHost = _connection.getVirtualHost(); - AMQQueue queue = null; - if (queueName == null && (queue = getDefaultQueue()) == null) - { - - _connection.closeConnection(AMQConstant.NOT_ALLOWED, "No queue specified.", getChannelId()); - } - else if ((queueName != null) && (queue = virtualHost.getQueue(queueName.toString())) == null) - { - closeChannel(AMQConstant.NOT_FOUND, "Queue '" + queueName + "' does not exist."); - } - else if (!queue.verifySessionAccess(this)) - { - _connection.closeConnection(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.closeConnection(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 + " ]"); - } - - VirtualHostImpl virtualHost = _connection.getVirtualHost(); - - - final boolean useDefaultQueue = queueName == null; - final AMQQueue queue = useDefaultQueue - ? getDefaultQueue() - : virtualHost.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.closeConnection(AMQConstant.NOT_ALLOWED, "Cannot unbind the queue '" - + queue.getName() - + "' from the default exchange", getChannelId()); - - } - else - { - - final ExchangeImpl exch = virtualHost.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.closeConnection(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() - { - MethodRegistry methodRegistry = _connection.getMethodRegistry(); - AMQMethodBody responseBody = methodRegistry.createTxCommitOkBody(); - _connection.writeFrame(responseBody.generateFrame(_channelId)); - } - }, 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(AMQQueue<?> queue) - { - AMQQueue<?> currentDefaultQueue = _defaultQueue; - if (queue != currentDefaultQueue) - { - if (currentDefaultQueue != null) - { - currentDefaultQueue.removeDeleteTask(_defaultQueueAssociationClearingTask); - } - if (queue != null) - { - queue.addDeleteTask(_defaultQueueAssociationClearingTask); - } - } - _defaultQueue = queue; - } - - private AMQQueue getDefaultQueue() - { - return _defaultQueue; - } - - private class DefaultQueueAssociationClearingTask implements Action<AMQQueue> - { - @Override - public void performAction(final AMQQueue queue) - { - if ( queue == _defaultQueue) - { - _defaultQueue = null; - } - } - } - - @Override - public void processPending() - { - - boolean desiredBlockingState = _blocking.get(); - if (desiredBlockingState != _wireBlockingState) - { - _wireBlockingState = desiredBlockingState; - flow(!desiredBlockingState); - _blockTime = desiredBlockingState ? System.currentTimeMillis() : 0; - } - - for(ConsumerTarget target : _tag2SubscriptionTargetMap.values()) - { - target.processPending(); - } - } -} diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQMessage.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQMessage.java deleted file mode 100644 index faa3831d70..0000000000 --- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQMessage.java +++ /dev/null @@ -1,108 +0,0 @@ -/* - * - * 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 org.apache.qpid.framing.AMQShortString; -import org.apache.qpid.framing.ContentHeaderBody; -import org.apache.qpid.framing.MessagePublishInfo; -import org.apache.qpid.server.message.AMQMessageHeader; -import org.apache.qpid.server.message.AbstractServerMessageImpl; -import org.apache.qpid.server.store.StoredMessage; - -/** - * A deliverable message. - */ -public class AMQMessage extends AbstractServerMessageImpl<AMQMessage, MessageMetaData> -{ - private final long _size; - - public AMQMessage(StoredMessage<MessageMetaData> handle) - { - this(handle, null); - } - - public AMQMessage(StoredMessage<MessageMetaData> handle, Object connectionReference) - { - super(handle, connectionReference); - _size = handle.getMetaData().getContentSize(); - } - - public MessageMetaData getMessageMetaData() - { - return getStoredMessage().getMetaData(); - } - - public ContentHeaderBody getContentHeaderBody() - { - return getMessageMetaData().getContentHeaderBody(); - } - - public String getInitialRoutingAddress() - { - MessageMetaData messageMetaData = getMessageMetaData(); - if (messageMetaData != null) - { - AMQShortString routingKey = messageMetaData.getMessagePublishInfo().getRoutingKey(); - if (routingKey != null) - { - return routingKey.asString(); - } - } - return null; - } - - public AMQMessageHeader getMessageHeader() - { - return getMessageMetaData().getMessageHeader(); - } - - public MessagePublishInfo getMessagePublishInfo() - { - return getMessageMetaData().getMessagePublishInfo(); - } - - public long getArrivalTime() - { - return getMessageMetaData().getArrivalTime(); - } - - public long getSize() - { - return _size; - } - - public boolean isImmediate() - { - return getMessagePublishInfo().isImmediate(); - } - - public boolean isMandatory() - { - return getMessagePublishInfo().isMandatory(); - } - - public long getExpiration() - { - return getMessageHeader().getExpiration(); - } - - -} diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQNoMethodHandlerException.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQNoMethodHandlerException.java deleted file mode 100644 index 42250190b4..0000000000 --- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQNoMethodHandlerException.java +++ /dev/null @@ -1,42 +0,0 @@ -/*
- *
- * 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 org.apache.qpid.AMQException;
-import org.apache.qpid.framing.AMQMethodBody;
-import org.apache.qpid.protocol.AMQMethodEvent;
-
-/**
- * AMQNoMethodHandlerException represents the case where no method handler exists to handle an AQMP method.
- *
- * <p>
- * TODO Not an AMQP exception as no status code.
- * <p>
- * TODO Missing method handler. Unlikely to ever happen, and if it does its a coding error. Consider replacing with a
- * Runtime.
- */
-public class AMQNoMethodHandlerException extends AMQException
-{
- public AMQNoMethodHandlerException(AMQMethodEvent<AMQMethodBody> evt)
- {
- super("AMQMethodEvent " + evt + " was not processed by any listener on Broker.");
- }
-}
diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngine.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngine.java deleted file mode 100644 index b1f2a6e841..0000000000 --- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngine.java +++ /dev/null @@ -1,2033 +0,0 @@ -/* - * - * 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 java.io.IOException; -import java.lang.reflect.InvocationHandler; -import java.lang.reflect.Method; -import java.lang.reflect.Proxy; -import java.net.InetSocketAddress; -import java.net.SocketAddress; -import java.nio.ByteBuffer; -import java.security.AccessControlException; -import java.security.AccessController; -import java.security.Principal; -import java.security.PrivilegedAction; -import java.util.ArrayList; -import java.util.HashMap; -import java.util.HashSet; -import java.util.List; -import java.util.Map; -import java.util.Queue; -import java.util.Set; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentLinkedQueue; -import java.util.concurrent.CopyOnWriteArrayList; -import java.util.concurrent.atomic.AtomicBoolean; -import java.util.concurrent.atomic.AtomicReference; - -import javax.security.auth.Subject; -import javax.security.sasl.SaslException; -import javax.security.sasl.SaslServer; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import org.apache.qpid.AMQConnectionException; -import org.apache.qpid.AMQException; -import org.apache.qpid.codec.AMQDecoder; -import org.apache.qpid.common.QpidProperties; -import org.apache.qpid.common.ServerPropertyNames; -import org.apache.qpid.framing.*; -import org.apache.qpid.properties.ConnectionStartProperties; -import org.apache.qpid.protocol.AMQConstant; -import org.apache.qpid.server.protocol.ServerProtocolEngine; -import org.apache.qpid.server.configuration.BrokerProperties; -import org.apache.qpid.server.connection.ConnectionPrincipal; -import org.apache.qpid.server.consumer.ConsumerImpl; -import org.apache.qpid.server.logging.EventLogger; -import org.apache.qpid.server.logging.LogSubject; -import org.apache.qpid.server.logging.messages.ConnectionMessages; -import org.apache.qpid.server.logging.subjects.ConnectionLogSubject; -import org.apache.qpid.server.message.InstanceProperties; -import org.apache.qpid.server.message.ServerMessage; -import org.apache.qpid.server.model.Broker; -import org.apache.qpid.server.model.Consumer; -import org.apache.qpid.server.model.Port; -import org.apache.qpid.server.model.State; -import org.apache.qpid.server.model.Transport; -import org.apache.qpid.server.model.port.AmqpPort; -import org.apache.qpid.server.protocol.AMQConnectionModel; -import org.apache.qpid.server.protocol.AMQSessionModel; -import org.apache.qpid.server.protocol.SessionModelListener; -import org.apache.qpid.server.security.SubjectCreator; -import org.apache.qpid.server.security.auth.AuthenticatedPrincipal; -import org.apache.qpid.server.security.auth.SubjectAuthenticationResult; -import org.apache.qpid.server.stats.StatisticsCounter; -import org.apache.qpid.server.store.StoreException; -import org.apache.qpid.server.util.Action; -import org.apache.qpid.server.util.ConnectionScopedRuntimeException; -import org.apache.qpid.server.util.ServerScopedRuntimeException; -import org.apache.qpid.server.virtualhost.VirtualHostImpl; -import org.apache.qpid.transport.ByteBufferSender; -import org.apache.qpid.transport.SenderClosedException; -import org.apache.qpid.transport.SenderException; -import org.apache.qpid.transport.TransportException; -import org.apache.qpid.transport.network.NetworkConnection; - -public class AMQProtocolEngine implements ServerProtocolEngine, - AMQConnectionModel<AMQProtocolEngine, AMQChannel>, - ServerMethodProcessor<ServerChannelMethodProcessor> -{ - - - - enum ConnectionState - { - INIT, - AWAIT_START_OK, - AWAIT_SECURE_OK, - AWAIT_TUNE_OK, - AWAIT_OPEN, - OPEN - } - - private static final Logger _logger = LoggerFactory.getLogger(AMQProtocolEngine.class); - - // to save boxing the channelId and looking up in a map... cache in an array the low numbered - // channels. This value must be of the form 2^x - 1. - private static final int CHANNEL_CACHE_SIZE = 0xff; - private static final String BROKER_DEBUG_BINARY_DATA_LENGTH = "broker.debug.binaryDataLength"; - private static final int DEFAULT_DEBUG_BINARY_DATA_LENGTH = 80; - private static final long AWAIT_CLOSED_TIMEOUT = 60000; - private final AmqpPort<?> _port; - private final long _creationTime; - private final AtomicBoolean _stateChanged = new AtomicBoolean(); - private final AtomicReference<Action<ServerProtocolEngine>> _workListener = new AtomicReference<>(); - - private AMQShortString _contextKey; - - private String _clientVersion = null; - private String _clientProduct = null; - private String _remoteProcessPid = null; - - private volatile VirtualHostImpl<?,?,?> _virtualHost; - - private final Map<Integer, AMQChannel> _channelMap = - new HashMap<>(); - private final CopyOnWriteArrayList<SessionModelListener> _sessionListeners = - new CopyOnWriteArrayList<>(); - - private final AMQChannel[] _cachedChannels = new AMQChannel[CHANNEL_CACHE_SIZE + 1]; - - private ConnectionState _state = ConnectionState.INIT; - - /** - * The channels that the latest call to {@link #received(ByteBuffer)} applied to. - * Used so we know which channels we need to call {@link AMQChannel#receivedComplete()} - * on after handling the frames. - */ - private final Set<AMQChannel> _channelsForCurrentMessage = new HashSet<>(); - - private AMQDecoder _decoder; - - private SaslServer _saslServer; - - private volatile boolean _closed; - - private long _maxNoOfChannels; - - private ProtocolVersion _protocolVersion = ProtocolVersion.getLatestSupportedVersion(); - private final MethodRegistry _methodRegistry = new MethodRegistry(_protocolVersion); - private final List<Action<? super AMQProtocolEngine>> _connectionCloseTaskList = - new CopyOnWriteArrayList<>(); - - private final Queue<Action<? super AMQProtocolEngine>> _asyncTaskList = - new ConcurrentLinkedQueue<>(); - - private Map<Integer, Long> _closingChannelsList = new ConcurrentHashMap<>(); - private ProtocolOutputConverter _protocolOutputConverter; - private final Subject _authorizedSubject = new Subject(); - - private final long _connectionId; - private Object _reference = new Object(); - - private LogSubject _logSubject; - - private volatile long _lastIoTime; - - private long _writtenBytes; - - private int _maxFrameSize; - private final AtomicBoolean _closing = new AtomicBoolean(false); - - private final StatisticsCounter _messagesDelivered, _dataDelivered, _messagesReceived, _dataReceived; - - private NetworkConnection _network; - private ByteBufferSender _sender; - - private volatile boolean _deferFlush; - private volatile long _lastReceivedTime = System.currentTimeMillis(); - private volatile long _lastWriteTime = System.currentTimeMillis(); - private boolean _blocking; - - private final Broker<?> _broker; - private final Transport _transport; - - private volatile boolean _closeWhenNoRoute; - private volatile boolean _stopped; - private long _readBytes; - private boolean _authenticated; - private boolean _compressionSupported; - private int _messageCompressionThreshold; - private int _currentClassId; - private int _currentMethodId; - private int _binaryDataLimit; - private long _maxMessageSize; - private volatile boolean _transportBlockedForWriting; - - private final AtomicReference<Thread> _messageAssignmentSuspended = new AtomicReference<>(); - - - @Override - public boolean isMessageAssignmentSuspended() - { - Thread lock = _messageAssignmentSuspended.get(); - return lock != null && _messageAssignmentSuspended.get() != Thread.currentThread(); - } - - @Override - public void setMessageAssignmentSuspended(final boolean messageAssignmentSuspended) - { - _messageAssignmentSuspended.set(messageAssignmentSuspended ? Thread.currentThread() : null); - for(AMQSessionModel<?,?> session : getSessionModels()) - { - for (Consumer<?> consumer : session.getConsumers()) - { - ConsumerImpl consumerImpl = (ConsumerImpl) consumer; - if (!messageAssignmentSuspended) - { - consumerImpl.getTarget().notifyCurrentState(); - } - else - { - // ensure that by the time the method returns, no consumer can be in the process of - // delivering a message. - consumerImpl.getSendLock(); - consumerImpl.releaseSendLock(); - } - } - } - } - - - public AMQProtocolEngine(Broker<?> broker, - final NetworkConnection network, - final long connectionId, - AmqpPort<?> port, - Transport transport) - { - _broker = broker; - _port = port; - _transport = transport; - _maxNoOfChannels = broker.getConnection_sessionCountLimit(); - _decoder = new BrokerDecoder(this); - _connectionId = connectionId; - _logSubject = new ConnectionLogSubject(this); - _binaryDataLimit = _broker.getContextKeys(false).contains(BROKER_DEBUG_BINARY_DATA_LENGTH) - ? _broker.getContextValue(Integer.class, BROKER_DEBUG_BINARY_DATA_LENGTH) - : DEFAULT_DEBUG_BINARY_DATA_LENGTH; - - int maxMessageSize = port.getContextValue(Integer.class, AmqpPort.PORT_MAX_MESSAGE_SIZE); - _maxMessageSize = (maxMessageSize > 0) ? (long) maxMessageSize : Long.MAX_VALUE; - - _authorizedSubject.getPrincipals().add(new ConnectionPrincipal(this)); - runAsSubject(new PrivilegedAction<Void>() - { - - @Override - public Void run() - { - setNetworkConnection(network); - - getEventLogger().message(ConnectionMessages.OPEN(null, null, null, null, false, false, false, false)); - - _closeWhenNoRoute = _broker.getConnection_closeWhenNoRoute(); - - return null; - } - }); - - _messagesDelivered = new StatisticsCounter("messages-delivered-" + _connectionId); - _dataDelivered = new StatisticsCounter("data-delivered-" + _connectionId); - _messagesReceived = new StatisticsCounter("messages-received-" + _connectionId); - _dataReceived = new StatisticsCounter("data-received-" + _connectionId); - _creationTime = System.currentTimeMillis(); - } - - private <T> T runAsSubject(PrivilegedAction<T> action) - { - return Subject.doAs(getAuthorizedSubject(), action); - } - - private boolean runningAsSubject() - { - return getAuthorizedSubject().equals(Subject.getSubject(AccessController.getContext())); - } - - @Override - public Subject getSubject() - { - return _authorizedSubject; - } - - @Override - public boolean isTransportBlockedForWriting() - { - return _transportBlockedForWriting; - } - - @Override - public void setTransportBlockedForWriting(final boolean blocked) - { - _transportBlockedForWriting = blocked; - for(AMQChannel channel : _channelMap.values()) - { - channel.transportStateChanged(); - } - } - - public void setNetworkConnection(NetworkConnection network) - { - setNetworkConnection(network, network.getSender()); - } - - public void setNetworkConnection(NetworkConnection network, ByteBufferSender sender) - { - _network = network; - _sender = sender; - } - - public void setMaxFrameSize(int frameMax) - { - _maxFrameSize = frameMax; - _decoder.setMaxFrameSize(frameMax); - } - - public long getMaxFrameSize() - { - return _maxFrameSize; - } - - public boolean isClosing() - { - return _closing.get(); - } - - - - public ClientDeliveryMethod createDeliveryMethod(int channelId) - { - return new WriteDeliverMethod(channelId); - } - - public void received(final ByteBuffer msg) - { - Subject.doAs(_authorizedSubject, new PrivilegedAction<Void>() - { - @Override - public Void run() - { - - final long arrivalTime = System.currentTimeMillis(); - if (!_authenticated && - (arrivalTime - _creationTime) > _port.getContextValue(Long.class, - Port.CONNECTION_MAXIMUM_AUTHENTICATION_DELAY)) - { - _logger.warn("Connection has taken more than " - + _port.getContextValue(Long.class, Port.CONNECTION_MAXIMUM_AUTHENTICATION_DELAY) - + "ms to establish identity. Closing as possible DoS."); - getEventLogger().message(ConnectionMessages.IDLE_CLOSE()); - closeNetworkConnection(); - } - _lastReceivedTime = arrivalTime; - _lastIoTime = arrivalTime; - _readBytes += msg.remaining(); - - try - { - _decoder.decodeBuffer(msg); - receivedComplete(); - } - catch (ConnectionScopedRuntimeException e) - { - _logger.error("Unexpected exception", e); - closeNetworkConnection(); - } - catch (AMQProtocolVersionException e) - { - _logger.error("Unexpected protocol version", e); - closeNetworkConnection(); - } - catch (SenderClosedException e) - { - _logger.debug("Sender was closed abruptly, closing network.", e); - closeNetworkConnection(); - } - catch (SenderException e) - { - _logger.info("Unexpected exception on send, closing network.", e); - closeNetworkConnection(); - } - catch (TransportException e) - { - _logger.error("Unexpected transport exception", e); - closeNetworkConnection(); - } - catch (AMQFrameDecodingException e) - { - _logger.error("Frame decoding", e); - closeNetworkConnection(); - } - catch (IOException e) - { - _logger.error("I/O Exception", e); - closeNetworkConnection(); - } - catch (StoreException e) - { - if (_virtualHost.getState() == State.ACTIVE) - { - throw e; - } - else - { - _logger.error("Store Exception ignored as virtual host no longer active", e); - } - } - return null; - } - }); - - } - - private void receivedComplete() - { - RuntimeException exception = null; - - for (AMQChannel channel : _channelsForCurrentMessage) - { - try - { - channel.receivedComplete(); - } - catch(RuntimeException exceptionForThisChannel) - { - if(exception == null) - { - exception = exceptionForThisChannel; - } - _logger.error("Error informing channel that receiving is complete. Channel: " + channel, - exceptionForThisChannel); - } - } - - _channelsForCurrentMessage.clear(); - - if(exception != null) - { - throw exception; - } - } - - - void channelRequiresSync(final AMQChannel amqChannel) - { - _channelsForCurrentMessage.add(amqChannel); - } - - private synchronized void protocolInitiationReceived(ProtocolInitiation pi) - { - // this ensures the codec never checks for a PI message again - _decoder.setExpectProtocolInitiation(false); - try - { - // Log incoming protocol negotiation request - getEventLogger().message(ConnectionMessages.OPEN(null, - pi.getProtocolMajor() + "-" + pi.getProtocolMinor(), - null, - null, - false, - true, - false, - false)); - - ProtocolVersion pv = pi.checkVersion(); // Fails if not correct - - setProtocolVersion(pv); - - StringBuilder mechanismBuilder = new StringBuilder(); - SubjectCreator subjectCreator = _port.getAuthenticationProvider().getSubjectCreator(_transport.isSecure()); - for(String mechanismName : subjectCreator.getMechanisms()) - { - if(mechanismBuilder.length() != 0) - { - mechanismBuilder.append(' '); - } - mechanismBuilder.append(mechanismName); - } - String mechanisms = mechanismBuilder.toString(); - - String locales = "en_US"; - - - FieldTable serverProperties = FieldTableFactory.newFieldTable(); - - serverProperties.setString(ServerPropertyNames.PRODUCT, - QpidProperties.getProductName()); - serverProperties.setString(ServerPropertyNames.VERSION, - QpidProperties.getReleaseVersion()); - serverProperties.setString(ServerPropertyNames.QPID_BUILD, - QpidProperties.getBuildVersion()); - serverProperties.setString(ServerPropertyNames.QPID_INSTANCE_NAME, - _broker.getName()); - serverProperties.setString(ConnectionStartProperties.QPID_CLOSE_WHEN_NO_ROUTE, - String.valueOf(_closeWhenNoRoute)); - serverProperties.setString(ConnectionStartProperties.QPID_MESSAGE_COMPRESSION_SUPPORTED, - String.valueOf(_broker.isMessageCompressionEnabled())); - serverProperties.setString(ConnectionStartProperties.QPID_CONFIRMED_PUBLISH_SUPPORTED, Boolean.TRUE.toString()); - - AMQMethodBody responseBody = getMethodRegistry().createConnectionStartBody((short) getProtocolMajorVersion(), - (short) pv.getActualMinorVersion(), - serverProperties, - mechanisms.getBytes(), - locales.getBytes()); - writeFrame(responseBody.generateFrame(0)); - _state = ConnectionState.AWAIT_START_OK; - - _sender.flush(); - - } - catch (AMQException e) - { - _logger.info("Received unsupported protocol initiation for protocol version: " + getProtocolVersion()); - - writeFrame(new ProtocolInitiation(ProtocolVersion.getLatestSupportedVersion())); - _sender.flush(); - } - } - - - - - public synchronized void writeFrame(AMQDataBlock frame) - { - if(_logger.isDebugEnabled()) - { - _logger.debug("SEND: " + frame); - } - - try - { - _writtenBytes += frame.writePayload(_sender); - } - catch (IOException e) - { - throw new ServerScopedRuntimeException(e); - } - - - final long time = System.currentTimeMillis(); - _lastIoTime = time; - _lastWriteTime = time; - - if(!_deferFlush) - { - _sender.flush(); - } - } - - public AMQShortString getContextKey() - { - return _contextKey; - } - - public void setContextKey(AMQShortString contextKey) - { - _contextKey = contextKey; - } - - public List<AMQChannel> getChannels() - { - synchronized (_channelMap) - { - return new ArrayList<>(_channelMap.values()); - } - } - - public AMQChannel getChannel(int channelId) - { - final AMQChannel channel = - ((channelId & CHANNEL_CACHE_SIZE) == channelId) ? _cachedChannels[channelId] : _channelMap.get(channelId); - if ((channel == null) || channel.isClosing()) - { - return null; - } - else - { - return channel; - } - } - - public boolean channelAwaitingClosure(int channelId) - { - return !_closingChannelsList.isEmpty() && _closingChannelsList.containsKey(channelId); - } - - public void addChannel(AMQChannel channel) - { - final int channelId = channel.getChannelId(); - - synchronized (_channelMap) - { - _channelMap.put(channel.getChannelId(), channel); - sessionAdded(channel); - if(_blocking) - { - channel.block(); - } - } - - if (((channelId & CHANNEL_CACHE_SIZE) == channelId)) - { - _cachedChannels[channelId] = channel; - } - } - - private void sessionAdded(final AMQSessionModel<?,?> session) - { - for(SessionModelListener l : _sessionListeners) - { - l.sessionAdded(session); - } - } - - private void sessionRemoved(final AMQSessionModel<?,?> session) - { - for(SessionModelListener l : _sessionListeners) - { - l.sessionRemoved(session); - } - } - - public long getMaximumNumberOfChannels() - { - return _maxNoOfChannels; - } - - public void setMaximumNumberOfChannels(Long value) - { - _maxNoOfChannels = value; - } - - - void closeChannel(AMQChannel channel) - { - closeChannel(channel, null, null, false); - } - - public void closeChannelAndWriteFrame(AMQChannel channel, AMQConstant cause, String message) - { - writeFrame(new AMQFrame(channel.getChannelId(), - getMethodRegistry().createChannelCloseBody(cause.getCode(), - AMQShortString.validValueOf(message), - _currentClassId, - _currentMethodId))); - closeChannel(channel, cause, message, true); - } - - public void closeChannel(int channelId, AMQConstant cause, String message) - { - final AMQChannel channel = getChannel(channelId); - if (channel == null) - { - throw new IllegalArgumentException("Unknown channel id"); - } - closeChannel(channel, cause, message, true); - } - - void closeChannel(AMQChannel channel, AMQConstant cause, String message, boolean mark) - { - int channelId = channel.getChannelId(); - try - { - channel.close(cause, message); - if(mark) - { - markChannelAwaitingCloseOk(channelId); - } - } - finally - { - removeChannel(channelId); - } - } - - - public void closeChannelOk(int channelId) - { - _closingChannelsList.remove(channelId); - } - - private void markChannelAwaitingCloseOk(int channelId) - { - _closingChannelsList.put(channelId, System.currentTimeMillis()); - } - - /** - * In our current implementation this is used by the clustering code. - * - * @param channelId The channel to remove - */ - public void removeChannel(int channelId) - { - AMQChannel session; - synchronized (_channelMap) - { - session = _channelMap.remove(channelId); - if ((channelId & CHANNEL_CACHE_SIZE) == channelId) - { - _cachedChannels[channelId] = null; - } - } - sessionRemoved(session); - } - - private void initHeartbeats(int delay) - { - if (delay > 0) - { - _network.setMaxWriteIdle(delay); - _network.setMaxReadIdle(BrokerProperties.HEARTBEAT_TIMEOUT_FACTOR * delay); - } - else - { - _network.setMaxWriteIdle(0); - _network.setMaxReadIdle(0); - } - } - - private void closeAllChannels() - { - try - { - RuntimeException firstException = null; - for (AMQChannel channel : getChannels()) - { - try - { - channel.close(); - } - catch (RuntimeException re) - { - if (!(re instanceof ConnectionScopedRuntimeException)) - { - _logger.error("Unexpected exception closing channel", re); - } - firstException = re; - } - } - - if (firstException != null) - { - throw firstException; - } - } - finally - { - synchronized (_channelMap) - { - _channelMap.clear(); - } - for (int i = 0; i <= CHANNEL_CACHE_SIZE; i++) - { - _cachedChannels[i] = null; - } - - } - } - - private void closeConnectionInternal(final boolean connectionDropped) - { - - if(runningAsSubject()) - { - if(_closing.compareAndSet(false,true)) - { - // force sync of outstanding async work - try - { - receivedComplete(); - } - finally - { - - finishClose(connectionDropped); - } - - } - else - { - awaitClosed(); - } - } - else - { - runAsSubject(new PrivilegedAction<Object>() - { - @Override - public Object run() - { - closeConnectionInternal(connectionDropped); - return null; - } - }); - - } - } - - private void finishClose(boolean connectionDropped) - { - if (!_closed) - { - - try - { - if (_virtualHost != null) - { - _virtualHost.getConnectionRegistry().deregisterConnection(this); - } - closeAllChannels(); - } - finally - { - try - { - for (Action<? super AMQProtocolEngine> task : _connectionCloseTaskList) - { - task.performAction(this); - } - } - finally - { - synchronized (this) - { - _closed = true; - notifyAll(); - } - getEventLogger().message(_logSubject, connectionDropped ? ConnectionMessages.DROPPED_CONNECTION() : ConnectionMessages.CLOSE()); - } - } - } - } - - private void awaitClosed() - { - synchronized(this) - { - final long endTime = System.currentTimeMillis() + AWAIT_CLOSED_TIMEOUT; - - while(!_closed && endTime > System.currentTimeMillis()) - { - try - { - wait(1000); - } - catch (InterruptedException e) - { - Thread.currentThread().interrupt(); - break; - } - } - - if (!_closed) - { - throw new ConnectionScopedRuntimeException("Connection " + this + " failed to become closed within " + AWAIT_CLOSED_TIMEOUT + "ms."); - } - } - } - - private void closeConnection(int channelId, AMQConnectionException e) - { - - if (_logger.isInfoEnabled()) - { - _logger.info("Closing connection due to: " + e); - } - closeConnection(channelId, e.getCloseFrame()); - } - - - void closeConnection(AMQConstant errorCode, - String message, int channelId) - { - - if (_logger.isInfoEnabled()) - { - _logger.info("Closing connection due to: " + message); - } - closeConnection(channelId, new AMQFrame(0, new ConnectionCloseBody(getProtocolVersion(), errorCode.getCode(), AMQShortString.validValueOf(message), _currentClassId, _currentMethodId))); - } - - private void closeConnection(int channelId, AMQFrame frame) - { - if(!_closing.get()) - { - try - { - markChannelAwaitingCloseOk(channelId); - closeConnectionInternal(false); - } - finally - { - try - { - writeFrame(frame); - } - finally - { - closeNetworkConnection(); - } - } - } - else - { - awaitClosed(); - } - } - - public void closeNetworkConnection() - { - _network.close(); - } - - public String toString() - { - return getRemoteAddress() + "(" + (getAuthorizedPrincipal() == null ? "?" : getAuthorizedPrincipal().getName() + ")"); - } - - private String getLocalFQDN() - { - SocketAddress address = _network.getLocalAddress(); - if (address instanceof InetSocketAddress) - { - return ((InetSocketAddress) address).getHostName(); - } - else - { - throw new IllegalArgumentException("Unsupported socket address class: " + address); - } - } - - public SaslServer getSaslServer() - { - return _saslServer; - } - - public void setSaslServer(SaslServer saslServer) - { - _saslServer = saslServer; - } - - public void setClientProperties(FieldTable clientProperties) - { - if (clientProperties != null) - { - String closeWhenNoRoute = clientProperties.getString(ConnectionStartProperties.QPID_CLOSE_WHEN_NO_ROUTE); - if (closeWhenNoRoute != null) - { - _closeWhenNoRoute = Boolean.parseBoolean(closeWhenNoRoute); - if(_logger.isDebugEnabled()) - { - _logger.debug("Client set closeWhenNoRoute=" + _closeWhenNoRoute + " for protocol engine " + this); - } - } - String compressionSupported = clientProperties.getString(ConnectionStartProperties.QPID_MESSAGE_COMPRESSION_SUPPORTED); - if (compressionSupported != null) - { - _compressionSupported = Boolean.parseBoolean(compressionSupported); - if(_logger.isDebugEnabled()) - { - _logger.debug("Client set compressionSupported=" + _compressionSupported + " for protocol engine " + this); - } - } - - _clientVersion = clientProperties.getString(ConnectionStartProperties.VERSION_0_8); - _clientProduct = clientProperties.getString(ConnectionStartProperties.PRODUCT); - _remoteProcessPid = clientProperties.getString(ConnectionStartProperties.PID); - - String clientId = clientProperties.getString(ConnectionStartProperties.CLIENT_ID_0_8); - if (clientId != null) - { - setContextKey(new AMQShortString(clientId)); - } - - getEventLogger().message(ConnectionMessages.OPEN(clientId, - _protocolVersion.toString(), - _clientVersion, - _clientProduct, - true, - true, - true, - true)); - } - } - - private void setProtocolVersion(ProtocolVersion pv) - { - _protocolVersion = pv; - _methodRegistry.setProtocolVersion(_protocolVersion); - _protocolOutputConverter = new ProtocolOutputConverterImpl(this); - } - - public byte getProtocolMajorVersion() - { - return _protocolVersion.getMajorVersion(); - } - - public ProtocolVersion getProtocolVersion() - { - return _protocolVersion; - } - - public byte getProtocolMinorVersion() - { - return _protocolVersion.getMinorVersion(); - } - - public MethodRegistry getRegistry() - { - return getMethodRegistry(); - } - - public VirtualHostImpl<?,?,?> getVirtualHost() - { - return _virtualHost; - } - - public void setVirtualHost(VirtualHostImpl<?,?,?> virtualHost) - { - _virtualHost = virtualHost; - - _virtualHost.getConnectionRegistry().registerConnection(this); - - - _messageCompressionThreshold = virtualHost.getContextValue(Integer.class, - Broker.MESSAGE_COMPRESSION_THRESHOLD_SIZE); - if(_messageCompressionThreshold <= 0) - { - _messageCompressionThreshold = Integer.MAX_VALUE; - } - } - - public void addDeleteTask(Action<? super AMQProtocolEngine> task) - { - _connectionCloseTaskList.add(task); - } - - public void removeDeleteTask(Action<? super AMQProtocolEngine> task) - { - _connectionCloseTaskList.remove(task); - } - - public ProtocolOutputConverter getProtocolOutputConverter() - { - return _protocolOutputConverter; - } - - public void setAuthorizedSubject(final Subject authorizedSubject) - { - if (authorizedSubject == null) - { - throw new IllegalArgumentException("authorizedSubject cannot be null"); - } - - _authenticated = true; - _authorizedSubject.getPrincipals().addAll(authorizedSubject.getPrincipals()); - _authorizedSubject.getPrivateCredentials().addAll(authorizedSubject.getPrivateCredentials()); - _authorizedSubject.getPublicCredentials().addAll(authorizedSubject.getPublicCredentials()); - - } - - public Subject getAuthorizedSubject() - { - return _authorizedSubject; - } - - public Principal getAuthorizedPrincipal() - { - - return _authorizedSubject.getPrincipals(AuthenticatedPrincipal.class).size() == 0 ? null : AuthenticatedPrincipal.getAuthenticatedPrincipalFromSubject(_authorizedSubject); - } - - public SocketAddress getRemoteAddress() - { - return _network.getRemoteAddress(); - } - - @Override - public String getRemoteProcessPid() - { - return _remoteProcessPid; - } - - public SocketAddress getLocalAddress() - { - return _network.getLocalAddress(); - } - - public Principal getPeerPrincipal() - { - return _network.getPeerPrincipal(); - } - - public MethodRegistry getMethodRegistry() - { - return _methodRegistry; - } - - public void closed() - { - try - { - try - { - closeConnectionInternal(true); - } - finally - { - closeNetworkConnection(); - } - } - catch (ConnectionScopedRuntimeException | TransportException e) - { - _logger.error("Could not close protocol engine", e); - } - } - - @Override - public void encryptedTransport() - { - } - - public void readerIdle() - { - Subject.doAs(_authorizedSubject, new PrivilegedAction<Object>() - { - @Override - public Object run() - { - getEventLogger().message(ConnectionMessages.IDLE_CLOSE()); - _network.close(); - return null; - } - }); - } - - public synchronized void writerIdle() - { - writeFrame(HeartbeatBody.FRAME); - } - - public void exception(Throwable throwable) - { - // noop - exception method is not used by new i/o layer - } - - public long getReadBytes() - { - return _readBytes; - } - - public long getWrittenBytes() - { - return _writtenBytes; - } - - public long getLastIoTime() - { - return _lastIoTime; - } - - @Override - public AmqpPort<?> getPort() - { - return _port; - } - - @Override - public Transport getTransport() - { - return _transport; - } - - @Override - public void stop() - { - _stopped = true; - } - - @Override - public boolean isStopped() - { - return _stopped; - } - - @Override - public String getVirtualHostName() - { - return _virtualHost == null ? null : _virtualHost.getName(); - } - - public long getLastReceivedTime() - { - return _lastReceivedTime; - } - - public String getClientVersion() - { - return _clientVersion; - } - - @Override - public String getClientProduct() - { - return _clientProduct; - } - - public long getSessionCountLimit() - { - return getMaximumNumberOfChannels(); - } - - public boolean isDurable() - { - return false; - } - - public long getConnectionId() - { - return _connectionId; - } - - public String getAddress() - { - return String.valueOf(getRemoteAddress()); - } - - public void closeSessionAsync(final AMQChannel session, final AMQConstant cause, final String message) - { - addAsyncTask(new Action<AMQProtocolEngine>() - { - - @Override - public void performAction(final AMQProtocolEngine object) - { - int channelId = session.getChannelId(); - closeChannel(channelId, cause, message); - - MethodRegistry methodRegistry = getMethodRegistry(); - ChannelCloseBody responseBody = - methodRegistry.createChannelCloseBody( - cause.getCode(), - AMQShortString.validValueOf(message), - 0, 0); - - writeFrame(responseBody.generateFrame(channelId)); - } - }); - - } - - public void closeAsync(final AMQConstant cause, final String message) - { - Action<AMQProtocolEngine> action = new Action<AMQProtocolEngine>() - { - @Override - public void performAction(final AMQProtocolEngine object) - { - closeConnection(0, new AMQConnectionException(cause, message, 0, 0, - getMethodRegistry(), - null)); - - } - }; - addAsyncTask(action); - } - - private void addAsyncTask(final Action<AMQProtocolEngine> action) - { - _asyncTaskList.add(action); - notifyWork(); - } - - public void block() - { - synchronized (_channelMap) - { - if(!_blocking) - { - _blocking = true; - for(AMQChannel channel : _channelMap.values()) - { - channel.block(); - } - } - } - } - - public void unblock() - { - synchronized (_channelMap) - { - if(_blocking) - { - _blocking = false; - for(AMQChannel channel : _channelMap.values()) - { - channel.unblock(); - } - } - } - } - - public boolean isClosed() - { - return _closed; - } - - public List<AMQChannel> getSessionModels() - { - return new ArrayList<>(getChannels()); - } - - public LogSubject getLogSubject() - { - return _logSubject; - } - - public void registerMessageDelivered(long messageSize) - { - _messagesDelivered.registerEvent(1L); - _dataDelivered.registerEvent(messageSize); - _virtualHost.registerMessageDelivered(messageSize); - } - - public void registerMessageReceived(long messageSize, long timestamp) - { - _messagesReceived.registerEvent(1L, timestamp); - _dataReceived.registerEvent(messageSize, timestamp); - _virtualHost.registerMessageReceived(messageSize, timestamp); - } - - public StatisticsCounter getMessageReceiptStatistics() - { - return _messagesReceived; - } - - public StatisticsCounter getDataReceiptStatistics() - { - return _dataReceived; - } - - public StatisticsCounter getMessageDeliveryStatistics() - { - return _messagesDelivered; - } - - public StatisticsCounter getDataDeliveryStatistics() - { - return _dataDelivered; - } - - public void resetStatistics() - { - _messagesDelivered.reset(); - _dataDelivered.reset(); - _messagesReceived.reset(); - _dataReceived.reset(); - } - - public String getRemoteAddressString() - { - return String.valueOf(getRemoteAddress()); - } - - public String getClientId() - { - return String.valueOf(getContextKey()); - } - - @Override - public String getRemoteContainerName() - { - return String.valueOf(getContextKey()); - } - - @Override - public void addSessionListener(final SessionModelListener listener) - { - _sessionListeners.add(listener); - } - - @Override - public void removeSessionListener(final SessionModelListener listener) - { - _sessionListeners.remove(listener); - } - - public void setDeferFlush(boolean deferFlush) - { - _deferFlush = deferFlush; - } - - @Override - public void receiveChannelOpen(final int channelId) - { - if(_logger.isDebugEnabled()) - { - _logger.debug("RECV[" + channelId + "] ChannelOpen"); - } - assertState(ConnectionState.OPEN); - - // Protect the broker against out of order frame request. - if (_virtualHost == null) - { - closeConnection(AMQConstant.COMMAND_INVALID, - "Virtualhost has not yet been set. ConnectionOpen has not been called.", channelId); - } - else if(getChannel(channelId) != null || channelAwaitingClosure(channelId)) - { - closeConnection(AMQConstant.CHANNEL_ERROR, "Channel " + channelId + " already exists", channelId); - } - else if(channelId > getMaximumNumberOfChannels()) - { - closeConnection(AMQConstant.CHANNEL_ERROR, - "Channel " + channelId + " cannot be created as the max allowed channel id is " - + getMaximumNumberOfChannels(), - channelId); - } - else - { - _logger.info("Connecting to: " + _virtualHost.getName()); - - final AMQChannel channel = new AMQChannel(this, channelId, _virtualHost.getMessageStore()); - - addChannel(channel); - - ChannelOpenOkBody response; - - - response = getMethodRegistry().createChannelOpenOkBody(); - - - writeFrame(response.generateFrame(channelId)); - } - } - - void assertState(final ConnectionState requiredState) - { - if(_state != requiredState) - { - closeConnection(AMQConstant.COMMAND_INVALID, "Command Invalid", 0); - - } - } - - @Override - public void receiveConnectionOpen(AMQShortString virtualHostName, - AMQShortString capabilities, - boolean insist) - { - if(_logger.isDebugEnabled()) - { - _logger.debug("RECV ConnectionOpen[" +" virtualHost: " + virtualHostName + " capabilities: " + capabilities + " insist: " + insist + " ]"); - } - - String virtualHostStr; - if ((virtualHostName != null) && virtualHostName.charAt(0) == '/') - { - virtualHostStr = virtualHostName.toString().substring(1); - } - else - { - virtualHostStr = virtualHostName == null ? null : virtualHostName.toString(); - } - - VirtualHostImpl<?,?,?> virtualHost = ((AmqpPort)getPort()).getVirtualHost(virtualHostStr); - - if (virtualHost == null) - { - closeConnection(AMQConstant.NOT_FOUND, - "Unknown virtual host: '" + virtualHostName + "'",0); - - } - else - { - // Check virtualhost access - if (virtualHost.getState() != State.ACTIVE) - { - String redirectHost = virtualHost.getRedirectHost(getPort()); - if(redirectHost != null) - { - closeConnection(0, new AMQFrame(0,new ConnectionRedirectBody(getProtocolVersion(),AMQShortString.valueOf(redirectHost), null))); - } - else - { - closeConnection(AMQConstant.CONNECTION_FORCED, - "Virtual host '" + virtualHost.getName() + "' is not active", 0); - } - - } - else - { - setVirtualHost(virtualHost); - try - { - - if(virtualHost.authoriseCreateConnection(this)) - { - if (getContextKey() == null) - { - setContextKey(new AMQShortString(Long.toString(System.currentTimeMillis()))); - } - - MethodRegistry methodRegistry = getMethodRegistry(); - AMQMethodBody responseBody = methodRegistry.createConnectionOpenOkBody(virtualHostName); - - writeFrame(responseBody.generateFrame(0)); - _state = ConnectionState.OPEN; - - } - else - { - closeConnection(AMQConstant.ACCESS_REFUSED, "Connection refused",0); - } - } - catch (AccessControlException e) - { - closeConnection(AMQConstant.ACCESS_REFUSED, e.getMessage(), 0); - } - } - } - } - - @Override - public void receiveConnectionClose(final int replyCode, - final AMQShortString replyText, - final int classId, - final int methodId) - { - if(_logger.isDebugEnabled()) - { - _logger.debug("RECV ConnectionClose[" +" replyCode: " + replyCode + " replyText: " + replyText + " classId: " + classId + " methodId: " + methodId + " ]"); - } - - if (_logger.isInfoEnabled()) - { - _logger.info("ConnectionClose received with reply code/reply text " + replyCode + "/" + - replyText + " for " + this); - } - try - { - closeConnectionInternal(false); - - MethodRegistry methodRegistry = getMethodRegistry(); - ConnectionCloseOkBody responseBody = methodRegistry.createConnectionCloseOkBody(); - writeFrame(responseBody.generateFrame(0)); - } - catch (Exception e) - { - _logger.error("Error closing connection for " + getRemoteAddressString(), e); - } - finally - { - closeNetworkConnection(); - } - } - - @Override - public void receiveConnectionCloseOk() - { - if(_logger.isDebugEnabled()) - { - _logger.debug("RECV ConnectionCloseOk"); - } - - _logger.info("Received Connection-close-ok"); - - try - { - closeConnectionInternal(false); - } - catch (Exception e) - { - _logger.error("Error closing connection: " + getRemoteAddressString(), e); - } - finally - { - closeNetworkConnection(); - } - - } - - @Override - public void receiveConnectionSecureOk(final byte[] response) - { - if(_logger.isDebugEnabled()) - { - _logger.debug("RECV ConnectionSecureOk[ response: ******** ] "); - } - - assertState(ConnectionState.AWAIT_SECURE_OK); - - Broker<?> broker = getBroker(); - - SubjectCreator subjectCreator = getSubjectCreator(); - - SaslServer ss = getSaslServer(); - if (ss == null) - { - closeConnection(AMQConstant.INTERNAL_ERROR, "No SASL context set up in connection",0 ); - } - MethodRegistry methodRegistry = getMethodRegistry(); - SubjectAuthenticationResult authResult = subjectCreator.authenticate(ss, response); - switch (authResult.getStatus()) - { - case ERROR: - Exception cause = authResult.getCause(); - - _logger.info("Authentication failed:" + (cause == null ? "" : cause.getMessage())); - - closeConnection(AMQConstant.NOT_ALLOWED, "Authentication failed",0); - - disposeSaslServer(); - break; - case SUCCESS: - if (_logger.isInfoEnabled()) - { - _logger.info("Connected as: " + authResult.getSubject()); - } - - int frameMax = broker.getContextValue(Integer.class, Broker.BROKER_FRAME_SIZE); - - if (frameMax <= 0) - { - frameMax = Integer.MAX_VALUE; - } - - ConnectionTuneBody tuneBody = - methodRegistry.createConnectionTuneBody(broker.getConnection_sessionCountLimit(), - frameMax, - broker.getConnection_heartBeatDelay()); - writeFrame(tuneBody.generateFrame(0)); - _state = ConnectionState.AWAIT_TUNE_OK; - setAuthorizedSubject(authResult.getSubject()); - disposeSaslServer(); - break; - case CONTINUE: - - ConnectionSecureBody - secureBody = methodRegistry.createConnectionSecureBody(authResult.getChallenge()); - writeFrame(secureBody.generateFrame(0)); - } - } - - - private void disposeSaslServer() - { - SaslServer ss = getSaslServer(); - if (ss != null) - { - setSaslServer(null); - try - { - ss.dispose(); - } - catch (SaslException e) - { - _logger.error("Error disposing of Sasl server: " + e); - } - } - } - - @Override - public void receiveConnectionStartOk(final FieldTable clientProperties, - final AMQShortString mechanism, - final byte[] response, - final AMQShortString locale) - { - if (_logger.isDebugEnabled()) - { - _logger.debug("RECV ConnectionStartOk[" - + " clientProperties: " - + clientProperties - + " mechanism: " - + mechanism - + " response: ********" - + " locale: " - + locale - + " ]"); - } - - assertState(ConnectionState.AWAIT_START_OK); - - Broker<?> broker = getBroker(); - - _logger.info("SASL Mechanism selected: " + mechanism); - _logger.info("Locale selected: " + locale); - - SubjectCreator subjectCreator = getSubjectCreator(); - SaslServer ss; - try - { - ss = subjectCreator.createSaslServer(String.valueOf(mechanism), - getLocalFQDN(), - getPeerPrincipal()); - - if (ss == null) - { - closeConnection(AMQConstant.RESOURCE_ERROR, "Unable to create SASL Server:" + mechanism, 0); - - } - else - { - //save clientProperties - setClientProperties(clientProperties); - - setSaslServer(ss); - - final SubjectAuthenticationResult authResult = subjectCreator.authenticate(ss, response); - - MethodRegistry methodRegistry = getMethodRegistry(); - - switch (authResult.getStatus()) - { - case ERROR: - Exception cause = authResult.getCause(); - - _logger.info("Authentication failed:" + (cause == null ? "" : cause.getMessage())); - - closeConnection(AMQConstant.NOT_ALLOWED, "Authentication failed", 0); - - disposeSaslServer(); - break; - - case SUCCESS: - if (_logger.isInfoEnabled()) - { - _logger.info("Connected as: " + authResult.getSubject()); - } - setAuthorizedSubject(authResult.getSubject()); - - int frameMax = broker.getContextValue(Integer.class, Broker.BROKER_FRAME_SIZE); - - if (frameMax <= 0) - { - frameMax = Integer.MAX_VALUE; - } - - ConnectionTuneBody - tuneBody = - methodRegistry.createConnectionTuneBody(broker.getConnection_sessionCountLimit(), - frameMax, - broker.getConnection_heartBeatDelay()); - writeFrame(tuneBody.generateFrame(0)); - _state = ConnectionState.AWAIT_TUNE_OK; - break; - case CONTINUE: - ConnectionSecureBody - secureBody = methodRegistry.createConnectionSecureBody(authResult.getChallenge()); - writeFrame(secureBody.generateFrame(0)); - - _state = ConnectionState.AWAIT_SECURE_OK; - } - } - } - catch (SaslException e) - { - disposeSaslServer(); - closeConnection(AMQConstant.INTERNAL_ERROR, "SASL error: " + e, 0); - } - } - - @Override - public void receiveConnectionTuneOk(final int channelMax, final long frameMax, final int heartbeat) - { - if(_logger.isDebugEnabled()) - { - _logger.debug("RECV ConnectionTuneOk[" +" channelMax: " + channelMax + " frameMax: " + frameMax + " heartbeat: " + heartbeat + " ]"); - } - - assertState(ConnectionState.AWAIT_TUNE_OK); - - initHeartbeats(heartbeat); - - int brokerFrameMax = getBroker().getContextValue(Integer.class, Broker.BROKER_FRAME_SIZE); - if (brokerFrameMax <= 0) - { - brokerFrameMax = Integer.MAX_VALUE; - } - - if (frameMax > (long) brokerFrameMax) - { - closeConnection(AMQConstant.SYNTAX_ERROR, - "Attempt to set max frame size to " + frameMax - + " greater than the broker will allow: " - + brokerFrameMax, 0); - } - else if (frameMax > 0 && frameMax < AMQConstant.FRAME_MIN_SIZE.getCode()) - { - closeConnection(AMQConstant.SYNTAX_ERROR, - "Attempt to set max frame size to " + frameMax - + " which is smaller than the specification defined minimum: " - + AMQConstant.FRAME_MIN_SIZE.getCode(), 0); - } - else - { - int calculatedFrameMax = frameMax == 0 ? brokerFrameMax : (int) frameMax; - setMaxFrameSize(calculatedFrameMax); - - //0 means no implied limit, except that forced by protocol limitations (0xFFFF) - setMaximumNumberOfChannels( ((channelMax == 0l) || (channelMax > 0xFFFFL)) - ? 0xFFFFL - : channelMax); - - } - _state = ConnectionState.AWAIT_OPEN; - - } - - public int getBinaryDataLimit() - { - return _binaryDataLimit; - } - - public long getMaxMessageSize() - { - return _maxMessageSize; - } - - public final class WriteDeliverMethod - implements ClientDeliveryMethod - { - private final int _channelId; - - public WriteDeliverMethod(int channelId) - { - _channelId = channelId; - } - - @Override - public long deliverToClient(final ConsumerImpl sub, final ServerMessage message, - final InstanceProperties props, final long deliveryTag) - { - long size = _protocolOutputConverter.writeDeliver(message, - props, - _channelId, - deliveryTag, - new AMQShortString(sub.getName())); - registerMessageDelivered(size); - return size; - } - - } - - public Object getReference() - { - return _reference; - } - - @Override - public long getLastReadTime() - { - return _lastReceivedTime; - } - - @Override - public long getLastWriteTime() - { - return _lastWriteTime; - } - - public boolean isCloseWhenNoRoute() - { - return _closeWhenNoRoute; - } - - public boolean isCompressionSupported() - { - return _compressionSupported && _broker.isMessageCompressionEnabled(); - } - - public int getMessageCompressionThreshold() - { - return _messageCompressionThreshold; - } - - public Broker<?> getBroker() - { - return _broker; - } - - public SubjectCreator getSubjectCreator() - { - return _port.getAuthenticationProvider().getSubjectCreator(getTransport().isSecure()); - } - - public EventLogger getEventLogger() - { - if(_virtualHost != null) - { - return _virtualHost.getEventLogger(); - } - else - { - return _broker.getEventLogger(); - } - } - - @Override - public ServerChannelMethodProcessor getChannelMethodProcessor(final int channelId) - { - assertState(ConnectionState.OPEN); - - ServerChannelMethodProcessor channelMethodProcessor = getChannel(channelId); - if(channelMethodProcessor == null) - { - channelMethodProcessor = (ServerChannelMethodProcessor) Proxy.newProxyInstance(ServerMethodDispatcher.class.getClassLoader(), - new Class[] { ServerChannelMethodProcessor.class }, new InvocationHandler() - { - @Override - public Object invoke(final Object proxy, final Method method, final Object[] args) - throws Throwable - { - if(method.getName().startsWith("receive")) - { - closeConnection(AMQConstant.CHANNEL_ERROR, - "Unknown channel id: " + channelId, - channelId); - return null; - } - else if(method.getName().equals("ignoreAllButCloseOk")) - { - return false; - } - return null; - } - }); - } - return channelMethodProcessor; - } - - @Override - public void receiveHeartbeat() - { - if(_logger.isDebugEnabled()) - { - _logger.debug("RECV Heartbeat"); - } - - // No op - } - - @Override - public void receiveProtocolHeader(final ProtocolInitiation protocolInitiation) - { - - if(_logger.isDebugEnabled()) - { - _logger.debug("RECV ProtocolHeader [" + protocolInitiation + " ]"); - } - - protocolInitiationReceived(protocolInitiation); - } - - @Override - public void setCurrentMethod(final int classId, final int methodId) - { - _currentClassId = classId; - _currentMethodId = methodId; - } - - @Override - public boolean ignoreAllButCloseOk() - { - return _closing.get(); - } - - @Override - public void processPending() - { - for (AMQSessionModel session : getSessionModels()) - { - session.processPending(); - } - - while(_asyncTaskList.peek() != null) - { - Action<? super AMQProtocolEngine> asyncAction = _asyncTaskList.poll(); - asyncAction.performAction(this); - } - - } - - @Override - public boolean hasWork() - { - return _stateChanged.get(); - } - - @Override - public void notifyWork() - { - _stateChanged.set(true); - - final Action<ServerProtocolEngine> listener = _workListener.get(); - if(listener != null) - { - - listener.performAction(this); - } - } - - @Override - public void clearWork() - { - _stateChanged.set(false); - } - - @Override - public void setWorkListener(final Action<ServerProtocolEngine> listener) - { - _workListener.set(listener); - } -} diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/BrokerDecoder.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/BrokerDecoder.java deleted file mode 100644 index 81e0b1b736..0000000000 --- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/BrokerDecoder.java +++ /dev/null @@ -1,118 +0,0 @@ -/* - * - * 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 java.io.IOException; -import java.security.PrivilegedActionException; -import java.security.PrivilegedExceptionAction; - -import javax.security.auth.Subject; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import org.apache.qpid.codec.MarkableDataInput; -import org.apache.qpid.codec.ServerDecoder; -import org.apache.qpid.framing.AMQFrameDecodingException; -import org.apache.qpid.server.util.ServerScopedRuntimeException; - -public class BrokerDecoder extends ServerDecoder -{ - private static final Logger _logger = LoggerFactory.getLogger(BrokerDecoder.class); - private final AMQProtocolEngine _connection; - /** - * Creates a new AMQP decoder. - * - * @param connection - */ - public BrokerDecoder(final AMQProtocolEngine connection) - { - super(connection); - _connection = connection; - } - - @Override - protected void processFrame(final int channelId, final byte type, final long bodySize, final MarkableDataInput in) - throws AMQFrameDecodingException, IOException - { - long startTime = 0; - if (_logger.isDebugEnabled()) - { - startTime = System.currentTimeMillis(); - } - Subject subject; - AMQChannel channel = _connection.getChannel(channelId); - if(channel == null) - { - subject = _connection.getSubject(); - } - else - { - _connection.channelRequiresSync(channel); - - subject = channel.getSubject(); - } - try - { - Subject.doAs(subject, new PrivilegedExceptionAction<Object>() - { - @Override - public Void run() throws IOException, AMQFrameDecodingException - { - doProcessFrame(channelId, type, bodySize, in); - return null; - } - }); - if(_logger.isDebugEnabled()) - { - _logger.debug("Frame handled in " + (System.currentTimeMillis() - startTime) + " ms."); - } - - } - catch (PrivilegedActionException e) - { - Throwable cause = e.getCause(); - if(cause instanceof IOException) - { - throw (IOException) cause; - } - else if(cause instanceof AMQFrameDecodingException) - { - throw (AMQFrameDecodingException) cause; - } - else if(cause instanceof RuntimeException) - { - throw (RuntimeException) cause; - } - else throw new ServerScopedRuntimeException(cause); - } - - } - - - private void doProcessFrame(final int channelId, final byte type, final long bodySize, final MarkableDataInput in) - throws AMQFrameDecodingException, IOException - { - super.processFrame(channelId, type, bodySize, in); - - } - -} diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ClientDeliveryMethod.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ClientDeliveryMethod.java deleted file mode 100644 index c7871e8b9a..0000000000 --- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ClientDeliveryMethod.java +++ /dev/null @@ -1,31 +0,0 @@ -/* -* -* 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 org.apache.qpid.server.consumer.ConsumerImpl; -import org.apache.qpid.server.message.InstanceProperties; -import org.apache.qpid.server.message.ServerMessage; - -public interface ClientDeliveryMethod -{ - long deliverToClient(final ConsumerImpl sub, final ServerMessage message, final InstanceProperties props, - final long deliveryTag); -} diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ConsumerTagInUseException.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ConsumerTagInUseException.java deleted file mode 100644 index 25c1462060..0000000000 --- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ConsumerTagInUseException.java +++ /dev/null @@ -1,29 +0,0 @@ -/* - * - * 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; - -public class ConsumerTagInUseException extends Exception -{ - public ConsumerTagInUseException(final String message) - { - super(message); - } -} diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ConsumerTarget_0_8.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ConsumerTarget_0_8.java deleted file mode 100644 index d46ed719b2..0000000000 --- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ConsumerTarget_0_8.java +++ /dev/null @@ -1,565 +0,0 @@ -/* - * - * 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 java.util.List; -import java.util.concurrent.CopyOnWriteArrayList; -import java.util.concurrent.atomic.AtomicBoolean; -import java.util.concurrent.atomic.AtomicLong; - -import org.apache.qpid.AMQException; -import org.apache.qpid.common.AMQPFilterTypes; -import org.apache.qpid.framing.AMQShortString; -import org.apache.qpid.framing.FieldTable; -import org.apache.qpid.server.consumer.AbstractConsumerTarget; -import org.apache.qpid.server.consumer.ConsumerImpl; -import org.apache.qpid.server.flow.FlowCreditManager; -import org.apache.qpid.server.message.InstanceProperties; -import org.apache.qpid.server.message.MessageInstance; -import org.apache.qpid.server.message.MessageReference; -import org.apache.qpid.server.message.ServerMessage; -import org.apache.qpid.server.protocol.AMQSessionModel; -import org.apache.qpid.server.queue.QueueEntry; -import org.apache.qpid.server.txn.AutoCommitTransaction; -import org.apache.qpid.server.txn.ServerTransaction; -import org.apache.qpid.server.util.StateChangeListener; - -/** - * Encapsulation of a subscription to a queue. - * <p> - * Ties together the protocol session of a subscriber, the consumer tag - * that was given out by the broker and the channel id. - */ -public abstract class ConsumerTarget_0_8 extends AbstractConsumerTarget implements FlowCreditManager.FlowCreditManagerListener -{ - - private final StateChangeListener<MessageInstance, MessageInstance.State> _entryReleaseListener = - new StateChangeListener<MessageInstance, MessageInstance.State>() - { - @Override - public void stateChanged(final MessageInstance entry, - final MessageInstance.State oldSate, - final MessageInstance.State newState) - { - if (oldSate == QueueEntry.State.ACQUIRED && newState != QueueEntry.State.ACQUIRED) - { - restoreCredit(entry.getMessage()); - } - entry.removeStateChangeListener(this); - } - }; - - private final ClientDeliveryMethod _deliveryMethod; - private final RecordDeliveryMethod _recordMethod; - - private final AtomicLong _unacknowledgedCount = new AtomicLong(0); - private final AtomicLong _unacknowledgedBytes = new AtomicLong(0); - private final List<ConsumerImpl> _consumers = new CopyOnWriteArrayList<>(); - private final AtomicBoolean _needToClose = new AtomicBoolean(); - - - public static ConsumerTarget_0_8 createBrowserTarget(AMQChannel channel, - AMQShortString consumerTag, FieldTable filters, - FlowCreditManager creditManager) - { - return new BrowserConsumer(channel, consumerTag, filters, creditManager, channel.getClientDeliveryMethod(), channel.getRecordDeliveryMethod()); - } - - public static ConsumerTarget_0_8 createGetNoAckTarget(final AMQChannel channel, - final AMQShortString consumerTag, - final FieldTable filters, - final FlowCreditManager creditManager, - final ClientDeliveryMethod deliveryMethod, - final RecordDeliveryMethod recordMethod) - { - return new GetNoAckConsumer(channel, consumerTag, filters, creditManager, deliveryMethod, recordMethod); - } - - public List<ConsumerImpl> getConsumers() - { - return _consumers; - } - - - static final class BrowserConsumer extends ConsumerTarget_0_8 - { - public BrowserConsumer(AMQChannel channel, - AMQShortString consumerTag, FieldTable filters, - FlowCreditManager creditManager, - ClientDeliveryMethod deliveryMethod, - RecordDeliveryMethod recordMethod) - { - super(channel, consumerTag, - filters, creditManager, deliveryMethod, recordMethod); - } - - /** - * This method can be called by each of the publisher threads. As a result all changes to the channel object must be - * thread safe. - * - * - * - * @param consumer - * @param entry - * @param batch - * @throws org.apache.qpid.AMQException - */ - @Override - public void doSend(final ConsumerImpl consumer, MessageInstance entry, boolean batch) - { - // We don't decrement the reference here as we don't want to consume the message - // but we do want to send it to the client. - - synchronized (getChannel()) - { - long deliveryTag = getChannel().getNextDeliveryTag(); - sendToClient(consumer, entry.getMessage(), entry.getInstanceProperties(), deliveryTag); - } - - } - - } - - public static ConsumerTarget_0_8 createNoAckTarget(AMQChannel channel, - AMQShortString consumerTag, FieldTable filters, - FlowCreditManager creditManager) - { - return new NoAckConsumer(channel, consumerTag, filters, creditManager, channel.getClientDeliveryMethod(), channel.getRecordDeliveryMethod()); - } - - public static ConsumerTarget_0_8 createNoAckTarget(AMQChannel channel, - AMQShortString consumerTag, FieldTable filters, - FlowCreditManager creditManager, - ClientDeliveryMethod deliveryMethod, - RecordDeliveryMethod recordMethod) throws AMQException - { - return new NoAckConsumer(channel, consumerTag, filters, creditManager, deliveryMethod, recordMethod); - } - - public static class NoAckConsumer extends ConsumerTarget_0_8 - { - private final AutoCommitTransaction _txn; - - public NoAckConsumer(AMQChannel channel, - AMQShortString consumerTag, FieldTable filters, - FlowCreditManager creditManager, - ClientDeliveryMethod deliveryMethod, - RecordDeliveryMethod recordMethod) - { - super(channel, consumerTag, filters, creditManager, deliveryMethod, recordMethod); - - _txn = new AutoCommitTransaction(channel.getVirtualHost().getMessageStore()); - } - - /** - * This method can be called by each of the publisher threads. As a result all changes to the channel object must be - * thread safe. - * - * @param consumer - * @param entry The message to send - * @param batch - */ - @Override - public void doSend(final ConsumerImpl consumer, MessageInstance entry, boolean batch) - { - // if we do not need to wait for client acknowledgements - // we can decrement the reference count immediately. - - // By doing this _before_ the send we ensure that it - // doesn't get sent if it can't be dequeued, preventing - // duplicate delivery on recovery. - - // The send may of course still fail, in which case, as - // the message is unacked, it will be lost. - _txn.dequeue(entry.getEnqueueRecord(), NOOP); - - ServerMessage message = entry.getMessage(); - MessageReference ref = message.newReference(); - InstanceProperties props = entry.getInstanceProperties(); - entry.delete(); - long size; - synchronized (getChannel()) - { - getChannel().getConnection().setDeferFlush(batch); - long deliveryTag = getChannel().getNextDeliveryTag(); - - size = sendToClient(consumer, message, props, deliveryTag); - - } - ref.release(); - - } - - private static final ServerTransaction.Action NOOP = - new ServerTransaction.Action() - { - @Override - public void postCommit() - { - } - - @Override - public void onRollback() - { - } - }; - } - - /** - * NoAck Subscription for use with BasicGet method. - */ - public static final class GetNoAckConsumer extends NoAckConsumer - { - public GetNoAckConsumer(AMQChannel channel, - AMQShortString consumerTag, FieldTable filters, - FlowCreditManager creditManager, - ClientDeliveryMethod deliveryMethod, - RecordDeliveryMethod recordMethod) - { - super(channel, consumerTag, filters, creditManager, deliveryMethod, recordMethod); - } - - } - - - public static ConsumerTarget_0_8 createAckTarget(AMQChannel channel, - AMQShortString consumerTag, FieldTable filters, - FlowCreditManager creditManager) - { - return new AckConsumer(channel,consumerTag,filters,creditManager, channel.getClientDeliveryMethod(), channel.getRecordDeliveryMethod()); - } - - - public static ConsumerTarget_0_8 createAckTarget(AMQChannel channel, - AMQShortString consumerTag, FieldTable filters, - FlowCreditManager creditManager, - ClientDeliveryMethod deliveryMethod, - RecordDeliveryMethod recordMethod) - { - return new AckConsumer(channel,consumerTag,filters,creditManager, deliveryMethod, recordMethod); - } - - static final class AckConsumer extends ConsumerTarget_0_8 - { - public AckConsumer(AMQChannel channel, - AMQShortString consumerTag, FieldTable filters, - FlowCreditManager creditManager, - ClientDeliveryMethod deliveryMethod, - RecordDeliveryMethod recordMethod) - { - super(channel, consumerTag, filters, creditManager, deliveryMethod, recordMethod); - } - - /** - * This method can be called by each of the publisher threads. As a result all changes to the channel object must be - * thread safe. - * - * @param consumer - * @param entry The message to send - * @param batch - */ - @Override - public void doSend(final ConsumerImpl consumer, MessageInstance entry, boolean batch) - { - - // put queue entry on a list and then notify the connection to read list. - - synchronized (getChannel()) - { - getChannel().getConnection().setDeferFlush(batch); - long deliveryTag = getChannel().getNextDeliveryTag(); - - addUnacknowledgedMessage(entry); - recordMessageDelivery(consumer, entry, deliveryTag); - entry.addStateChangeListener(getReleasedStateChangeListener()); - long size = sendToClient(consumer, entry.getMessage(), entry.getInstanceProperties(), deliveryTag); - entry.incrementDeliveryCount(); - } - - - } - - - - - - } - - private final AMQChannel _channel; - - private final AMQShortString _consumerTag; - - private final FlowCreditManager _creditManager; - - private final Boolean _autoClose; - - private final AtomicBoolean _deleted = new AtomicBoolean(false); - - - - - public ConsumerTarget_0_8(AMQChannel channel, - AMQShortString consumerTag, - FieldTable arguments, - FlowCreditManager creditManager, - ClientDeliveryMethod deliveryMethod, - RecordDeliveryMethod recordMethod) - { - super(State.ACTIVE); - - _channel = channel; - _consumerTag = consumerTag; - - _creditManager = creditManager; - creditManager.addStateListener(this); - - _deliveryMethod = deliveryMethod; - _recordMethod = recordMethod; - - if (arguments != null) - { - Object autoClose = arguments.get(AMQPFilterTypes.AUTO_CLOSE.getValue()); - if (autoClose != null) - { - _autoClose = (Boolean) autoClose; - } - else - { - _autoClose = false; - } - } - else - { - _autoClose = false; - } - } - - @Override - public void consumerRemoved(final ConsumerImpl sub) - { - _consumers.remove(sub); - if(_consumers.isEmpty()) - { - close(); - } - } - - @Override - public void consumerAdded(final ConsumerImpl sub) - { - _consumers.add( sub ); - } - - public AMQSessionModel getSessionModel() - { - return _channel; - } - - public String toString() - { - String subscriber = "[channel=" + _channel + - ", consumerTag=" + _consumerTag + - ", session=" + getConnection().getRemoteAddressString(); - - return subscriber + "]"; - } - - @Override - public boolean doIsSuspended() - { - return getState()!=State.ACTIVE || _channel.isSuspended() || _deleted.get() || _channel.getConnectionModel().isStopped(); - } - - /** - * Callback indicating that a queue has been deleted. - * - */ - public void queueDeleted() - { - _deleted.set(true); - } - - public boolean isAutoClose() - { - return _autoClose; - } - - public FlowCreditManager getCreditManager() - { - return _creditManager; - } - - - public boolean close() - { - boolean closed = false; - State state = getState(); - - getSendLock(); - - try - { - while(!closed && state != State.CLOSED) - { - closed = updateState(state, State.CLOSED); - if(!closed) - { - state = getState(); - } - } - _creditManager.removeListener(this); - return closed; - } - finally - { - releaseSendLock(); - } - } - - - public boolean allocateCredit(ServerMessage msg) - { - return _creditManager.useCreditForMessage(msg.getSize()); - } - - public AMQChannel getChannel() - { - return _channel; - } - - public AMQShortString getConsumerTag() - { - return _consumerTag; - } - - private AMQProtocolEngine getConnection() - { - return _channel.getConnection(); - } - - public void restoreCredit(final ServerMessage message) - { - _creditManager.restoreCredit(1, message.getSize()); - } - - protected final StateChangeListener<MessageInstance, MessageInstance.State> getReleasedStateChangeListener() - { - return _entryReleaseListener; - } - - public void creditStateChanged(boolean hasCredit) - { - - if(hasCredit) - { - if(!updateState(State.SUSPENDED, State.ACTIVE)) - { - // this is a hack to get round the issue of increasing bytes credit - notifyCurrentState(); - } - } - else - { - updateState(State.ACTIVE, State.SUSPENDED); - } - } - - protected long sendToClient(final ConsumerImpl consumer, final ServerMessage message, - final InstanceProperties props, - final long deliveryTag) - { - return _deliveryMethod.deliverToClient(consumer, message, props, deliveryTag); - - } - - - protected void recordMessageDelivery(final ConsumerImpl consumer, - final MessageInstance entry, - final long deliveryTag) - { - _recordMethod.recordMessageDelivery(consumer, entry, deliveryTag); - } - - - public void confirmAutoClose() - { - ProtocolOutputConverter converter = getChannel().getConnection().getProtocolOutputConverter(); - converter.confirmConsumerAutoClose(getChannel().getChannelId(), getConsumerTag()); - } - - public void queueEmpty() - { - if (isAutoClose()) - { - _needToClose.set(true); - getChannel().getConnection().notifyWork(); - } - } - - @Override - protected void processClosed() - { - if (_needToClose.get() && getState() != State.CLOSED) - { - close(); - confirmAutoClose(); - } - } - - public void flushBatched() - { - _channel.getConnection().setDeferFlush(false); - } - - protected void addUnacknowledgedMessage(MessageInstance entry) - { - final long size = entry.getMessage().getSize(); - _unacknowledgedBytes.addAndGet(size); - _unacknowledgedCount.incrementAndGet(); - entry.addStateChangeListener(new StateChangeListener<MessageInstance, MessageInstance.State>() - { - public void stateChanged(MessageInstance entry, MessageInstance.State oldState, MessageInstance.State newState) - { - if(oldState.equals(MessageInstance.State.ACQUIRED) && !newState.equals(MessageInstance.State.ACQUIRED)) - { - _unacknowledgedBytes.addAndGet(-size); - _unacknowledgedCount.decrementAndGet(); - entry.removeStateChangeListener(this); - } - } - }); - } - - @Override - public void acquisitionRemoved(final MessageInstance node) - { - } - - public long getUnacknowledgedBytes() - { - return _unacknowledgedBytes.longValue(); - } - - public long getUnacknowledgedMessages() - { - return _unacknowledgedCount.longValue(); - } -} diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ExtractResendAndRequeue.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ExtractResendAndRequeue.java deleted file mode 100644 index 45ab62ee05..0000000000 --- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ExtractResendAndRequeue.java +++ /dev/null @@ -1,79 +0,0 @@ -/* - * - * 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 java.util.Map; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import org.apache.qpid.server.consumer.ConsumerImpl; -import org.apache.qpid.server.message.MessageInstance; - -public class ExtractResendAndRequeue implements UnacknowledgedMessageMap.Visitor -{ - private static final Logger _log = LoggerFactory.getLogger(ExtractResendAndRequeue.class); - - private final Map<Long, MessageInstance> _msgToRequeue; - private final Map<Long, MessageInstance> _msgToResend; - private final UnacknowledgedMessageMap _unacknowledgedMessageMap; - - public ExtractResendAndRequeue(UnacknowledgedMessageMap unacknowledgedMessageMap, - Map<Long, MessageInstance> msgToRequeue, - Map<Long, MessageInstance> msgToResend) - { - _unacknowledgedMessageMap = unacknowledgedMessageMap; - _msgToRequeue = msgToRequeue; - _msgToResend = msgToResend; - } - - public boolean callback(final long deliveryTag, MessageInstance message) - { - - message.setRedelivered(); - final ConsumerImpl consumer = message.getDeliveredConsumer(); - if (consumer != null) - { - // Consumer exists - if (!consumer.isClosed()) - { - _msgToResend.put(deliveryTag, message); - } - else // consumer has gone - { - _msgToRequeue.put(deliveryTag, message); - } - } - else - { - _log.info("No DeadLetter Queue and requeue not requested so dropping message:" + message); - } - - // false means continue processing - return false; - } - - public void visitComplete() - { - _unacknowledgedMessageMap.clear(); - } - -} diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/IncomingMessage.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/IncomingMessage.java deleted file mode 100644 index 6e622fb59a..0000000000 --- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/IncomingMessage.java +++ /dev/null @@ -1,107 +0,0 @@ -/* - * - * 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 java.util.ArrayList; -import java.util.List; - -import org.apache.qpid.framing.AMQShortString; -import org.apache.qpid.framing.ContentBody; -import org.apache.qpid.framing.ContentHeaderBody; -import org.apache.qpid.framing.MessagePublishInfo; -import org.apache.qpid.server.message.MessageDestination; - -public class IncomingMessage -{ - - private final MessagePublishInfo _messagePublishInfo; - private ContentHeaderBody _contentHeaderBody; - private MessageDestination _messageDestination; - - /** - * Keeps a track of how many bytes we have received in body frames - */ - private long _bodyLengthReceived = 0; - private List<ContentBody> _contentChunks = new ArrayList<ContentBody>(); - - public IncomingMessage(MessagePublishInfo info) - { - _messagePublishInfo = info; - } - - public void setContentHeaderBody(final ContentHeaderBody contentHeaderBody) - { - _contentHeaderBody = contentHeaderBody; - } - - public MessagePublishInfo getMessagePublishInfo() - { - return _messagePublishInfo; - } - - public long addContentBodyFrame(final ContentBody contentChunk) - { - _bodyLengthReceived += contentChunk.getSize(); - _contentChunks.add(contentChunk); - return _bodyLengthReceived; - } - - public boolean allContentReceived() - { - return (_bodyLengthReceived == getContentHeader().getBodySize()); - } - - public AMQShortString getExchangeName() - { - return _messagePublishInfo.getExchange(); - } - - public MessageDestination getDestination() - { - return _messageDestination; - } - - public ContentHeaderBody getContentHeader() - { - return _contentHeaderBody; - } - - public long getSize() - { - return getContentHeader().getBodySize(); - } - - public void setMessageDestination(final MessageDestination e) - { - _messageDestination = e; - } - - public int getBodyCount() - { - return _contentChunks.size(); - } - - public ContentBody getContentChunk(int index) - { - return _contentChunks.get(index); - } - -} diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/MessageConverter_Internal_to_v0_8.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/MessageConverter_Internal_to_v0_8.java deleted file mode 100644 index ef42a1311f..0000000000 --- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/MessageConverter_Internal_to_v0_8.java +++ /dev/null @@ -1,244 +0,0 @@ -/* - * - * 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 java.io.ByteArrayOutputStream; -import java.io.IOException; -import java.io.ObjectOutputStream; -import java.nio.ByteBuffer; -import java.nio.charset.Charset; -import java.util.LinkedHashMap; -import java.util.List; -import java.util.Map; - -import org.apache.qpid.framing.AMQShortString; -import org.apache.qpid.framing.BasicContentHeaderProperties; -import org.apache.qpid.framing.ContentHeaderBody; -import org.apache.qpid.framing.FieldTable; -import org.apache.qpid.framing.MessagePublishInfo; -import org.apache.qpid.server.message.internal.InternalMessage; -import org.apache.qpid.server.plugin.MessageConverter; -import org.apache.qpid.server.plugin.PluggableService; -import org.apache.qpid.server.store.StoredMessage; -import org.apache.qpid.server.util.ConnectionScopedRuntimeException; -import org.apache.qpid.server.virtualhost.VirtualHostImpl; -import org.apache.qpid.transport.codec.BBEncoder; - -@PluggableService -public class MessageConverter_Internal_to_v0_8 implements MessageConverter<InternalMessage, AMQMessage> -{ - private static final int BASIC_CLASS_ID = 60; - private static final Charset UTF_8 = Charset.forName("UTF-8"); - - - public Class<InternalMessage> getInputClass() - { - return InternalMessage.class; - } - - @Override - public Class<AMQMessage> getOutputClass() - { - return AMQMessage.class; - } - - @Override - public AMQMessage convert(InternalMessage serverMsg, VirtualHostImpl vhost) - { - return new AMQMessage(convertToStoredMessage(serverMsg), null); - } - - private StoredMessage<MessageMetaData> convertToStoredMessage(final InternalMessage serverMsg) - { - final byte[] messageContent = convertToBody(serverMsg.getMessageBody()); - final MessageMetaData messageMetaData_0_8 = convertMetaData(serverMsg, - getBodyMimeType(serverMsg.getMessageBody()), - messageContent.length); - - return new StoredMessage<MessageMetaData>() - { - @Override - public MessageMetaData getMetaData() - { - return messageMetaData_0_8; - } - - @Override - public long getMessageNumber() - { - return serverMsg.getMessageNumber(); - } - - @Override - public int getContent(int offsetInMessage, ByteBuffer dst) - { - int size = messageContent.length - offsetInMessage; - if(dst.remaining() < size) - { - size = dst.remaining(); - } - ByteBuffer buf = ByteBuffer.wrap(messageContent, offsetInMessage, size); - dst.put(buf); - return size; - } - - @Override - public ByteBuffer getContent(int offsetInMessage, int size) - { - return ByteBuffer.wrap(messageContent, offsetInMessage, size); - } - - @Override - public void remove() - { - throw new UnsupportedOperationException(); - } - - @Override - public boolean isInMemory() - { - return true; - } - - @Override - public boolean flowToDisk() - { - return false; - } - }; - } - - private MessageMetaData convertMetaData(final InternalMessage serverMsg, final String bodyMimeType, final int size) - { - - MessagePublishInfo publishInfo = new MessagePublishInfo(AMQShortString.EMPTY_STRING, - false, - false, - AMQShortString.valueOf(serverMsg.getInitialRoutingAddress())); - - - final BasicContentHeaderProperties props = new BasicContentHeaderProperties(); - props.setAppId(serverMsg.getMessageHeader().getAppId()); - props.setContentType(bodyMimeType); - props.setCorrelationId(serverMsg.getMessageHeader().getCorrelationId()); - props.setDeliveryMode(serverMsg.isPersistent() ? BasicContentHeaderProperties.PERSISTENT : BasicContentHeaderProperties.NON_PERSISTENT); - props.setExpiration(serverMsg.getExpiration()); - props.setMessageId(serverMsg.getMessageHeader().getMessageId()); - props.setPriority(serverMsg.getMessageHeader().getPriority()); - props.setReplyTo(serverMsg.getMessageHeader().getReplyTo()); - props.setTimestamp(serverMsg.getMessageHeader().getTimestamp()); - props.setUserId(serverMsg.getMessageHeader().getUserId()); - - - Map<String,Object> headerProps = new LinkedHashMap<String, Object>(); - - for(String headerName : serverMsg.getMessageHeader().getHeaderNames()) - { - headerProps.put(headerName, serverMsg.getMessageHeader().getHeader(headerName)); - } - - props.setHeaders(FieldTable.convertToFieldTable(headerProps)); - - final ContentHeaderBody chb = new ContentHeaderBody(props); - chb.setBodySize(size); - return new MessageMetaData(publishInfo, chb, serverMsg.getArrivalTime()); - } - - - @Override - public String getType() - { - return "Internal to v0-8"; - } - - - public static byte[] convertToBody(Object object) - { - if(object instanceof String) - { - return ((String)object).getBytes(UTF_8); - } - else if(object instanceof byte[]) - { - return (byte[]) object; - } - else if(object instanceof Map) - { - BBEncoder encoder = new BBEncoder(1024); - encoder.writeMap((Map)object); - ByteBuffer buf = encoder.segment(); - int remaining = buf.remaining(); - byte[] data = new byte[remaining]; - buf.get(data); - return data; - - } - else if(object instanceof List) - { - BBEncoder encoder = new BBEncoder(1024); - encoder.writeList((List) object); - ByteBuffer buf = encoder.segment(); - int remaining = buf.remaining(); - byte[] data = new byte[remaining]; - buf.get(data); - return data; - } - else - { - ByteArrayOutputStream bytesOut = new ByteArrayOutputStream(); - try - { - ObjectOutputStream os = new ObjectOutputStream(bytesOut); - os.writeObject(object); - return bytesOut.toByteArray(); - } - catch (IOException e) - { - throw new ConnectionScopedRuntimeException(e); - } - } - } - - public static String getBodyMimeType(Object object) - { - if(object instanceof String) - { - return "text/plain"; - } - else if(object instanceof byte[]) - { - return "application/octet-stream"; - } - else if(object instanceof Map) - { - return "amqp/map"; - } - else if(object instanceof List) - { - return "amqp/list"; - } - else - { - return "application/java-object-stream"; - } - } - -} diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/MessageConverter_v0_8_to_Internal.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/MessageConverter_v0_8_to_Internal.java deleted file mode 100644 index 0b98699b4a..0000000000 --- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/MessageConverter_v0_8_to_Internal.java +++ /dev/null @@ -1,358 +0,0 @@ -/* - * - * 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 java.io.EOFException; -import java.net.URISyntaxException; -import java.nio.ByteBuffer; -import java.util.ArrayList; -import java.util.Collection; -import java.util.LinkedHashMap; -import java.util.List; -import java.util.Map; -import java.util.Set; - -import org.apache.qpid.framing.AMQShortString; -import org.apache.qpid.server.message.AMQMessageHeader; -import org.apache.qpid.server.message.internal.InternalMessage; -import org.apache.qpid.server.plugin.MessageConverter; -import org.apache.qpid.server.plugin.PluggableService; -import org.apache.qpid.server.util.ConnectionScopedRuntimeException; -import org.apache.qpid.server.virtualhost.VirtualHostImpl; -import org.apache.qpid.transport.codec.BBDecoder; -import org.apache.qpid.typedmessage.TypedBytesContentReader; -import org.apache.qpid.typedmessage.TypedBytesFormatException; -import org.apache.qpid.url.AMQBindingURL; - -@PluggableService -public class MessageConverter_v0_8_to_Internal implements MessageConverter<AMQMessage, InternalMessage> -{ - @Override - public Class<AMQMessage> getInputClass() - { - return AMQMessage.class; - } - - @Override - public Class<InternalMessage> getOutputClass() - { - return InternalMessage.class; - } - - @Override - public InternalMessage convert(AMQMessage serverMessage, VirtualHostImpl vhost) - { - final String mimeType = serverMessage.getMessageHeader().getMimeType(); - byte[] data = new byte[(int) serverMessage.getSize()]; - serverMessage.getContent(ByteBuffer.wrap(data), 0); - - Object body = convertMessageBody(mimeType, data); - - return InternalMessage.convert(serverMessage.getMessageNumber(), serverMessage.isPersistent(), - new DelegatingMessageHeader(serverMessage.getMessageHeader()), body); - } - - private static class ReplyToComponents - { - private String _exchange; - private String _queue; - private String _routingKey; - - public void setExchange(final String exchange) - { - _exchange = exchange; - } - - public void setQueue(final String queue) - { - _queue = queue; - } - - public void setRoutingKey(final String routingKey) - { - _routingKey = routingKey; - } - - public String getExchange() - { - return _exchange; - } - - public String getQueue() - { - return _queue; - } - - public String getRoutingKey() - { - return _routingKey; - } - - public boolean hasExchange() - { - return _exchange != null; - } - - public boolean hasQueue() - { - return _queue != null; - } - - public boolean hasRoutingKey() - { - return _routingKey != null; - } - } - - private static class DelegatingMessageHeader implements AMQMessageHeader - { - private final AMQMessageHeader _delegate; - - private DelegatingMessageHeader(final AMQMessageHeader delegate) - { - _delegate = delegate; - } - - @Override - public String getCorrelationId() - { - return _delegate.getCorrelationId(); - } - - @Override - public long getExpiration() - { - return _delegate.getExpiration(); - } - - @Override - public String getUserId() - { - return _delegate.getUserId(); - } - - @Override - public String getAppId() - { - return _delegate.getAppId(); - } - - @Override - public String getMessageId() - { - return _delegate.getMessageId(); - } - - @Override - public String getMimeType() - { - return _delegate.getMimeType(); - } - - @Override - public String getEncoding() - { - return _delegate.getEncoding(); - } - - @Override - public byte getPriority() - { - return _delegate.getPriority(); - } - - @Override - public long getTimestamp() - { - return _delegate.getTimestamp(); - } - - @Override - public String getType() - { - return _delegate.getType(); - } - - @Override - public String getReplyTo() - { - String originalReplyTo = _delegate.getReplyTo(); - ReplyToComponents replyTo = convertReplyTo(originalReplyTo); - if(replyTo != null) - { - if(replyTo.hasExchange()) - { - return replyTo.getExchange() + (replyTo.hasRoutingKey() ? "/" + replyTo.getRoutingKey() : ""); - } - else - { - return replyTo.hasQueue() ? replyTo.getQueue() : replyTo.getRoutingKey(); - } - } - else - { - return originalReplyTo; - } - } - - private ReplyToComponents convertReplyTo(final String origReplyToString) - { - try - { - AMQBindingURL burl = new AMQBindingURL(origReplyToString); - ReplyToComponents replyTo = new ReplyToComponents(); - AMQShortString routingKey = burl.getRoutingKey(); - if(routingKey != null) - { - replyTo.setRoutingKey(routingKey.asString()); - } - - AMQShortString exchangeName = burl.getExchangeName(); - if(exchangeName != null) - { - replyTo.setExchange(exchangeName.asString()); - } - - AMQShortString queueName = burl.getQueueName(); - if(queueName != null) - { - replyTo.setQueue(queueName.asString()); - } - return replyTo; - } - catch (URISyntaxException e) - { - return null; - } - } - - @Override - public Object getHeader(final String name) - { - return _delegate.getHeader(name); - } - - @Override - public boolean containsHeaders(final Set<String> names) - { - return _delegate.containsHeaders(names); - } - - @Override - public boolean containsHeader(final String name) - { - return _delegate.containsHeader(name); - } - - @Override - public Collection<String> getHeaderNames() - { - return _delegate.getHeaderNames(); - } - } - - - private static Object convertMessageBody(String mimeType, byte[] data) - { - if("text/plain".equals(mimeType) || "text/xml".equals(mimeType)) - { - String text = new String(data); - return text; - } - else if("jms/map-message".equals(mimeType)) - { - TypedBytesContentReader reader = new TypedBytesContentReader(ByteBuffer.wrap(data)); - - LinkedHashMap map = new LinkedHashMap(); - final int entries = reader.readIntImpl(); - for (int i = 0; i < entries; i++) - { - try - { - String propName = reader.readStringImpl(); - Object value = reader.readObject(); - - map.put(propName, value); - } - catch (EOFException e) - { - throw new IllegalArgumentException(e); - } - catch (TypedBytesFormatException e) - { - throw new IllegalArgumentException(e); - } - - } - - return map; - - } - else if("amqp/map".equals(mimeType)) - { - BBDecoder decoder = new BBDecoder(); - decoder.init(ByteBuffer.wrap(data)); - final Map<String,Object> map = decoder.readMap(); - - return map; - - } - else if("amqp/list".equals(mimeType)) - { - BBDecoder decoder = new BBDecoder(); - decoder.init(ByteBuffer.wrap(data)); - return decoder.readList(); - } - else if("jms/stream-message".equals(mimeType)) - { - TypedBytesContentReader reader = new TypedBytesContentReader(ByteBuffer.wrap(data)); - - List list = new ArrayList(); - while (reader.remaining() != 0) - { - try - { - list.add(reader.readObject()); - } - catch (TypedBytesFormatException e) - { - throw new ConnectionScopedRuntimeException(e); - } - catch (EOFException e) - { - throw new ConnectionScopedRuntimeException(e); - } - } - return list; - } - else - { - return data; - - } - } - - @Override - public String getType() - { - return "v0-8 to Internal"; - } -} diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/MessageMetaData.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/MessageMetaData.java deleted file mode 100644 index 6b2902d0fa..0000000000 --- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/MessageMetaData.java +++ /dev/null @@ -1,296 +0,0 @@ -/* - * - * 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 java.io.DataInputStream; -import java.io.DataOutputStream; -import java.io.IOException; -import java.nio.ByteBuffer; -import java.util.Collection; -import java.util.Set; - -import org.apache.qpid.framing.AMQFrameDecodingException; -import org.apache.qpid.framing.AMQProtocolVersionException; -import org.apache.qpid.framing.AMQShortString; -import org.apache.qpid.framing.BasicContentHeaderProperties; -import org.apache.qpid.framing.ContentHeaderBody; -import org.apache.qpid.framing.EncodingUtils; -import org.apache.qpid.framing.FieldTable; -import org.apache.qpid.framing.MessagePublishInfo; -import org.apache.qpid.server.message.AMQMessageHeader; -import org.apache.qpid.server.plugin.MessageMetaDataType; -import org.apache.qpid.server.store.StorableMessageMetaData; -import org.apache.qpid.server.util.ByteBufferOutputStream; -import org.apache.qpid.server.util.ConnectionScopedRuntimeException; -import org.apache.qpid.util.ByteBufferInputStream; - -/** - * Encapsulates a publish body and a content header. In the context of the message store these are treated as a - * single unit. - */ -public class MessageMetaData implements StorableMessageMetaData -{ - private final MessagePublishInfo _messagePublishInfo; - - private final ContentHeaderBody _contentHeaderBody; - - - private long _arrivalTime; - private static final byte MANDATORY_FLAG = 1; - private static final byte IMMEDIATE_FLAG = 2; - public static final MessageMetaDataType.Factory<MessageMetaData> FACTORY = new MetaDataFactory(); - private static final MessageMetaDataType_0_8 TYPE = new MessageMetaDataType_0_8(); - - public MessageMetaData(MessagePublishInfo publishBody, ContentHeaderBody contentHeaderBody) - { - this(publishBody,contentHeaderBody, System.currentTimeMillis()); - } - - public MessageMetaData(MessagePublishInfo publishBody, - ContentHeaderBody contentHeaderBody, - long arrivalTime) - { - _contentHeaderBody = contentHeaderBody; - _messagePublishInfo = publishBody; - _arrivalTime = arrivalTime; - } - - - public ContentHeaderBody getContentHeaderBody() - { - return _contentHeaderBody; - } - - public MessagePublishInfo getMessagePublishInfo() - { - return _messagePublishInfo; - } - - public long getArrivalTime() - { - return _arrivalTime; - } - - public MessageMetaDataType getType() - { - return TYPE; - } - - public int getStorableSize() - { - int size = _contentHeaderBody.getSize(); - size += 4; - size += EncodingUtils.encodedShortStringLength(_messagePublishInfo.getExchange()); - size += EncodingUtils.encodedShortStringLength(_messagePublishInfo.getRoutingKey()); - size += 1; // flags for immediate/mandatory - size += EncodingUtils.encodedLongLength(); - - return size; - } - - - public int writeToBuffer(ByteBuffer dest) - { - int oldPosition = dest.position(); - try - { - - DataOutputStream dataOutputStream = new DataOutputStream(new ByteBufferOutputStream(dest)); - EncodingUtils.writeInteger(dataOutputStream, _contentHeaderBody.getSize()); - _contentHeaderBody.writePayload(dataOutputStream); - EncodingUtils.writeShortStringBytes(dataOutputStream, _messagePublishInfo.getExchange()); - EncodingUtils.writeShortStringBytes(dataOutputStream, _messagePublishInfo.getRoutingKey()); - byte flags = 0; - if(_messagePublishInfo.isMandatory()) - { - flags |= MANDATORY_FLAG; - } - if(_messagePublishInfo.isImmediate()) - { - flags |= IMMEDIATE_FLAG; - } - dest.put(flags); - dest.putLong(_arrivalTime); - - } - catch (IOException e) - { - // This shouldn't happen as we are not actually using anything that can throw an IO Exception - throw new ConnectionScopedRuntimeException(e); - } - - return dest.position()-oldPosition; - } - - public int getContentSize() - { - return (int) _contentHeaderBody.getBodySize(); - } - - public boolean isPersistent() - { - return _contentHeaderBody.getProperties().getDeliveryMode() == BasicContentHeaderProperties.PERSISTENT; - } - - private static class MetaDataFactory implements MessageMetaDataType.Factory - { - - - public MessageMetaData createMetaData(ByteBuffer buf) - { - try - { - ByteBufferInputStream bbis = new ByteBufferInputStream(buf); - DataInputStream dais = new DataInputStream(bbis); - int size = EncodingUtils.readInteger(dais); - ContentHeaderBody chb = ContentHeaderBody.createFromBuffer(dais, size); - final AMQShortString exchange = EncodingUtils.readAMQShortString(dais); - final AMQShortString routingKey = EncodingUtils.readAMQShortString(dais); - - final byte flags = EncodingUtils.readByte(dais); - long arrivalTime = EncodingUtils.readLong(dais); - - MessagePublishInfo publishBody = - new MessagePublishInfo(exchange, - (flags & IMMEDIATE_FLAG) != 0, - (flags & MANDATORY_FLAG) != 0, - routingKey); - - return new MessageMetaData(publishBody, chb, arrivalTime); - } - catch (IOException e) - { - throw new ConnectionScopedRuntimeException(e); - } - catch (AMQProtocolVersionException e) - { - throw new ConnectionScopedRuntimeException(e); - } - catch (AMQFrameDecodingException e) - { - throw new ConnectionScopedRuntimeException(e); - } - - } - } - - public AMQMessageHeader getMessageHeader() - { - return new MessageHeaderAdapter(); - } - - private final class MessageHeaderAdapter implements AMQMessageHeader - { - private BasicContentHeaderProperties getProperties() - { - return getContentHeaderBody().getProperties(); - } - - public String getUserId() - { - return getProperties().getUserIdAsString(); - } - - public String getAppId() - { - return getProperties().getAppIdAsString(); - } - - public String getCorrelationId() - { - return getProperties().getCorrelationIdAsString(); - } - - public long getExpiration() - { - return getProperties().getExpiration(); - } - - public String getMessageId() - { - return getProperties().getMessageIdAsString(); - } - - public String getMimeType() - { - return getProperties().getContentTypeAsString(); - } - - public String getEncoding() - { - return getProperties().getEncodingAsString(); - } - - public byte getPriority() - { - return getProperties().getPriority(); - } - - public long getTimestamp() - { - return getProperties().getTimestamp(); - } - - public String getType() - { - return getProperties().getTypeAsString(); - } - - public String getReplyTo() - { - return getProperties().getReplyToAsString(); - } - - public Object getHeader(String name) - { - FieldTable ft = getProperties().getHeaders(); - return ft.get(name); - } - - public boolean containsHeaders(Set<String> names) - { - FieldTable ft = getProperties().getHeaders(); - for(String name : names) - { - if(!ft.containsKey(name)) - { - return false; - } - } - return true; - } - - @Override - public Collection<String> getHeaderNames() - { - return getProperties().getHeaders().keys(); - } - - public boolean containsHeader(String name) - { - FieldTable ft = getProperties().getHeaders(); - return ft.containsKey(name); - } - - - - } -} diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/MessageMetaDataType_0_8.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/MessageMetaDataType_0_8.java deleted file mode 100644 index 5e263cd3b0..0000000000 --- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/MessageMetaDataType_0_8.java +++ /dev/null @@ -1,70 +0,0 @@ -/* - * - * 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 java.nio.ByteBuffer; - -import org.apache.qpid.server.message.ServerMessage; -import org.apache.qpid.server.plugin.MessageMetaDataType; -import org.apache.qpid.server.plugin.PluggableService; -import org.apache.qpid.server.store.StoredMessage; - -@PluggableService -public class MessageMetaDataType_0_8 implements MessageMetaDataType<MessageMetaData> -{ - - public static final int TYPE = 0; - public static final String V0_8 = "v0_8"; - - @Override - public int ordinal() - { - return TYPE; - } - - @Override - public MessageMetaData createMetaData(ByteBuffer buf) - { - return MessageMetaData.FACTORY.createMetaData(buf); - } - - @Override - public ServerMessage<MessageMetaData> createMessage(StoredMessage<MessageMetaData> msg) - { - return new AMQMessage(msg); - } - - public int hashCode() - { - return ordinal(); - } - - public boolean equals(Object o) - { - return o != null && o.getClass() == getClass(); - } - - @Override - public String getType() - { - return V0_8; - } -} diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/MessageOnlyCreditManager.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/MessageOnlyCreditManager.java deleted file mode 100644 index af54c911dc..0000000000 --- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/MessageOnlyCreditManager.java +++ /dev/null @@ -1,73 +0,0 @@ -/* -* -* 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 java.util.concurrent.atomic.AtomicLong; - -import org.apache.qpid.server.flow.AbstractFlowCreditManager; -import org.apache.qpid.server.flow.FlowCreditManager; - -public class MessageOnlyCreditManager extends AbstractFlowCreditManager implements FlowCreditManager -{ - private final AtomicLong _messageCredit; - - public MessageOnlyCreditManager(final long initialCredit) - { - _messageCredit = new AtomicLong(initialCredit); - } - - public void restoreCredit(long messageCredit, long bytesCredit) - { - _messageCredit.addAndGet(messageCredit); - setSuspended(false); - - } - - public boolean hasCredit() - { - return _messageCredit.get() > 0L; - } - - public boolean useCreditForMessage(long msgSize) - { - if(hasCredit()) - { - if(_messageCredit.addAndGet(-1L) >= 0) - { - setSuspended(false); - return true; - } - else - { - _messageCredit.addAndGet(1L); - setSuspended(true); - return false; - } - } - else - { - setSuspended(true); - return false; - } - - } - -} diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/NoAckCreditManager.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/NoAckCreditManager.java deleted file mode 100644 index 6e5aab2dd5..0000000000 --- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/NoAckCreditManager.java +++ /dev/null @@ -1,56 +0,0 @@ -/* - * 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 org.apache.qpid.server.protocol.ServerProtocolEngine; -import org.apache.qpid.server.flow.AbstractFlowCreditManager; - -public class NoAckCreditManager extends AbstractFlowCreditManager -{ - private final ServerProtocolEngine _serverProtocolEngine; - - public NoAckCreditManager(ServerProtocolEngine serverProtocolEngine) - { - _serverProtocolEngine = serverProtocolEngine; - } - - @Override - public void restoreCredit(final long messageCredit, final long bytesCredit) - { - setSuspended(!hasCredit()); - } - - @Override - public boolean hasCredit() - { - return !_serverProtocolEngine.isTransportBlockedForWriting(); - } - - @Override - public boolean useCreditForMessage(final long msgSize) - { - if (!hasCredit()) - { - setSuspended(true); - return false; - } - return true; - } -} diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/Pre0_10CreditManager.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/Pre0_10CreditManager.java deleted file mode 100644 index a869a707e1..0000000000 --- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/Pre0_10CreditManager.java +++ /dev/null @@ -1,190 +0,0 @@ -/* -* -* 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 org.apache.qpid.server.protocol.ServerProtocolEngine; -import org.apache.qpid.server.flow.AbstractFlowCreditManager; -import org.apache.qpid.server.flow.FlowCreditManager; - -public class Pre0_10CreditManager extends AbstractFlowCreditManager implements FlowCreditManager -{ - - private final ServerProtocolEngine _protocolEngine; - private volatile long _bytesCreditLimit; - private volatile long _messageCreditLimit; - - private volatile long _bytesCredit; - private volatile long _messageCredit; - - public Pre0_10CreditManager(long bytesCreditLimit, - long messageCreditLimit, - ServerProtocolEngine protocolEngine) - { - _protocolEngine = protocolEngine; - _bytesCreditLimit = bytesCreditLimit; - _messageCreditLimit = messageCreditLimit; - _bytesCredit = bytesCreditLimit; - _messageCredit = messageCreditLimit; - } - - - - public synchronized void setCreditLimits(final long bytesCreditLimit, final long messageCreditLimit) - { - long bytesCreditChange = bytesCreditLimit - _bytesCreditLimit; - long messageCreditChange = messageCreditLimit - _messageCreditLimit; - - - - if(bytesCreditChange != 0L) - { - if(bytesCreditLimit == 0L) - { - _bytesCredit = 0; - } - else - { - _bytesCredit += bytesCreditChange; - } - } - - - if(messageCreditChange != 0L) - { - if(messageCreditLimit == 0L) - { - _messageCredit = 0; - } - else - { - _messageCredit += messageCreditChange; - } - } - - - _bytesCreditLimit = bytesCreditLimit; - _messageCreditLimit = messageCreditLimit; - - setSuspended(!hasCredit()); - - } - - - public synchronized void restoreCredit(final long messageCredit, final long bytesCredit) - { - final long messageCreditLimit = _messageCreditLimit; - boolean notifyIncrease = true; - if(messageCreditLimit != 0L) - { - notifyIncrease = (_messageCredit != 0); - long newCredit = _messageCredit + messageCredit; - _messageCredit = newCredit > messageCreditLimit ? messageCreditLimit : newCredit; - } - - - final long bytesCreditLimit = _bytesCreditLimit; - if(bytesCreditLimit != 0L) - { - long newCredit = _bytesCredit + bytesCredit; - _bytesCredit = newCredit > bytesCreditLimit ? bytesCreditLimit : newCredit; - if(notifyIncrease && bytesCredit>0) - { - notifyIncreaseBytesCredit(); - } - } - - - - setSuspended(!hasCredit()); - - } - - public synchronized boolean hasCredit() - { - return (_bytesCreditLimit == 0L || _bytesCredit > 0) - && (_messageCreditLimit == 0L || _messageCredit > 0) - && !_protocolEngine.isTransportBlockedForWriting(); - } - - public synchronized boolean useCreditForMessage(final long msgSize) - { - if (_protocolEngine.isTransportBlockedForWriting()) - { - setSuspended(true); - return false; - } - else if(_messageCreditLimit != 0L) - { - if(_messageCredit != 0L) - { - if(_bytesCreditLimit == 0L) - { - _messageCredit--; - - return true; - } - else - { - if((_bytesCredit >= msgSize) || (_bytesCredit == _bytesCreditLimit)) - { - _messageCredit--; - _bytesCredit -= msgSize; - - return true; - } - else - { - return false; - } - } - } - else - { - setSuspended(true); - return false; - } - } - else - { - if(_bytesCreditLimit == 0L) - { - - return true; - } - else - { - if((_bytesCredit >= msgSize) || (_bytesCredit == _bytesCreditLimit)) - { - _bytesCredit -= msgSize; - - return true; - } - else - { - return false; - } - } - - } - - } -} diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ProtocolEngineCreator_0_8.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ProtocolEngineCreator_0_8.java deleted file mode 100644 index e8cf028069..0000000000 --- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ProtocolEngineCreator_0_8.java +++ /dev/null @@ -1,82 +0,0 @@ -/* - * - * 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 org.apache.qpid.server.protocol.ServerProtocolEngine; -import org.apache.qpid.server.model.Broker; -import org.apache.qpid.server.model.Protocol; -import org.apache.qpid.server.model.Transport; -import org.apache.qpid.server.model.port.AmqpPort; -import org.apache.qpid.server.plugin.PluggableService; -import org.apache.qpid.server.plugin.ProtocolEngineCreator; -import org.apache.qpid.transport.network.NetworkConnection; - -@PluggableService -public class ProtocolEngineCreator_0_8 implements ProtocolEngineCreator -{ - private static final byte[] AMQP_0_8_HEADER = - new byte[] { (byte) 'A', - (byte) 'M', - (byte) 'Q', - (byte) 'P', - (byte) 1, - (byte) 1, - (byte) 8, - (byte) 0 - }; - - - public ProtocolEngineCreator_0_8() - { - } - - public Protocol getVersion() - { - return Protocol.AMQP_0_8; - } - - public byte[] getHeaderIdentifier() - { - return AMQP_0_8_HEADER; - } - - public ServerProtocolEngine newProtocolEngine(Broker<?> broker, - NetworkConnection network, - AmqpPort<?> port, - Transport transport, - long id) - { - return new AMQProtocolEngine(broker, network, id, port, transport); - } - - private static ProtocolEngineCreator INSTANCE = new ProtocolEngineCreator_0_8(); - - public static ProtocolEngineCreator getInstance() - { - return INSTANCE; - } - - @Override - public String getType() - { - return getVersion().toString(); - } -} diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ProtocolEngineCreator_0_9.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ProtocolEngineCreator_0_9.java deleted file mode 100644 index 8817e79aff..0000000000 --- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ProtocolEngineCreator_0_9.java +++ /dev/null @@ -1,82 +0,0 @@ -/* - * - * 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 org.apache.qpid.server.protocol.ServerProtocolEngine; -import org.apache.qpid.server.model.Broker; -import org.apache.qpid.server.model.Protocol; -import org.apache.qpid.server.model.Transport; -import org.apache.qpid.server.model.port.AmqpPort; -import org.apache.qpid.server.plugin.PluggableService; -import org.apache.qpid.server.plugin.ProtocolEngineCreator; -import org.apache.qpid.transport.network.NetworkConnection; - -@PluggableService -public class ProtocolEngineCreator_0_9 implements ProtocolEngineCreator -{ - private static final byte[] AMQP_0_9_HEADER = - new byte[] { (byte) 'A', - (byte) 'M', - (byte) 'Q', - (byte) 'P', - (byte) 1, - (byte) 1, - (byte) 0, - (byte) 9 - }; - - public ProtocolEngineCreator_0_9() - { - } - - public Protocol getVersion() - { - return Protocol.AMQP_0_9; - } - - - public byte[] getHeaderIdentifier() - { - return AMQP_0_9_HEADER; - } - - public ServerProtocolEngine newProtocolEngine(Broker<?> broker, - NetworkConnection network, - AmqpPort<?> port, - Transport transport, - long id) - { - return new AMQProtocolEngine(broker, network, id, port, transport); - } - - private static ProtocolEngineCreator INSTANCE = new ProtocolEngineCreator_0_9(); - - public static ProtocolEngineCreator getInstance() - { - return INSTANCE; - } - - @Override - public String getType() - { - return getVersion().toString(); - } -} diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ProtocolEngineCreator_0_9_1.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ProtocolEngineCreator_0_9_1.java deleted file mode 100644 index af37b17d85..0000000000 --- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ProtocolEngineCreator_0_9_1.java +++ /dev/null @@ -1,84 +0,0 @@ -/* - * - * 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 org.apache.qpid.server.protocol.ServerProtocolEngine; -import org.apache.qpid.server.model.Broker; -import org.apache.qpid.server.model.Protocol; -import org.apache.qpid.server.model.Transport; -import org.apache.qpid.server.model.port.AmqpPort; -import org.apache.qpid.server.plugin.PluggableService; -import org.apache.qpid.server.plugin.ProtocolEngineCreator; -import org.apache.qpid.transport.network.NetworkConnection; - -@PluggableService -public class ProtocolEngineCreator_0_9_1 implements ProtocolEngineCreator -{ - - private static final byte[] AMQP_0_9_1_HEADER = - new byte[] { (byte) 'A', - (byte) 'M', - (byte) 'Q', - (byte) 'P', - (byte) 0, - (byte) 0, - (byte) 9, - (byte) 1 - }; - - public ProtocolEngineCreator_0_9_1() - { - } - - public Protocol getVersion() - { - return Protocol.AMQP_0_9_1; - } - - - public byte[] getHeaderIdentifier() - { - return AMQP_0_9_1_HEADER; - } - - public ServerProtocolEngine newProtocolEngine(Broker<?> broker, - NetworkConnection network, - AmqpPort<?> port, - Transport transport, - long id) - { - return new AMQProtocolEngine(broker, network, id, port, transport); - } - - - private static ProtocolEngineCreator INSTANCE = new ProtocolEngineCreator_0_9_1(); - - public static ProtocolEngineCreator getInstance() - { - return INSTANCE; - } - - @Override - public String getType() - { - return getVersion().toString(); - } -} diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ProtocolOutputConverter.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ProtocolOutputConverter.java deleted file mode 100644 index 69f71f14a9..0000000000 --- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ProtocolOutputConverter.java +++ /dev/null @@ -1,55 +0,0 @@ -/* - * - * 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. - * - */ - -/* - * This file is auto-generated by Qpid Gentools v.0.1 - do not modify. - * Supported AMQP versions: - * 8-0 - */ -package org.apache.qpid.server.protocol.v0_8; - -import org.apache.qpid.framing.AMQDataBlock; -import org.apache.qpid.framing.AMQShortString; -import org.apache.qpid.framing.ContentHeaderBody; -import org.apache.qpid.framing.MessagePublishInfo; -import org.apache.qpid.server.message.InstanceProperties; -import org.apache.qpid.server.message.MessageContentSource; -import org.apache.qpid.server.message.ServerMessage; - -public interface ProtocolOutputConverter -{ - void confirmConsumerAutoClose(int channelId, AMQShortString consumerTag); - - long writeDeliver(final ServerMessage msg, - final InstanceProperties props, int channelId, - long deliveryTag, - AMQShortString consumerTag); - - long writeGetOk(final ServerMessage msg, - final InstanceProperties props, - int channelId, - long deliveryTag, - int queueSize); - - void writeReturn(MessagePublishInfo messagePublishInfo, ContentHeaderBody header, MessageContentSource msgContent, int channelId, int replyCode, AMQShortString replyText); - - void writeFrame(AMQDataBlock block); -} diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ProtocolOutputConverterImpl.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ProtocolOutputConverterImpl.java deleted file mode 100644 index 15ce9262fc..0000000000 --- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ProtocolOutputConverterImpl.java +++ /dev/null @@ -1,543 +0,0 @@ -/* - * - * 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 java.io.DataOutput; -import java.io.IOException; -import java.nio.ByteBuffer; - -import org.apache.qpid.AMQException; -import org.apache.qpid.framing.AMQBody; -import org.apache.qpid.framing.AMQDataBlock; -import org.apache.qpid.framing.AMQFrame; -import org.apache.qpid.framing.AMQMethodBody; -import org.apache.qpid.framing.AMQShortString; -import org.apache.qpid.framing.BasicCancelOkBody; -import org.apache.qpid.framing.BasicContentHeaderProperties; -import org.apache.qpid.framing.BasicGetOkBody; -import org.apache.qpid.framing.BasicReturnBody; -import org.apache.qpid.framing.ContentHeaderBody; -import org.apache.qpid.framing.MessagePublishInfo; -import org.apache.qpid.protocol.AMQVersionAwareProtocolSession; -import org.apache.qpid.server.message.InstanceProperties; -import org.apache.qpid.server.message.MessageContentSource; -import org.apache.qpid.server.message.ServerMessage; -import org.apache.qpid.server.plugin.MessageConverter; -import org.apache.qpid.server.protocol.MessageConverterRegistry; -import org.apache.qpid.transport.ByteBufferSender; -import org.apache.qpid.util.GZIPUtils; - -public class ProtocolOutputConverterImpl implements ProtocolOutputConverter -{ - private static final int BASIC_CLASS_ID = 60; - - private final AMQProtocolEngine _connection; - private static final AMQShortString GZIP_ENCODING = AMQShortString.valueOf(GZIPUtils.GZIP_CONTENT_ENCODING); - - public ProtocolOutputConverterImpl(AMQProtocolEngine connection) - { - _connection = connection; - } - - - public long writeDeliver(final ServerMessage m, - final InstanceProperties props, int channelId, - long deliveryTag, - AMQShortString consumerTag) - { - final AMQMessage msg = convertToAMQMessage(m); - final boolean isRedelivered = Boolean.TRUE.equals(props.getProperty(InstanceProperties.Property.REDELIVERED)); - AMQBody deliverBody = createEncodedDeliverBody(msg, isRedelivered, deliveryTag, consumerTag); - return writeMessageDelivery(msg, channelId, deliverBody); - } - - private AMQMessage convertToAMQMessage(ServerMessage serverMessage) - { - if(serverMessage instanceof AMQMessage) - { - return (AMQMessage) serverMessage; - } - else - { - return getMessageConverter(serverMessage).convert(serverMessage, _connection.getVirtualHost()); - } - } - - private <M extends ServerMessage> MessageConverter<M, AMQMessage> getMessageConverter(M message) - { - Class<M> clazz = (Class<M>) message.getClass(); - return MessageConverterRegistry.getConverter(clazz, AMQMessage.class); - } - - private long writeMessageDelivery(AMQMessage message, int channelId, AMQBody deliverBody) - { - return writeMessageDelivery(message, message.getContentHeaderBody(), channelId, deliverBody); - } - - private long writeMessageDelivery(MessageContentSource message, ContentHeaderBody contentHeaderBody, int channelId, AMQBody deliverBody) - { - - int bodySize = (int) message.getSize(); - boolean msgCompressed = isCompressed(contentHeaderBody); - byte[] modifiedContent; - - // straight through case - boolean compressionSupported = _connection.isCompressionSupported(); - - if(msgCompressed && !compressionSupported && - (modifiedContent = GZIPUtils.uncompressBufferToArray(message.getContent(0,bodySize))) != null) - { - BasicContentHeaderProperties modifiedProps = - new BasicContentHeaderProperties(contentHeaderBody.getProperties()); - modifiedProps.setEncoding((String)null); - - writeMessageDeliveryModified(channelId, deliverBody, modifiedProps, modifiedContent); - - return modifiedContent.length; - } - else if(!msgCompressed - && compressionSupported - && contentHeaderBody.getProperties().getEncoding()==null - && bodySize > _connection.getMessageCompressionThreshold() - && (modifiedContent = GZIPUtils.compressBufferToArray(message.getContent(0, bodySize))) != null) - { - BasicContentHeaderProperties modifiedProps = - new BasicContentHeaderProperties(contentHeaderBody.getProperties()); - modifiedProps.setEncoding(GZIP_ENCODING); - - writeMessageDeliveryModified(channelId, deliverBody, modifiedProps, modifiedContent); - - return modifiedContent.length; - } - else - { - writeMessageDeliveryUnchanged(message, contentHeaderBody, channelId, deliverBody, bodySize); - - return bodySize; - } - } - - private int writeMessageDeliveryModified(final int channelId, - final AMQBody deliverBody, - final BasicContentHeaderProperties modifiedProps, - final byte[] content) - { - final int bodySize; - bodySize = content.length; - ContentHeaderBody modifiedHeaderBody = - new ContentHeaderBody(modifiedProps, bodySize); - final MessageContentSource wrappedSource = new MessageContentSource() - { - @Override - public int getContent(final ByteBuffer buf, final int offset) - { - int size = Math.min(buf.remaining(), content.length - offset); - buf.put(content, offset, size); - return size; - } - - @Override - public ByteBuffer getContent(final int offset, final int size) - { - return ByteBuffer.wrap(content, offset, size); - } - - @Override - public long getSize() - { - return content.length; - } - }; - writeMessageDeliveryUnchanged(wrappedSource, modifiedHeaderBody, channelId, deliverBody, bodySize); - return bodySize; - } - - private void writeMessageDeliveryUnchanged(final MessageContentSource message, - final ContentHeaderBody contentHeaderBody, - final int channelId, final AMQBody deliverBody, final int bodySize) - { - if (bodySize == 0) - { - SmallCompositeAMQBodyBlock compositeBlock = new SmallCompositeAMQBodyBlock(channelId, deliverBody, - contentHeaderBody); - - writeFrame(compositeBlock); - } - else - { - int maxBodySize = (int) _connection.getMaxFrameSize() - AMQFrame.getFrameOverhead(); - - - int capacity = bodySize > maxBodySize ? maxBodySize : bodySize; - - int writtenSize = capacity; - - AMQBody firstContentBody = new MessageContentSourceBody(message, 0, capacity); - - CompositeAMQBodyBlock - compositeBlock = - new CompositeAMQBodyBlock(channelId, deliverBody, contentHeaderBody, firstContentBody); - writeFrame(compositeBlock); - - while (writtenSize < bodySize) - { - capacity = bodySize - writtenSize > maxBodySize ? maxBodySize : bodySize - writtenSize; - MessageContentSourceBody body = new MessageContentSourceBody(message, writtenSize, capacity); - writtenSize += capacity; - - writeFrame(new AMQFrame(channelId, body)); - } - } - } - - private boolean isCompressed(final ContentHeaderBody contentHeaderBody) - { - return GZIP_ENCODING.equals(contentHeaderBody.getProperties().getEncoding()); - } - - private class MessageContentSourceBody implements AMQBody - { - public static final byte TYPE = 3; - private final int _length; - private final MessageContentSource _message; - private final int _offset; - - public MessageContentSourceBody(MessageContentSource message, int offset, int length) - { - _message = message; - _offset = offset; - _length = length; - } - - public byte getFrameType() - { - return TYPE; - } - - public int getSize() - { - return _length; - } - - public void writePayload(DataOutput buffer) throws IOException - { - ByteBuffer buf = _message.getContent(_offset, _length); - - if(buf.hasArray()) - { - buffer.write(buf.array(), buf.arrayOffset()+buf.position(), buf.remaining()); - } - else - { - - byte[] data = new byte[_length]; - - buf.get(data); - - buffer.write(data); - } - } - - @Override - public long writePayload(final ByteBufferSender sender) throws IOException - { - ByteBuffer buf = _message.getContent(_offset, _length); - long size = buf.remaining(); - sender.send(buf.duplicate()); - return size; - } - - public void handle(int channelId, AMQVersionAwareProtocolSession amqProtocolSession) throws AMQException - { - throw new UnsupportedOperationException(); - } - - @Override - public String toString() - { - return "[" + getClass().getSimpleName() + " offset: " + _offset + ", length: " + _length + "]"; - } - - } - - public long writeGetOk(final ServerMessage msg, - final InstanceProperties props, - int channelId, - long deliveryTag, - int queueSize) - { - AMQBody deliver = createEncodedGetOkBody(msg, props, deliveryTag, queueSize); - return writeMessageDelivery(convertToAMQMessage(msg), channelId, deliver); - } - - - private AMQBody createEncodedDeliverBody(AMQMessage message, - boolean isRedelivered, - final long deliveryTag, - final AMQShortString consumerTag) - { - - final AMQShortString exchangeName; - final AMQShortString routingKey; - - final MessagePublishInfo pb = message.getMessagePublishInfo(); - exchangeName = pb.getExchange(); - routingKey = pb.getRoutingKey(); - - final AMQBody returnBlock = new EncodedDeliveryBody(deliveryTag, routingKey, exchangeName, consumerTag, isRedelivered); - return returnBlock; - } - - private class EncodedDeliveryBody implements AMQBody - { - private final long _deliveryTag; - private final AMQShortString _routingKey; - private final AMQShortString _exchangeName; - private final AMQShortString _consumerTag; - private final boolean _isRedelivered; - private AMQBody _underlyingBody; - - private EncodedDeliveryBody(long deliveryTag, AMQShortString routingKey, AMQShortString exchangeName, AMQShortString consumerTag, boolean isRedelivered) - { - _deliveryTag = deliveryTag; - _routingKey = routingKey; - _exchangeName = exchangeName; - _consumerTag = consumerTag; - _isRedelivered = isRedelivered; - } - - public AMQBody createAMQBody() - { - return _connection.getMethodRegistry().createBasicDeliverBody(_consumerTag, - _deliveryTag, - _isRedelivered, - _exchangeName, - _routingKey); - } - - public byte getFrameType() - { - return AMQMethodBody.TYPE; - } - - public int getSize() - { - if(_underlyingBody == null) - { - _underlyingBody = createAMQBody(); - } - return _underlyingBody.getSize(); - } - - public void writePayload(DataOutput buffer) throws IOException - { - if(_underlyingBody == null) - { - _underlyingBody = createAMQBody(); - } - _underlyingBody.writePayload(buffer); - } - - public long writePayload(ByteBufferSender sender) throws IOException - { - if(_underlyingBody == null) - { - _underlyingBody = createAMQBody(); - } - return _underlyingBody.writePayload(sender); - } - - public void handle(final int channelId, final AMQVersionAwareProtocolSession amqProtocolSession) - throws AMQException - { - throw new AMQException("This block should never be dispatched!"); - } - - @Override - public String toString() - { - return "[" + getClass().getSimpleName() + " underlyingBody: " + String.valueOf(_underlyingBody) + "]"; - } - } - - private AMQBody createEncodedGetOkBody(ServerMessage msg, InstanceProperties props, long deliveryTag, int queueSize) - { - final AMQShortString exchangeName; - final AMQShortString routingKey; - - final AMQMessage message = convertToAMQMessage(msg); - final MessagePublishInfo pb = message.getMessagePublishInfo(); - exchangeName = pb.getExchange(); - routingKey = pb.getRoutingKey(); - - final boolean isRedelivered = Boolean.TRUE.equals(props.getProperty(InstanceProperties.Property.REDELIVERED)); - - BasicGetOkBody getOkBody = - _connection.getMethodRegistry().createBasicGetOkBody(deliveryTag, - isRedelivered, - exchangeName, - routingKey, - queueSize); - - return getOkBody; - } - - private AMQBody createEncodedReturnFrame(MessagePublishInfo messagePublishInfo, - int replyCode, - AMQShortString replyText) - { - - BasicReturnBody basicReturnBody = - _connection.getMethodRegistry().createBasicReturnBody(replyCode, - replyText, - messagePublishInfo.getExchange(), - messagePublishInfo.getRoutingKey()); - - - return basicReturnBody; - } - - public void writeReturn(MessagePublishInfo messagePublishInfo, ContentHeaderBody header, MessageContentSource message, int channelId, int replyCode, AMQShortString replyText) - { - - AMQBody returnFrame = createEncodedReturnFrame(messagePublishInfo, replyCode, replyText); - - writeMessageDelivery(message, header, channelId, returnFrame); - } - - - public void writeFrame(AMQDataBlock block) - { - _connection.writeFrame(block); - } - - - public void confirmConsumerAutoClose(int channelId, AMQShortString consumerTag) - { - - BasicCancelOkBody basicCancelOkBody = _connection.getMethodRegistry().createBasicCancelOkBody(consumerTag); - writeFrame(basicCancelOkBody.generateFrame(channelId)); - - } - - - public static final class CompositeAMQBodyBlock extends AMQDataBlock - { - public static final int OVERHEAD = 3 * AMQFrame.getFrameOverhead(); - - private final AMQBody _methodBody; - private final AMQBody _headerBody; - private final AMQBody _contentBody; - private final int _channel; - - - public CompositeAMQBodyBlock(int channel, AMQBody methodBody, AMQBody headerBody, AMQBody contentBody) - { - _channel = channel; - _methodBody = methodBody; - _headerBody = headerBody; - _contentBody = contentBody; - } - - public long getSize() - { - return OVERHEAD + _methodBody.getSize() + _headerBody.getSize() + _contentBody.getSize(); - } - - public void writePayload(DataOutput buffer) throws IOException - { - AMQFrame.writeFrames(buffer, _channel, _methodBody, _headerBody, _contentBody); - } - - @Override - public long writePayload(final ByteBufferSender sender) throws IOException - { - long size = (new AMQFrame(_channel, _methodBody)).writePayload(sender); - - size += (new AMQFrame(_channel, _headerBody)).writePayload(sender); - - size += (new AMQFrame(_channel, _contentBody)).writePayload(sender); - - return size; - } - - @Override - public String toString() - { - StringBuilder builder = new StringBuilder(); - builder.append("[").append(getClass().getSimpleName()) - .append(" methodBody=").append(_methodBody) - .append(", headerBody=").append(_headerBody) - .append(", contentBody=").append(_contentBody) - .append(", channel=").append(_channel).append("]"); - return builder.toString(); - } - - } - - public static final class SmallCompositeAMQBodyBlock extends AMQDataBlock - { - public static final int OVERHEAD = 2 * AMQFrame.getFrameOverhead(); - - private final AMQBody _methodBody; - private final AMQBody _headerBody; - private final int _channel; - - - public SmallCompositeAMQBodyBlock(int channel, AMQBody methodBody, AMQBody headerBody) - { - _channel = channel; - _methodBody = methodBody; - _headerBody = headerBody; - - } - - public long getSize() - { - return OVERHEAD + _methodBody.getSize() + _headerBody.getSize() ; - } - - public void writePayload(DataOutput buffer) throws IOException - { - AMQFrame.writeFrames(buffer, _channel, _methodBody, _headerBody); - } - - @Override - public long writePayload(final ByteBufferSender sender) throws IOException - { - long size = (new AMQFrame(_channel, _methodBody)).writePayload(sender); - size += (new AMQFrame(_channel, _headerBody)).writePayload(sender); - return size; - } - - @Override - public String toString() - { - StringBuilder builder = new StringBuilder(); - builder.append(getClass().getSimpleName()) - .append("methodBody=").append(_methodBody) - .append(", headerBody=").append(_headerBody) - .append(", channel=").append(_channel).append("]"); - return builder.toString(); - } - } - -} diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/RecordDeliveryMethod.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/RecordDeliveryMethod.java deleted file mode 100644 index c13ff17f67..0000000000 --- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/RecordDeliveryMethod.java +++ /dev/null @@ -1,29 +0,0 @@ -/* -* -* 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 org.apache.qpid.server.consumer.ConsumerImpl; -import org.apache.qpid.server.message.MessageInstance; - -public interface RecordDeliveryMethod -{ - void recordMessageDelivery(final ConsumerImpl sub, final MessageInstance entry, final long deliveryTag); -} diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/UnacknowledgedMessageMap.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/UnacknowledgedMessageMap.java deleted file mode 100644 index 198b7fe21b..0000000000 --- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/UnacknowledgedMessageMap.java +++ /dev/null @@ -1,70 +0,0 @@ -/* - * - * 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 java.util.Collection; -import java.util.Map; -import java.util.Set; - -import org.apache.qpid.AMQException; -import org.apache.qpid.server.message.MessageInstance; - - -public interface UnacknowledgedMessageMap -{ - public interface Visitor - { - /** - * @param deliveryTag - *@param message the message being iterated over @return true to stop iteration, false to continue - * @throws AMQException - */ - boolean callback(final long deliveryTag, MessageInstance message); - - void visitComplete(); - } - - void visit(Visitor visitor); - - void add(long deliveryTag, MessageInstance message); - - MessageInstance remove(long deliveryTag); - - Collection<MessageInstance> cancelAllMessages(); - - int size(); - - void clear(); - - MessageInstance get(long deliveryTag); - - /** - * Get the set of delivery tags that are outstanding. - * - * @return a set of delivery tags - */ - Set<Long> getDeliveryTags(); - - Collection<MessageInstance> acknowledge(long deliveryTag, boolean multiple); - void collect(long key, boolean multiple, Map<Long, MessageInstance> msgs); -} - - diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/UnacknowledgedMessageMapImpl.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/UnacknowledgedMessageMapImpl.java deleted file mode 100644 index 2d39daed1c..0000000000 --- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/UnacknowledgedMessageMapImpl.java +++ /dev/null @@ -1,178 +0,0 @@ -/* - * - * 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 java.util.ArrayList; -import java.util.Collection; -import java.util.LinkedHashMap; -import java.util.List; -import java.util.Map; -import java.util.Set; - -import org.apache.qpid.server.message.MessageInstance; - -public class UnacknowledgedMessageMapImpl implements UnacknowledgedMessageMap -{ - private final Object _lock = new Object(); - - private Map<Long, MessageInstance> _map; - - private final int _prefetchLimit; - - public UnacknowledgedMessageMapImpl(int prefetchLimit) - { - _prefetchLimit = prefetchLimit; - _map = new LinkedHashMap<>(prefetchLimit); - } - - public void collect(long deliveryTag, boolean multiple, Map<Long, MessageInstance> msgs) - { - if (multiple) - { - collect(deliveryTag, msgs); - } - else - { - final MessageInstance entry = get(deliveryTag); - if(entry != null) - { - msgs.put(deliveryTag, entry); - } - } - - } - - public void remove(Map<Long,MessageInstance> msgs) - { - synchronized (_lock) - { - for (Long deliveryTag : msgs.keySet()) - { - remove(deliveryTag); - } - } - } - - public MessageInstance remove(long deliveryTag) - { - synchronized (_lock) - { - - MessageInstance message = _map.remove(deliveryTag); - return message; - } - } - - public void visit(Visitor visitor) - { - synchronized (_lock) - { - Set<Map.Entry<Long, MessageInstance>> currentEntries = _map.entrySet(); - for (Map.Entry<Long, MessageInstance> entry : currentEntries) - { - visitor.callback(entry.getKey().longValue(), entry.getValue()); - } - visitor.visitComplete(); - } - } - - public void add(long deliveryTag, MessageInstance message) - { - synchronized (_lock) - { - _map.put(deliveryTag, message); - } - } - - public Collection<MessageInstance> cancelAllMessages() - { - synchronized (_lock) - { - Collection<MessageInstance> currentEntries = _map.values(); - _map = new LinkedHashMap<>(_prefetchLimit); - return currentEntries; - } - } - - public int size() - { - synchronized (_lock) - { - return _map.size(); - } - } - - public void clear() - { - synchronized (_lock) - { - _map.clear(); - } - } - - public MessageInstance get(long key) - { - synchronized (_lock) - { - return _map.get(key); - } - } - - public Set<Long> getDeliveryTags() - { - synchronized (_lock) - { - return _map.keySet(); - } - } - - public Collection<MessageInstance> acknowledge(long deliveryTag, boolean multiple) - { - Map<Long, MessageInstance> ackedMessageMap = new LinkedHashMap<Long,MessageInstance>(); - collect(deliveryTag, multiple, ackedMessageMap); - remove(ackedMessageMap); - List<MessageInstance> acknowledged = new ArrayList<>(); - for(MessageInstance instance : ackedMessageMap.values()) - { - if(instance.lockAcquisition()) - { - acknowledged.add(instance); - } - } - return acknowledged; - } - - private void collect(long key, Map<Long, MessageInstance> msgs) - { - synchronized (_lock) - { - for (Map.Entry<Long, MessageInstance> entry : _map.entrySet()) - { - msgs.put(entry.getKey(),entry.getValue()); - if (entry.getKey() == key) - { - break; - } - } - } - } - -} diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/UnexpectedMethodException.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/UnexpectedMethodException.java deleted file mode 100644 index 432a725c86..0000000000 --- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/UnexpectedMethodException.java +++ /dev/null @@ -1,36 +0,0 @@ -/*
- *
- * 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 org.apache.qpid.AMQException;
-import org.apache.qpid.framing.AMQMethodBody;
-
-public class UnexpectedMethodException extends AMQException
-{
-
- private static final long serialVersionUID = -255921574946294892L;
-
- public UnexpectedMethodException(AMQMethodBody body)
- {
- super("Unexpected method received: " + body.getClass().getName());
- }
-}
diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/AMQChannelTest.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/AMQChannelTest.java deleted file mode 100644 index a6725a6d58..0000000000 --- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/AMQChannelTest.java +++ /dev/null @@ -1,180 +0,0 @@ -/* - * - * 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.mockito.Matchers.eq; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.when; - -import java.util.ArrayList; -import java.util.HashMap; -import java.util.List; -import java.util.Map; - -import org.apache.qpid.framing.AMQDataBlock; -import org.apache.qpid.framing.AMQFrame; -import org.apache.qpid.framing.AMQShortString; -import org.apache.qpid.framing.BasicContentHeaderProperties; -import org.apache.qpid.framing.ChannelCloseBody; -import org.apache.qpid.framing.ContentHeaderBody; -import org.apache.qpid.framing.MessagePublishInfo; -import org.apache.qpid.protocol.AMQConstant; -import org.apache.qpid.server.configuration.BrokerProperties; -import org.apache.qpid.server.exchange.ExchangeImpl; -import org.apache.qpid.server.message.MessageContentSource; -import org.apache.qpid.server.model.Broker; -import org.apache.qpid.server.model.port.AmqpPort; -import org.apache.qpid.server.util.BrokerTestHelper; -import org.apache.qpid.server.virtualhost.VirtualHostImpl; -import org.apache.qpid.test.utils.QpidTestCase; - -public class AMQChannelTest extends QpidTestCase -{ - private VirtualHostImpl _virtualHost; - private AMQProtocolEngine _protocolSession; - private Map<Integer,String> _replies; - private Broker _broker; - - @Override - public void setUp() throws Exception - { - super.setUp(); - BrokerTestHelper.setUp(); - _virtualHost = BrokerTestHelper.createVirtualHost(getTestName()); - _broker = BrokerTestHelper.createBrokerMock(); - AmqpPort port = mock(AmqpPort.class); - when(port.getContextValue(eq(Integer.class), eq(AmqpPort.PORT_MAX_MESSAGE_SIZE))).thenReturn(AmqpPort.DEFAULT_MAX_MESSAGE_SIZE); - - _protocolSession = new InternalTestProtocolSession(_virtualHost, _broker, port) - { - @Override - public void writeReturn(MessagePublishInfo messagePublishInfo, - ContentHeaderBody header, - MessageContentSource msgContent, - int channelId, - int replyCode, - AMQShortString replyText) - { - _replies.put(replyCode, replyText.asString()); - } - }; - _replies = new HashMap<Integer, String>(); - } - - @Override - public void tearDown() throws Exception - { - try - { - _virtualHost.close(); - } - finally - { - BrokerTestHelper.tearDown(); - super.tearDown(); - } - } - - public void testCompareTo() throws Exception - { - AMQChannel channel1 = new AMQChannel(_protocolSession, 1, _virtualHost.getMessageStore()); - - AmqpPort port = mock(AmqpPort.class); - when(port.getContextValue(eq(Integer.class), eq(AmqpPort.PORT_MAX_MESSAGE_SIZE))).thenReturn(AmqpPort.DEFAULT_MAX_MESSAGE_SIZE); - // create a channel with the same channelId but on a different session - AMQChannel channel2 = new AMQChannel(new InternalTestProtocolSession(_virtualHost, _broker, port), 1, _virtualHost.getMessageStore()); - assertFalse("Unexpected compare result", channel1.compareTo(channel2) == 0); - assertEquals("Unexpected compare result", 0, channel1.compareTo(channel1)); - } - - public void testPublishContentHeaderWhenMessageAuthorizationFails() throws Exception - { - setTestSystemProperty(BrokerProperties.PROPERTY_MSG_AUTH, "true"); - AMQChannel channel = new AMQChannel(_protocolSession, 1, _virtualHost.getMessageStore()); - channel.setLocalTransactional(); - - MessagePublishInfo info = new MessagePublishInfo(new AMQShortString("test"), false, false, null); - ExchangeImpl e = mock(ExchangeImpl.class); - ContentHeaderBody contentHeaderBody= mock(ContentHeaderBody.class); - BasicContentHeaderProperties properties = mock(BasicContentHeaderProperties.class); - - when(contentHeaderBody.getProperties()).thenReturn(properties); - when(properties.getUserId()).thenReturn(new AMQShortString(_protocolSession.getAuthorizedPrincipal().getName() + "_incorrect")); - - channel.setPublishFrame(info, e); - channel.publishContentHeader(contentHeaderBody); - channel.commit(null, false); - - assertEquals("Unexpected number of replies", 1, _replies.size()); - assertEquals("Message authorization passed", "Access Refused", _replies.get(403)); - } - - public void testPublishContentHeaderWhenMessageAuthorizationPasses() throws Exception - { - setTestSystemProperty(BrokerProperties.PROPERTY_MSG_AUTH, "true"); - AMQChannel channel = new AMQChannel(_protocolSession, 1, _virtualHost.getMessageStore()); - channel.setLocalTransactional(); - - MessagePublishInfo info = new MessagePublishInfo(new AMQShortString("test"), false, false, null); - ExchangeImpl e = mock(ExchangeImpl.class); - ContentHeaderBody contentHeaderBody= mock(ContentHeaderBody.class); - BasicContentHeaderProperties properties = mock(BasicContentHeaderProperties.class); - - when(contentHeaderBody.getProperties()).thenReturn(properties); - when(properties.getUserId()).thenReturn(new AMQShortString(_protocolSession.getAuthorizedPrincipal().getName())); - - channel.setPublishFrame(info, e); - channel.publishContentHeader(contentHeaderBody); - channel.commit(null, false); - - assertEquals("Unexpected number of replies", 0, _replies.size()); - } - - public void testOverlargeMessage() throws Exception - { - - AmqpPort port = mock(AmqpPort.class); - when(port.getContextValue(eq(Integer.class), eq(AmqpPort.PORT_MAX_MESSAGE_SIZE))).thenReturn(1024); - final List<AMQDataBlock> frames = new ArrayList<>(); - _protocolSession = new InternalTestProtocolSession(_virtualHost, _broker, port) - { - @Override - public synchronized void writeFrame(final AMQDataBlock frame) - { - frames.add(frame); - } - }; - - AMQChannel channel = new AMQChannel(_protocolSession, 1, _virtualHost.getMessageStore()); - - channel.receiveBasicPublish(AMQShortString.EMPTY_STRING, AMQShortString.EMPTY_STRING, false, false); - - final BasicContentHeaderProperties properties = new BasicContentHeaderProperties(); - channel.receiveMessageHeader(properties, 2048l); - - frames.toString(); - - assertEquals(1, frames.size()); - assertEquals(ChannelCloseBody.class, ((AMQFrame) frames.get(0)).getBodyFrame().getClass()); - assertEquals(AMQConstant.MESSAGE_TOO_LARGE.getCode(), ((ChannelCloseBody)((AMQFrame)frames.get(0)).getBodyFrame()).getReplyCode()); - } - -} diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngineTest.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngineTest.java deleted file mode 100644 index cc5da731b8..0000000000 --- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngineTest.java +++ /dev/null @@ -1,106 +0,0 @@ -/* - * - * 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.mockito.Matchers.any; -import static org.mockito.Matchers.eq; -import static org.mockito.Mockito.doThrow; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.when; - -import java.nio.ByteBuffer; -import java.util.HashMap; -import java.util.Map; - -import org.apache.qpid.AMQException; -import org.apache.qpid.framing.AMQFrameDecodingException; -import org.apache.qpid.framing.FieldTable; -import org.apache.qpid.properties.ConnectionStartProperties; -import org.apache.qpid.protocol.AMQConstant; -import org.apache.qpid.server.model.Broker; -import org.apache.qpid.server.model.Transport; -import org.apache.qpid.server.model.port.AmqpPort; -import org.apache.qpid.server.util.BrokerTestHelper; -import org.apache.qpid.server.util.ConnectionScopedRuntimeException; -import org.apache.qpid.server.util.ServerScopedRuntimeException; -import org.apache.qpid.test.utils.QpidTestCase; -import org.apache.qpid.transport.ByteBufferSender; -import org.apache.qpid.transport.SenderException; -import org.apache.qpid.transport.network.NetworkConnection; - -public class AMQProtocolEngineTest extends QpidTestCase -{ - private Broker<?> _broker; - private AmqpPort<?> _port; - private NetworkConnection _network; - private Transport _transport; - - public void setUp() throws Exception - { - super.setUp(); - BrokerTestHelper.setUp(); - _broker = BrokerTestHelper.createBrokerMock(); - when(_broker.getConnection_closeWhenNoRoute()).thenReturn(true); - - _port = mock(AmqpPort.class); - when(_port.getContextValue(eq(Integer.class), eq(AmqpPort.PORT_MAX_MESSAGE_SIZE))).thenReturn(AmqpPort.DEFAULT_MAX_MESSAGE_SIZE); - - _network = mock(NetworkConnection.class); - _transport = Transport.TCP; - } - - public void tearDown() throws Exception - { - try - { - super.tearDown(); - } - finally - { - BrokerTestHelper.tearDown(); - } - } - - public void testSetClientPropertiesForNoRouteProvidedAsString() - { - AMQProtocolEngine engine = new AMQProtocolEngine(_broker, _network, 0, _port, _transport); - assertTrue("Unexpected closeWhenNoRoute before client properties set", engine.isCloseWhenNoRoute()); - - Map<String, Object> clientProperties = new HashMap<String, Object>(); - clientProperties.put(ConnectionStartProperties.QPID_CLOSE_WHEN_NO_ROUTE, Boolean.FALSE.toString()); - engine.setClientProperties(FieldTable.convertToFieldTable(clientProperties)); - - assertFalse("Unexpected closeWhenNoRoute after client properties set", engine.isCloseWhenNoRoute()); - } - - public void testSetClientPropertiesForNoRouteProvidedAsBoolean() - { - AMQProtocolEngine engine = new AMQProtocolEngine(_broker, _network, 0, _port, _transport); - assertTrue("Unexpected closeWhenNoRoute before client properties set", engine.isCloseWhenNoRoute()); - - Map<String, Object> clientProperties = new HashMap<String, Object>(); - clientProperties.put(ConnectionStartProperties.QPID_CLOSE_WHEN_NO_ROUTE, Boolean.FALSE); - engine.setClientProperties(FieldTable.convertToFieldTable(clientProperties)); - - assertFalse("Unexpected closeWhenNoRoute after client properties set", engine.isCloseWhenNoRoute()); - } - -} diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/AckTest.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/AckTest.java deleted file mode 100644 index 406566cd4c..0000000000 --- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/AckTest.java +++ /dev/null @@ -1,353 +0,0 @@ -/* - * - * 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 java.util.ArrayList; -import java.util.EnumSet; -import java.util.Set; - -import org.apache.qpid.AMQException; -import org.apache.qpid.framing.AMQShortString; -import org.apache.qpid.framing.BasicContentHeaderProperties; -import org.apache.qpid.framing.ContentHeaderBody; -import org.apache.qpid.framing.MessagePublishInfo; -import org.apache.qpid.server.consumer.ConsumerImpl; -import org.apache.qpid.server.message.MessageInstance; -import org.apache.qpid.server.queue.AMQQueue; -import org.apache.qpid.server.store.MessageEnqueueRecord; -import org.apache.qpid.server.store.StoredMessage; -import org.apache.qpid.server.store.TestMemoryMessageStore; -import org.apache.qpid.server.txn.AutoCommitTransaction; -import org.apache.qpid.server.txn.ServerTransaction; -import org.apache.qpid.server.util.BrokerTestHelper; -import org.apache.qpid.server.virtualhost.VirtualHostImpl; -import org.apache.qpid.test.utils.QpidTestCase; - -/** - * Tests that acknowledgements are handled correctly. - */ -public class AckTest extends QpidTestCase -{ - private ConsumerTarget_0_8 _subscriptionTarget; - private ConsumerImpl _consumer; - - private AMQProtocolEngine _protocolEngine; - - private TestMemoryMessageStore _messageStore; - - private AMQChannel _channel; - - private AMQQueue _queue; - - private static final AMQShortString DEFAULT_CONSUMER_TAG = new AMQShortString("conTag"); - private VirtualHostImpl _virtualHost; - - @Override - public void setUp() throws Exception - { - super.setUp(); - BrokerTestHelper.setUp(); - _channel = BrokerTestHelper_0_8.createChannel(5); - _protocolEngine = _channel.getConnection(); - _virtualHost = _protocolEngine.getVirtualHost(); - _queue = BrokerTestHelper.createQueue(getTestName(), _virtualHost); - _messageStore = (TestMemoryMessageStore)_virtualHost.getMessageStore(); - } - - @Override - protected void tearDown() throws Exception - { - BrokerTestHelper.tearDown(); - super.tearDown(); - } - - private void publishMessages(int count) throws AMQException - { - publishMessages(count, false); - } - - private void publishMessages(int count, boolean persistent) throws AMQException - { - for (int i = 1; i <= count; i++) - { - MessagePublishInfo publishBody = new MessagePublishInfo(new AMQShortString("someExchange"), false, false, - new AMQShortString("rk")); - BasicContentHeaderProperties b = new BasicContentHeaderProperties(); - ContentHeaderBody cb = new ContentHeaderBody(b); - - if (persistent) - { - //This is DeliveryMode.PERSISTENT - b.setDeliveryMode((byte) 2); - } - - // The test is easier to construct if we have direct access to the subscription - ArrayList<AMQQueue> qs = new ArrayList<AMQQueue>(); - qs.add(_queue); - - final MessageMetaData mmd = new MessageMetaData(publishBody,cb, System.currentTimeMillis()); - - final StoredMessage<MessageMetaData> result =_messageStore.addMessage(mmd).allContentAdded(); - - final StoredMessage storedMessage = result; - final AMQMessage message = new AMQMessage(storedMessage); - ServerTransaction txn = new AutoCommitTransaction(_messageStore); - txn.enqueue(_queue, message, - new ServerTransaction.EnqueueAction() - { - public void postCommit(MessageEnqueueRecord... records) - { - _queue.enqueue(message,null, null); - } - - public void onRollback() - { - } - }); - - } - try - { - Thread.sleep(2000L); - } - catch (InterruptedException e) - { - Thread.currentThread().interrupt(); - } - - } - - /** - * Tests that the acknowledgements are correctly associated with a channel and - * order is preserved when acks are enabled - */ - public void testAckChannelAssociationTest() throws Exception - { - _subscriptionTarget = ConsumerTarget_0_8.createAckTarget(_channel, - DEFAULT_CONSUMER_TAG, - null, - new LimitlessCreditManager()); - _consumer = _queue.addConsumer(_subscriptionTarget, null, AMQMessage.class, DEFAULT_CONSUMER_TAG.toString(), - EnumSet.of(ConsumerImpl.Option.SEES_REQUEUES, - ConsumerImpl.Option.ACQUIRES)); - final int msgCount = 10; - publishMessages(msgCount, true); - UnacknowledgedMessageMap map = _channel.getUnacknowledgedMessageMap(); - assertEquals("Unexpected size for unacknowledged message map",msgCount,map.size()); - - Set<Long> deliveryTagSet = map.getDeliveryTags(); - int i = 1; - for (long deliveryTag : deliveryTagSet) - { - assertTrue(deliveryTag == i); - i++; - MessageInstance unackedMsg = map.get(deliveryTag); - assertTrue(unackedMsg.getOwningResource() == _queue); - } - - } - - /** - * Tests that in no-ack mode no messages are retained - */ - public void testNoAckMode() throws Exception - { - // false arg means no acks expected - _subscriptionTarget = ConsumerTarget_0_8.createNoAckTarget(_channel, - DEFAULT_CONSUMER_TAG, - null, - new LimitlessCreditManager()); - _consumer = _queue.addConsumer(_subscriptionTarget, - null, - AMQMessage.class, - DEFAULT_CONSUMER_TAG.toString(), - EnumSet.of(ConsumerImpl.Option.SEES_REQUEUES, - ConsumerImpl.Option.ACQUIRES)); - final int msgCount = 10; - publishMessages(msgCount); - UnacknowledgedMessageMap map = _channel.getUnacknowledgedMessageMap(); - assertTrue(map.size() == 0); - assertTrue(_messageStore.getMessageCount() == 0); - - - } - - /** - * Tests that in no-ack mode no messages are retained - */ - public void testPersistentNoAckMode() throws Exception - { - // false arg means no acks expected - - _subscriptionTarget = ConsumerTarget_0_8.createNoAckTarget(_channel, - DEFAULT_CONSUMER_TAG, - null, - new LimitlessCreditManager()); - _consumer = _queue.addConsumer(_subscriptionTarget, null, AMQMessage.class, DEFAULT_CONSUMER_TAG.toString(), - EnumSet.of(ConsumerImpl.Option.SEES_REQUEUES, ConsumerImpl.Option.ACQUIRES)); - final int msgCount = 10; - publishMessages(msgCount, true); - - UnacknowledgedMessageMap map = _channel.getUnacknowledgedMessageMap(); - assertTrue(map.size() == 0); - assertTrue(_messageStore.getMessageCount() == 0); - - - } - - /** - * Tests that a single acknowledgement is handled correctly (i.e multiple flag not - * set case) - */ - public void testSingleAckReceivedTest() throws Exception - { - - _subscriptionTarget = ConsumerTarget_0_8.createAckTarget(_channel, - DEFAULT_CONSUMER_TAG, - null, - new LimitlessCreditManager()); - _consumer = _queue.addConsumer(_subscriptionTarget, null, AMQMessage.class, DEFAULT_CONSUMER_TAG.toString(), - EnumSet.of(ConsumerImpl.Option.SEES_REQUEUES, - ConsumerImpl.Option.ACQUIRES)); - - final int msgCount = 10; - publishMessages(msgCount); - - _channel.acknowledgeMessage(5, false); - UnacknowledgedMessageMap map = _channel.getUnacknowledgedMessageMap(); - assertEquals("Map not expected size",msgCount - 1,map.size()); - - Set<Long> deliveryTagSet = map.getDeliveryTags(); - int i = 1; - for (long deliveryTag : deliveryTagSet) - { - assertTrue(deliveryTag == i); - MessageInstance unackedMsg = map.get(deliveryTag); - assertTrue(unackedMsg.getOwningResource() == _queue); - // 5 is the delivery tag of the message that *should* be removed - if (++i == 5) - { - ++i; - } - } - } - - /** - * Tests that a single acknowledgement is handled correctly (i.e multiple flag not - * set case) - */ - public void testMultiAckReceivedTest() throws Exception - { - - _subscriptionTarget = ConsumerTarget_0_8.createAckTarget(_channel, - DEFAULT_CONSUMER_TAG, - null, - new LimitlessCreditManager()); - _consumer = _queue.addConsumer(_subscriptionTarget, null, AMQMessage.class, DEFAULT_CONSUMER_TAG.toString(), - EnumSet.of(ConsumerImpl.Option.SEES_REQUEUES, - ConsumerImpl.Option.ACQUIRES)); - - final int msgCount = 10; - publishMessages(msgCount); - - - - _channel.acknowledgeMessage(5, true); - UnacknowledgedMessageMap map = _channel.getUnacknowledgedMessageMap(); - assertTrue(map.size() == 5); - - Set<Long> deliveryTagSet = map.getDeliveryTags(); - int i = 1; - for (long deliveryTag : deliveryTagSet) - { - assertTrue(deliveryTag == i + 5); - MessageInstance unackedMsg = map.get(deliveryTag); - assertTrue(unackedMsg.getOwningResource() == _queue); - ++i; - } - } - - /** - * Tests that a multiple acknowledgement is handled correctly. When ack'ing all pending msgs. - */ - public void testMultiAckAllReceivedTest() throws Exception - { - - _subscriptionTarget = ConsumerTarget_0_8.createAckTarget(_channel, - DEFAULT_CONSUMER_TAG, - null, - new LimitlessCreditManager()); - _consumer = _queue.addConsumer(_subscriptionTarget, null, AMQMessage.class, DEFAULT_CONSUMER_TAG.toString(), - EnumSet.of(ConsumerImpl.Option.SEES_REQUEUES, - ConsumerImpl.Option.ACQUIRES)); - - final int msgCount = 10; - publishMessages(msgCount); - - _channel.acknowledgeMessage(0, true); - UnacknowledgedMessageMap map = _channel.getUnacknowledgedMessageMap(); - assertTrue(map.size() == 0); - - Set<Long> deliveryTagSet = map.getDeliveryTags(); - int i = 1; - for (long deliveryTag : deliveryTagSet) - { - assertTrue(deliveryTag == i + 5); - MessageInstance unackedMsg = map.get(deliveryTag); - assertTrue(unackedMsg.getOwningResource() == _queue); - ++i; - } - } - - /** - * A regression fixing QPID-1136 showed this up - * - * @throws Exception - */ - public void testMessageDequeueRestoresCreditTest() throws Exception - { - // Send 10 messages - Pre0_10CreditManager creditManager = new Pre0_10CreditManager(0l, 1, _protocolEngine); - - - _subscriptionTarget = ConsumerTarget_0_8.createAckTarget(_channel, DEFAULT_CONSUMER_TAG, null, creditManager); - _consumer = _queue.addConsumer(_subscriptionTarget, null, AMQMessage.class, DEFAULT_CONSUMER_TAG.toString(), - EnumSet.of(ConsumerImpl.Option.SEES_REQUEUES, - ConsumerImpl.Option.ACQUIRES)); - - final int msgCount = 1; - publishMessages(msgCount); - - _consumer.externalStateChange(); - - _channel.acknowledgeMessage(1, false); - - // Check credit available - assertTrue("No credit available", creditManager.hasCredit()); - - } - - public static junit.framework.Test suite() - { - return new junit.framework.TestSuite(AckTest.class); - } -} diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/AcknowledgeTest.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/AcknowledgeTest.java deleted file mode 100644 index 7dd4734e6b..0000000000 --- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/AcknowledgeTest.java +++ /dev/null @@ -1,185 +0,0 @@ -/* - * - * 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 java.util.Collections; -import java.util.List; - -import org.apache.qpid.exchange.ExchangeDefaults; -import org.apache.qpid.framing.AMQShortString; -import org.apache.qpid.server.message.MessageSource; -import org.apache.qpid.server.queue.AMQQueue; -import org.apache.qpid.server.store.MessageCounter; -import org.apache.qpid.server.store.MessageStore; -import org.apache.qpid.server.util.BrokerTestHelper; -import org.apache.qpid.server.virtualhost.VirtualHostImpl; -import org.apache.qpid.test.utils.QpidTestCase; - -public class AcknowledgeTest extends QpidTestCase -{ - private AMQChannel _channel; - private AMQQueue _queue; - private MessageStore _messageStore; - private String _queueName; - - @Override - public void setUp() throws Exception - { - super.setUp(); - BrokerTestHelper.setUp(); - _channel = BrokerTestHelper_0_8.createChannel(); - VirtualHostImpl virtualHost = _channel.getVirtualHost(); - _queueName = getTestName(); - _queue = BrokerTestHelper.createQueue(_queueName, virtualHost); - _messageStore = virtualHost.getMessageStore(); - } - - @Override - public void tearDown() throws Exception - { - try - { - if (_channel != null) - { - _channel.getVirtualHost().close(); - } - } - finally - { - BrokerTestHelper.tearDown(); - super.tearDown(); - } - } - - private AMQChannel getChannel() - { - return _channel; - } - - private InternalTestProtocolSession getSession() - { - return (InternalTestProtocolSession)_channel.getConnection(); - } - - private AMQQueue getQueue() - { - return _queue; - } - - public void testTransactionalSingleAck() throws Exception - { - getChannel().setLocalTransactional(); - runMessageAck(1, 1, 1, false, 0); - } - - public void testTransactionalMultiAck() throws Exception - { - getChannel().setLocalTransactional(); - runMessageAck(10, 1, 5, true, 5); - } - - public void testTransactionalAckAll() throws Exception - { - getChannel().setLocalTransactional(); - runMessageAck(10, 1, 0, true, 0); - } - - public void testNonTransactionalSingleAck() throws Exception - { - runMessageAck(1, 1, 1, false, 0); - } - - public void testNonTransactionalMultiAck() throws Exception - { - runMessageAck(10, 1, 5, true, 5); - } - - public void testNonTransactionalAckAll() throws Exception - { - runMessageAck(10, 1, 0, true, 0); - } - - protected void runMessageAck(int sendMessageCount, long firstDeliveryTag, long acknowledgeDeliveryTag, boolean acknowledgeMultiple, int remainingUnackedMessages) throws Exception - { - //Check store is empty - checkStoreContents(0); - - //Send required messages to the queue - BrokerTestHelper_0_8.publishMessages(getChannel(), - sendMessageCount, - _queueName, - ExchangeDefaults.DEFAULT_EXCHANGE_NAME); - - if (getChannel().isTransactional()) - { - getChannel().commit(null, false); - } - - //Ensure they are stored - checkStoreContents(sendMessageCount); - - //Check that there are no unacked messages - assertEquals("Channel should have no unacked msgs ", 0, getChannel().getUnacknowledgedMessageMap().size()); - - //Subscribe to the queue - AMQShortString subscriber = _channel.consumeFromSource(null, - Collections.<MessageSource>singleton(_queue), - true, null, true, false); - - getQueue().deliverAsync(); - - //Wait for the messages to be delivered - getSession().awaitDelivery(sendMessageCount); - - //Check that they are all waiting to be acknowledged - assertEquals("Channel should have unacked msgs", sendMessageCount, getChannel().getUnacknowledgedMessageMap().size()); - - List<InternalTestProtocolSession.DeliveryPair> messages = getSession().getDelivers(getChannel().getChannelId(), subscriber, sendMessageCount); - - //Double check we received the right number of messages - assertEquals(sendMessageCount, messages.size()); - - //Check that the first message has the expected deliveryTag - assertEquals("First message does not have expected deliveryTag", firstDeliveryTag, messages.get(0).getDeliveryTag()); - - //Send required Acknowledgement - getChannel().acknowledgeMessage(acknowledgeDeliveryTag, acknowledgeMultiple); - - if (getChannel().isTransactional()) - { - getChannel().commit(null, false); - } - - // Check Remaining Acknowledgements - assertEquals("Channel unacked msgs count incorrect", remainingUnackedMessages, getChannel().getUnacknowledgedMessageMap().size()); - - //Check store contents are also correct. - checkStoreContents(remainingUnackedMessages); - } - - private void checkStoreContents(int messageCount) - { - MessageCounter counter = new MessageCounter(); - _messageStore.newMessageStoreReader().visitMessages(counter); - assertEquals("Message header count incorrect in the MetaDataMap", messageCount, counter.getCount()); - } - -} diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/BrokerTestHelper_0_8.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/BrokerTestHelper_0_8.java deleted file mode 100644 index a0a3d60458..0000000000 --- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/BrokerTestHelper_0_8.java +++ /dev/null @@ -1,109 +0,0 @@ -/* - * - * 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.mockito.Matchers.eq; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.when; - -import org.apache.qpid.AMQException; -import org.apache.qpid.framing.AMQShortString; -import org.apache.qpid.framing.BasicContentHeaderProperties; -import org.apache.qpid.framing.ContentHeaderBody; -import org.apache.qpid.framing.MessagePublishInfo; -import org.apache.qpid.server.message.MessageDestination; -import org.apache.qpid.server.model.port.AmqpPort; -import org.apache.qpid.server.util.BrokerTestHelper; -import org.apache.qpid.server.virtualhost.VirtualHostImpl; - -public class BrokerTestHelper_0_8 extends BrokerTestHelper -{ - - public static AMQChannel createChannel(int channelId, AMQProtocolEngine session) throws AMQException - { - AMQChannel channel = new AMQChannel(session, channelId, session.getVirtualHost().getMessageStore()); - session.addChannel(channel); - return channel; - } - - public static AMQChannel createChannel(int channelId) throws Exception - { - InternalTestProtocolSession session = createProtocolSession(); - return createChannel(channelId, session); - } - - public static AMQChannel createChannel() throws Exception - { - return createChannel(1); - } - - public static InternalTestProtocolSession createProtocolSession() throws Exception - { - return createProtocolSession("test"); - } - - public static InternalTestProtocolSession createProtocolSession(String hostName) throws Exception - { - VirtualHostImpl virtualHost = createVirtualHost(hostName); - - AmqpPort port = mock(AmqpPort.class); - when(port.getContextValue(eq(Integer.class), eq(AmqpPort.PORT_MAX_MESSAGE_SIZE))).thenReturn(AmqpPort.DEFAULT_MAX_MESSAGE_SIZE); - return new InternalTestProtocolSession(virtualHost, createBrokerMock(), port); - } - - public static void publishMessages(AMQChannel channel, int numberOfMessages, String queueName, String exchangeName) - throws AMQException - { - AMQShortString routingKey = new AMQShortString(queueName); - AMQShortString exchangeNameAsShortString = new AMQShortString(exchangeName); - MessagePublishInfo info = new MessagePublishInfo(exchangeNameAsShortString, false, false, routingKey); - - MessageDestination destination; - if(exchangeName == null || "".equals(exchangeName)) - { - destination = channel.getVirtualHost().getDefaultDestination(); - } - else - { - destination = channel.getVirtualHost().getExchange(exchangeName); - } - for (int count = 0; count < numberOfMessages; count++) - { - channel.setPublishFrame(info, destination); - - - // Set Minimum properties - BasicContentHeaderProperties properties = new BasicContentHeaderProperties(); - - - properties.setExpiration(0L); - properties.setTimestamp(System.currentTimeMillis()); - - // Make Message Persistent - properties.setDeliveryMode((byte) 2); - - ContentHeaderBody headerBody = new ContentHeaderBody(properties, 0); - - channel.publishContentHeader(headerBody); - } - channel.sync(); - } -} diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/ExtractResendAndRequeueTest.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/ExtractResendAndRequeueTest.java deleted file mode 100644 index 6d3e648369..0000000000 --- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/ExtractResendAndRequeueTest.java +++ /dev/null @@ -1,184 +0,0 @@ -/* - * - * 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 junit.framework.TestCase; - -import org.apache.qpid.AMQException; -import org.apache.qpid.server.consumer.ConsumerImpl; -import org.apache.qpid.server.message.MessageInstance; -import org.apache.qpid.server.message.ServerMessage; -import org.apache.qpid.server.queue.AMQQueue; -import org.apache.qpid.server.queue.QueueEntry; -import org.mockito.invocation.InvocationOnMock; -import org.mockito.stubbing.Answer; - -import java.util.LinkedHashMap; -import java.util.LinkedList; -import java.util.Map; - -import static org.mockito.Mockito.doAnswer; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.when; - -/** - * QPID-1385 : Race condition between added to unacked map and resending due to a rollback. - * - * In AMQChannel _unackedMap.clear() was done after the visit. This meant that the clear was not in the same - * synchronized block as as the preparation to resend. - * - * This clearing/prep for resend was done as a result of the rollback call. HOWEVER, the delivery thread was still - * in the process of sending messages to the client. It is therefore possible that a message could block on the - * _unackedMap lock waiting for the visit to complete so that it can add the new message to the unackedMap.... - * which is then cleared by the resend/rollback thread. - * - * This problem was encountered by the testSend2ThenRollback test. - * - * To try and increase the chance of the race condition occurring this test will send multiple messages so that the - * delivery thread will be in progress while the rollback method is called. Hopefully this will cause the - * deliveryTag to be lost - */ -public class ExtractResendAndRequeueTest extends TestCase -{ - - private UnacknowledgedMessageMapImpl _unacknowledgedMessageMap; - private static final int INITIAL_MSG_COUNT = 10; - private AMQQueue _queue; - private LinkedList<MessageInstance> _referenceList = new LinkedList<MessageInstance>(); - private ConsumerImpl _consumer; - private boolean _queueDeleted; - - @Override - public void setUp() throws AMQException - { - _queueDeleted = false; - _unacknowledgedMessageMap = new UnacknowledgedMessageMapImpl(100); - _queue = mock(AMQQueue.class); - when(_queue.getName()).thenReturn(getName()); - when(_queue.isDeleted()).thenReturn(_queueDeleted); - _consumer = mock(ConsumerImpl.class); - when(_consumer.getConsumerNumber()).thenReturn(ConsumerImpl.CONSUMER_NUMBER_GENERATOR.getAndIncrement()); - - - long id = 0; - - // Add initial messages to QueueEntryList - for (int count = 0; count < INITIAL_MSG_COUNT; count++) - { - ServerMessage msg = mock(ServerMessage.class); - when(msg.getMessageNumber()).thenReturn(id); - final QueueEntry entry = mock(QueueEntry.class); - when(entry.getMessage()).thenReturn(msg); - when(entry.getQueue()).thenReturn(_queue); - when(entry.isQueueDeleted()).thenReturn(_queueDeleted); - doAnswer(new Answer() - { - @Override - public Object answer(final InvocationOnMock invocation) throws Throwable - { - when(entry.isDeleted()).thenReturn(true); - return null; - } - }).when(entry).delete(); - - _unacknowledgedMessageMap.add(id, entry); - _referenceList.add(entry); - //Increment ID; - id++; - } - - assertEquals("Map does not contain correct setup data", INITIAL_MSG_COUNT, _unacknowledgedMessageMap.size()); - } - - /** - * Helper method to create a new subscription and acquire the given messages. - * - * @param messageList The messages to acquire - * - * @return Subscription that performed the acquire - */ - private void acquireMessages(LinkedList<MessageInstance> messageList) - { - - // Acquire messages in subscription - for(MessageInstance entry : messageList) - { - when(entry.getDeliveredConsumer()).thenReturn(_consumer); - } - } - - /** - * This is the normal consumer rollback method. - * - * An active consumer that has acquired messages expects those messages to be reset when rollback is requested. - * - * This test validates that the msgToResend map includes all the messages and none are left behind. - * - * @throws AMQException the visit interface throws this - */ - public void testResend() throws AMQException - { - //We don't need the subscription object here. - acquireMessages(_referenceList); - - final Map<Long, MessageInstance> msgToRequeue = new LinkedHashMap<Long, MessageInstance>(); - final Map<Long, MessageInstance> msgToResend = new LinkedHashMap<Long, MessageInstance>(); - - // requeueIfUnableToResend doesn't matter here. - _unacknowledgedMessageMap.visit(new ExtractResendAndRequeue(_unacknowledgedMessageMap, msgToRequeue, - msgToResend)); - - assertEquals("Message count for resend not correct.", INITIAL_MSG_COUNT, msgToResend.size()); - assertEquals("Message count for requeue not correct.", 0, msgToRequeue.size()); - assertEquals("Map was not emptied", 0, _unacknowledgedMessageMap.size()); - } - - /** - * This is the normal consumer close method. - * - * When a consumer that has acquired messages expects closes the messages that it has acquired should be removed from - * the unacknowledgedMap and placed in msgToRequeue - * - * This test validates that the msgToRequeue map includes all the messages and none are left behind. - * - * @throws AMQException the visit interface throws this - */ - public void testRequeueDueToSubscriptionClosure() throws AMQException - { - acquireMessages(_referenceList); - - // Close subscription - when(_consumer.isClosed()).thenReturn(true); - - final Map<Long, MessageInstance> msgToRequeue = new LinkedHashMap<Long, MessageInstance>(); - final Map<Long, MessageInstance> msgToResend = new LinkedHashMap<Long, MessageInstance>(); - - // requeueIfUnableToResend doesn't matter here. - _unacknowledgedMessageMap.visit(new ExtractResendAndRequeue(_unacknowledgedMessageMap, msgToRequeue, - msgToResend)); - - assertEquals("Message count for resend not correct.", 0, msgToResend.size()); - assertEquals("Message count for requeue not correct.", INITIAL_MSG_COUNT, msgToRequeue.size()); - assertEquals("Map was not emptied", 0, _unacknowledgedMessageMap.size()); - } - - -} diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/InternalTestProtocolSession.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/InternalTestProtocolSession.java deleted file mode 100644 index 7e68bee661..0000000000 --- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/InternalTestProtocolSession.java +++ /dev/null @@ -1,357 +0,0 @@ -/* - * - * 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 java.net.InetSocketAddress; -import java.net.SocketAddress; -import java.nio.ByteBuffer; -import java.security.Principal; -import java.util.ArrayList; -import java.util.Collections; -import java.util.HashMap; -import java.util.LinkedList; -import java.util.List; -import java.util.Map; -import java.util.concurrent.atomic.AtomicInteger; -import java.util.concurrent.atomic.AtomicLong; - -import javax.security.auth.Subject; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import org.apache.qpid.AMQException; -import org.apache.qpid.framing.AMQShortString; -import org.apache.qpid.framing.ContentHeaderBody; -import org.apache.qpid.framing.MessagePublishInfo; -import org.apache.qpid.server.consumer.ConsumerImpl; -import org.apache.qpid.server.message.InstanceProperties; -import org.apache.qpid.server.message.MessageContentSource; -import org.apache.qpid.server.message.ServerMessage; -import org.apache.qpid.server.model.Broker; -import org.apache.qpid.server.model.port.AmqpPort; -import org.apache.qpid.server.security.auth.AuthenticatedPrincipal; -import org.apache.qpid.server.security.auth.UsernamePrincipal; -import org.apache.qpid.server.virtualhost.VirtualHostImpl; -import org.apache.qpid.transport.ByteBufferSender; -import org.apache.qpid.transport.network.NetworkConnection; - -public class InternalTestProtocolSession extends AMQProtocolEngine implements ProtocolOutputConverter -{ - private static final Logger _logger = LoggerFactory.getLogger(InternalTestProtocolSession.class); - // ChannelID(LIST) -> LinkedList<Pair> - private final Map<Integer, Map<String, LinkedList<DeliveryPair>>> _channelDelivers; - private AtomicInteger _deliveryCount = new AtomicInteger(0); - private static final AtomicLong ID_GENERATOR = new AtomicLong(0); - - public InternalTestProtocolSession(VirtualHostImpl virtualHost, Broker<?> broker, final AmqpPort<?> port) throws AMQException - { - super(broker, new TestNetworkConnection(), ID_GENERATOR.getAndIncrement(), port, null); - - _channelDelivers = new HashMap<Integer, Map<String, LinkedList<DeliveryPair>>>(); - - setTestAuthorizedSubject(); - setVirtualHost(virtualHost); - } - - private void setTestAuthorizedSubject() - { - Principal principal = new AuthenticatedPrincipal(new UsernamePrincipal("InternalTestProtocolSession")); - Subject authorizedSubject = new Subject( - true, - Collections.singleton(principal), - Collections.emptySet(), - Collections.emptySet()); - - setAuthorizedSubject(authorizedSubject); - } - - public ProtocolOutputConverter getProtocolOutputConverter() - { - return this; - } - - public byte getProtocolMajorVersion() - { - return (byte) 8; - } - - public void writeReturn(MessagePublishInfo messagePublishInfo, - ContentHeaderBody header, - MessageContentSource msgContent, - int channelId, - int replyCode, - AMQShortString replyText) - { - //To change body of implemented methods use File | Settings | File Templates. - } - - public byte getProtocolMinorVersion() - { - return (byte) 0; - } - - // *** - - public List<DeliveryPair> getDelivers(int channelId, AMQShortString consumerTag, int count) - { - synchronized (_channelDelivers) - { - List<DeliveryPair> all =_channelDelivers.get(channelId).get(AMQShortString.toString(consumerTag)); - - if (all == null) - { - return new ArrayList<DeliveryPair>(0); - } - - List<DeliveryPair> msgs = all.subList(0, count); - - List<DeliveryPair> response = new ArrayList<DeliveryPair>(msgs); - - //Remove the msgs from the receivedList. - msgs.clear(); - - return response; - } - } - - public ClientDeliveryMethod createDeliveryMethod(int channelId) - { - return new InternalWriteDeliverMethod(channelId); - } - - public void confirmConsumerAutoClose(int channelId, AMQShortString consumerTag) - { - } - - public long writeDeliver(final ServerMessage msg, - final InstanceProperties props, int channelId, - long deliveryTag, - AMQShortString consumerTag) - { - _deliveryCount.incrementAndGet(); - long size = msg.getSize(); - synchronized (_channelDelivers) - { - Map<String, LinkedList<DeliveryPair>> consumers = _channelDelivers.get(channelId); - - if (consumers == null) - { - consumers = new HashMap<String, LinkedList<DeliveryPair>>(); - _channelDelivers.put(channelId, consumers); - } - - LinkedList<DeliveryPair> consumerDelivers = consumers.get(AMQShortString.toString(consumerTag)); - - if (consumerDelivers == null) - { - consumerDelivers = new LinkedList<DeliveryPair>(); - consumers.put(consumerTag.toString(), consumerDelivers); - } - - consumerDelivers.add(new DeliveryPair(deliveryTag, msg)); - } - return size; - } - - public long writeGetOk(final ServerMessage msg, - final InstanceProperties props, - int channelId, - long deliveryTag, - int queueSize) - { - return msg.getSize(); - } - - public void awaitDelivery(int msgs) - { - while (msgs > _deliveryCount.get()) - { - try - { - Thread.sleep(100); - } - catch (InterruptedException e) - { - _logger.error("Thread interrupted", e); - } - } - } - - public class DeliveryPair - { - private long _deliveryTag; - private ServerMessage _message; - - public DeliveryPair(long deliveryTag, ServerMessage message) - { - _deliveryTag = deliveryTag; - _message = message; - } - - public ServerMessage getMessage() - { - return _message; - } - - public long getDeliveryTag() - { - return _deliveryTag; - } - } - - public void closeNetworkConnection() - { - // Override as we don't have a real IOSession to close. - // The alternative is to fully implement the TestIOSession to return a CloseFuture from close(); - // Then the AMQMinaProtocolSession can join on the returning future without a NPE. - } - - private class InternalWriteDeliverMethod implements ClientDeliveryMethod - { - private int _channelId; - - public InternalWriteDeliverMethod(int channelId) - { - _channelId = channelId; - } - - - @Override - public long deliverToClient(ConsumerImpl sub, ServerMessage message, - InstanceProperties props, long deliveryTag) - { - _deliveryCount.incrementAndGet(); - long size = message.getSize(); - synchronized (_channelDelivers) - { - Map<String, LinkedList<DeliveryPair>> consumers = _channelDelivers.get(_channelId); - - if (consumers == null) - { - consumers = new HashMap<String, LinkedList<DeliveryPair>>(); - _channelDelivers.put(_channelId, consumers); - } - - LinkedList<DeliveryPair> consumerDelivers = consumers.get(sub.getName()); - - if (consumerDelivers == null) - { - consumerDelivers = new LinkedList<DeliveryPair>(); - consumers.put(sub.getName(), consumerDelivers); - } - - consumerDelivers.add(new DeliveryPair(deliveryTag, message)); - } - return size; - } - } - - void assertState(final ConnectionState requiredState) - { - // no-op - } - - - private static final AtomicInteger portNumber = new AtomicInteger(0); - - private static class TestNetworkConnection implements NetworkConnection - { - private String _remoteHost = "127.0.0.1"; - private String _localHost = "127.0.0.1"; - private int _port = portNumber.incrementAndGet(); - private final ByteBufferSender _sender; - - public TestNetworkConnection() - { - _sender = new ByteBufferSender() - { - public void send(ByteBuffer msg) - { - } - - public void flush() - { - } - - public void close() - { - } - }; - } - - @Override - public SocketAddress getLocalAddress() - { - return new InetSocketAddress(_localHost, _port); - } - - @Override - public SocketAddress getRemoteAddress() - { - return new InetSocketAddress(_remoteHost, _port); - } - - @Override - public void setMaxReadIdle(int idleTime) - { - } - - @Override - public Principal getPeerPrincipal() - { - return null; - } - - @Override - public int getMaxReadIdle() - { - return 0; - } - - @Override - public int getMaxWriteIdle() - { - return 0; - } - - @Override - public void setMaxWriteIdle(int idleTime) - { - } - - @Override - public void close() - { - } - - @Override - public ByteBufferSender getSender() - { - return _sender; - } - - @Override - public void start() - { - } - } -} diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/LimitlessCreditManager.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/LimitlessCreditManager.java deleted file mode 100644 index c4c89ac24a..0000000000 --- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/LimitlessCreditManager.java +++ /dev/null @@ -1,47 +0,0 @@ -/* -* -* 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 org.apache.qpid.server.flow.AbstractFlowCreditManager; -import org.apache.qpid.server.flow.FlowCreditManager; - -public class LimitlessCreditManager extends AbstractFlowCreditManager implements FlowCreditManager -{ - - public void restoreCredit(long messageCredit, long bytesCredit) - { - } - - public void removeAllCredit() - { - } - - public boolean hasCredit() - { - return true; - } - - public boolean useCreditForMessage(long msgSize) - { - return true; - } -} diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/MaxChannelsTest.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/MaxChannelsTest.java deleted file mode 100644 index 459fc94484..0000000000 --- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/MaxChannelsTest.java +++ /dev/null @@ -1,72 +0,0 @@ -/* - * - * 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 org.apache.qpid.server.util.BrokerTestHelper; -import org.apache.qpid.test.utils.QpidTestCase; - -public class MaxChannelsTest extends QpidTestCase -{ - private AMQProtocolEngine _session; - - @Override - public void setUp() throws Exception - { - super.setUp(); - BrokerTestHelper.setUp(); - _session = BrokerTestHelper_0_8.createProtocolSession(); - } - - public void testChannels() throws Exception - { - // check the channel count is correct - int channelCount = _session.getChannels().size(); - assertEquals("Initial channel count wrong", 0, channelCount); - - long maxChannels = 10L; - _session.setMaximumNumberOfChannels(maxChannels); - assertEquals("Number of channels not correctly set.", maxChannels, _session.getMaximumNumberOfChannels()); - - for (long currentChannel = 1L; currentChannel <= maxChannels; currentChannel++) - { - _session.receiveChannelOpen( (int) currentChannel); - } - assertFalse("Connection should not be closed after opening " + maxChannels + " channels",_session.isClosed()); - assertEquals("Maximum number of channels not set.", maxChannels, _session.getChannels().size()); - _session.receiveChannelOpen((int) maxChannels+1); - assertTrue("Connection should be closed after opening " + (maxChannels + 1) + " channels",_session.isClosed()); - } - - @Override - public void tearDown() throws Exception - { - try - { - _session.getVirtualHost().close(); - } - finally - { - BrokerTestHelper.tearDown(); - super.tearDown(); - } - } - -} diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/MockAMQMessage.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/MockAMQMessage.java deleted file mode 100644 index 1cc3607298..0000000000 --- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/MockAMQMessage.java +++ /dev/null @@ -1,40 +0,0 @@ -/* - * - * 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; - -public class MockAMQMessage extends AMQMessage -{ - public MockAMQMessage(long messageId) - { - super(new MockStoredMessage(messageId)); - } - - public MockAMQMessage(long messageId, String headerName, Object headerValue) - { - super(new MockStoredMessage(messageId, headerName, headerValue)); - } - - @Override - public long getSize() - { - return 0l; - } -} diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/MockStoredMessage.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/MockStoredMessage.java deleted file mode 100755 index c6aea39aa6..0000000000 --- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/MockStoredMessage.java +++ /dev/null @@ -1,127 +0,0 @@ -/* -* -* 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 java.nio.ByteBuffer; - -import org.apache.qpid.framing.BasicContentHeaderProperties; -import org.apache.qpid.framing.ContentHeaderBody; -import org.apache.qpid.framing.FieldTable; -import org.apache.qpid.framing.MessagePublishInfo; -import org.apache.qpid.server.store.MessageHandle; -import org.apache.qpid.server.store.StoredMessage; - -public class MockStoredMessage implements StoredMessage<MessageMetaData>, MessageHandle<MessageMetaData> -{ - private long _messageId; - private MessageMetaData _metaData; - private final ByteBuffer _content; - - public MockStoredMessage(long messageId) - { - this(messageId, (String)null, null); - } - - public MockStoredMessage(long messageId, String headerName, Object headerValue) - { - this(messageId, new MessagePublishInfo(null, false, false, null), new ContentHeaderBody(new BasicContentHeaderProperties()), headerName, headerValue); - } - - public MockStoredMessage(long messageId, MessagePublishInfo info, ContentHeaderBody chb) - { - this(messageId, info, chb, null, null); - } - - public MockStoredMessage(long messageId, MessagePublishInfo info, ContentHeaderBody chb, String headerName, Object headerValue) - { - _messageId = messageId; - if (headerName != null) - { - FieldTable headers = new FieldTable(); - headers.setString(headerName, headerValue == null? null :String.valueOf(headerValue)); - ( chb.getProperties()).setHeaders(headers); - } - _metaData = new MessageMetaData(info, chb); - _content = ByteBuffer.allocate(_metaData.getContentSize()); - } - - public MessageMetaData getMetaData() - { - return _metaData; - } - - public long getMessageNumber() - { - return _messageId; - } - - public void addContent(ByteBuffer src) - { - src = src.duplicate(); - _content.put(src); - } - - @Override - public StoredMessage<MessageMetaData> allContentAdded() - { - _content.flip(); - return this; - } - - public int getContent(int offset, ByteBuffer dst) - { - ByteBuffer src = _content.duplicate(); - src.position(offset); - src = src.slice(); - if(dst.remaining() < src.limit()) - { - src.limit(dst.remaining()); - } - dst.put(src); - return src.limit(); - } - - - - public ByteBuffer getContent(int offsetInMessage, int size) - { - ByteBuffer buf = ByteBuffer.allocate(size); - getContent(offsetInMessage, buf); - buf.position(0); - return buf; - } - - public void remove() - { - } - - @Override - public boolean isInMemory() - { - return true; - } - - @Override - public boolean flowToDisk() - { - return false; - } -} diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/QueueBrowserUsesNoAckTest.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/QueueBrowserUsesNoAckTest.java deleted file mode 100644 index 264350ff8d..0000000000 --- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/QueueBrowserUsesNoAckTest.java +++ /dev/null @@ -1,150 +0,0 @@ -/* - * - * 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 java.util.Collections; -import java.util.List; - -import org.apache.qpid.common.AMQPFilterTypes; -import org.apache.qpid.exchange.ExchangeDefaults; -import org.apache.qpid.framing.AMQShortString; -import org.apache.qpid.framing.FieldTable; -import org.apache.qpid.server.message.MessageSource; -import org.apache.qpid.server.queue.AMQQueue; -import org.apache.qpid.server.store.MessageCounter; -import org.apache.qpid.server.store.MessageStore; -import org.apache.qpid.server.util.BrokerTestHelper; -import org.apache.qpid.server.virtualhost.VirtualHostImpl; -import org.apache.qpid.test.utils.QpidTestCase; - -public class QueueBrowserUsesNoAckTest extends QpidTestCase -{ - private AMQChannel _channel; - private AMQQueue _queue; - private MessageStore _messageStore; - private String _queueName; - - @Override - public void setUp() throws Exception - { - super.setUp(); - BrokerTestHelper.setUp(); - _channel = BrokerTestHelper_0_8.createChannel(); - VirtualHostImpl virtualHost = _channel.getVirtualHost(); - _queueName = getTestName(); - _queue = BrokerTestHelper.createQueue(_queueName, virtualHost); - _messageStore = virtualHost.getMessageStore(); - } - - @Override - public void tearDown() throws Exception - { - try - { - if (_channel != null) - { - _channel.getVirtualHost().close(); - } - } - finally - { - BrokerTestHelper.tearDown(); - super.tearDown(); - } - } - - private AMQChannel getChannel() - { - return _channel; - } - - private InternalTestProtocolSession getSession() - { - return (InternalTestProtocolSession)_channel.getConnection(); - } - - private AMQQueue getQueue() - { - return _queue; - } - - public void testQueueBrowserUsesNoAck() throws Exception - { - int sendMessageCount = 2; - int prefetch = 1; - - //Check store is empty - checkStoreContents(0); - - //Send required messages to the queue - BrokerTestHelper_0_8.publishMessages(getChannel(), - sendMessageCount, - _queueName, - ExchangeDefaults.DEFAULT_EXCHANGE_NAME); - - //Ensure they are stored - checkStoreContents(sendMessageCount); - - //Check that there are no unacked messages - assertEquals("Channel should have no unacked msgs ", 0, - getChannel().getUnacknowledgedMessageMap().size()); - - //Set the prefetch on the session to be less than the sent messages - getChannel().setCredit(0, prefetch); - - //browse the queue - AMQShortString browser = browse(getChannel(), getQueue()); - - getQueue().deliverAsync(); - - //Wait for messages to fill the prefetch - getSession().awaitDelivery(prefetch); - - //Get those messages - List<InternalTestProtocolSession.DeliveryPair> messages = - getSession().getDelivers(getChannel().getChannelId(), browser, - prefetch); - - //Ensure we received the prefetched messages - assertEquals(prefetch, messages.size()); - - //Check the process didn't suspend the subscription as this would - // indicate we are using the prefetch credit. i.e. using acks not No-Ack - assertTrue("The subscription has been suspended", - !getChannel().getSubscription(browser).isSuspended()); - } - - private void checkStoreContents(int messageCount) - { - MessageCounter counter = new MessageCounter(); - _messageStore.newMessageStoreReader().visitMessages(counter); - - assertEquals("Message header count incorrect in the MetaDataMap", messageCount, counter.getCount()); - } - - private AMQShortString browse(AMQChannel channel, AMQQueue queue) throws Exception - { - FieldTable filters = new FieldTable(); - filters.put(AMQPFilterTypes.NO_CONSUME.getValue(), true); - - return channel.consumeFromSource(null, Collections.<MessageSource>singleton(queue), true, filters, true, false); - } -} diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/ReferenceCountingTest.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/ReferenceCountingTest.java deleted file mode 100644 index de3b68a0bc..0000000000 --- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/ReferenceCountingTest.java +++ /dev/null @@ -1,169 +0,0 @@ -/* - * - * 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 java.util.UUID; - -import org.apache.qpid.AMQException; -import org.apache.qpid.framing.BasicContentHeaderProperties; -import org.apache.qpid.framing.ContentHeaderBody; -import org.apache.qpid.framing.MessagePublishInfo; -import org.apache.qpid.server.message.EnqueueableMessage; -import org.apache.qpid.server.message.MessageReference; -import org.apache.qpid.server.store.MessageCounter; -import org.apache.qpid.server.store.MessageDurability; -import org.apache.qpid.server.store.StoredMessage; -import org.apache.qpid.server.store.TestMemoryMessageStore; -import org.apache.qpid.server.store.Transaction; -import org.apache.qpid.server.store.TransactionLogResource; -import org.apache.qpid.test.utils.QpidTestCase; - -/** - * Tests that reference counting works correctly with AMQMessage and the message store - */ -public class ReferenceCountingTest extends QpidTestCase -{ - private TestMemoryMessageStore _store; - - - protected void setUp() throws Exception - { - _store = new TestMemoryMessageStore(); - } - - /** - * Check that when the reference count is decremented the message removes itself from the store - */ - public void testMessageGetsRemoved() throws AMQException - { - ContentHeaderBody chb = createPersistentContentHeader(); - - MessagePublishInfo info = new MessagePublishInfo(null, false, false, null); - - final MessageMetaData mmd = new MessageMetaData(info, chb); - - StoredMessage storedMessage = _store.addMessage(mmd).allContentAdded(); - Transaction txn = _store.newTransaction(); - txn.enqueueMessage(createTransactionLogResource("dummyQ"), createEnqueueableMessage(storedMessage)); - txn.commitTran(); - AMQMessage message = new AMQMessage(storedMessage); - - MessageReference ref = message.newReference(); - - assertEquals(1, getStoreMessageCount()); - - ref.release(); - - assertEquals(0, getStoreMessageCount()); - } - - private int getStoreMessageCount() - { - MessageCounter counter = new MessageCounter(); - _store.newMessageStoreReader().visitMessages(counter); - return counter.getCount(); - } - - private ContentHeaderBody createPersistentContentHeader() - { - BasicContentHeaderProperties bchp = new BasicContentHeaderProperties(); - bchp.setDeliveryMode((byte)2); - ContentHeaderBody chb = new ContentHeaderBody(bchp); - return chb; - } - - public void testMessageRemains() throws AMQException - { - - MessagePublishInfo info = new MessagePublishInfo(null, false, false, null); - - final ContentHeaderBody chb = createPersistentContentHeader(); - - final MessageMetaData mmd = new MessageMetaData(info, chb); - - StoredMessage storedMessage = _store.addMessage(mmd).allContentAdded(); - Transaction txn = _store.newTransaction(); - txn.enqueueMessage(createTransactionLogResource("dummyQ"), createEnqueueableMessage(storedMessage)); - txn.commitTran(); - AMQMessage message = new AMQMessage(storedMessage); - - - MessageReference ref = message.newReference(); - - assertEquals(1, getStoreMessageCount()); - MessageReference ref2 = message.newReference(); - ref.release(); - assertEquals(1, getStoreMessageCount()); - } - - private TransactionLogResource createTransactionLogResource(final String queueName) - { - return new TransactionLogResource() - { - @Override - public String getName() - { - return queueName; - } - - @Override - public UUID getId() - { - return UUID.nameUUIDFromBytes(queueName.getBytes()); - } - - @Override - public MessageDurability getMessageDurability() - { - return MessageDurability.DEFAULT; - } - }; - } - - private EnqueueableMessage createEnqueueableMessage(final StoredMessage storedMessage) - { - return new EnqueueableMessage() - { - @Override - public long getMessageNumber() - { - return storedMessage.getMessageNumber(); - } - - @Override - public boolean isPersistent() - { - return true; - } - - @Override - public StoredMessage getStoredMessage() - { - return storedMessage; - } - }; - } - - public static junit.framework.Test suite() - { - return new junit.framework.TestSuite(ReferenceCountingTest.class); - } -} diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/UnacknowledgedMessageMapTest.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/UnacknowledgedMessageMapTest.java deleted file mode 100644 index ca52173e66..0000000000 --- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/UnacknowledgedMessageMapTest.java +++ /dev/null @@ -1,84 +0,0 @@ -/* - * - * 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.mockito.Mockito.mock; -import static org.mockito.Mockito.when; - -import java.util.Collection; - -import junit.framework.TestCase; - -import org.apache.qpid.server.message.MessageInstance; - -public class UnacknowledgedMessageMapTest extends TestCase -{ - public void testDeletedMessagesCantBeAcknowledged() - { - UnacknowledgedMessageMap map = new UnacknowledgedMessageMapImpl(100); - final int expectedSize = 5; - MessageInstance[] msgs = populateMap(map,expectedSize); - assertEquals(expectedSize,map.size()); - Collection<MessageInstance> acknowledged = map.acknowledge(100, true); - assertEquals(expectedSize, acknowledged.size()); - assertEquals(0,map.size()); - for(int i = 0; i < expectedSize; i++) - { - assertTrue("Message " + i + " is missing", acknowledged.contains(msgs[i])); - } - - map = new UnacknowledgedMessageMapImpl(100); - msgs = populateMap(map,expectedSize); - // simulate some messages being ttl expired - when(msgs[2].lockAcquisition()).thenReturn(Boolean.FALSE); - when(msgs[4].lockAcquisition()).thenReturn(Boolean.FALSE); - - assertEquals(expectedSize,map.size()); - - - acknowledged = map.acknowledge(100, true); - assertEquals(expectedSize-2, acknowledged.size()); - assertEquals(0,map.size()); - for(int i = 0; i < expectedSize; i++) - { - assertEquals(i != 2 && i != 4, acknowledged.contains(msgs[i])); - } - - } - - public MessageInstance[] populateMap(final UnacknowledgedMessageMap map, int size) - { - MessageInstance[] msgs = new MessageInstance[size]; - for(int i = 0; i < size; i++) - { - msgs[i] = createMessageInstance(i); - map.add((long)i,msgs[i]); - } - return msgs; - } - - private MessageInstance createMessageInstance(final int id) - { - MessageInstance instance = mock(MessageInstance.class); - when(instance.lockAcquisition()).thenReturn(Boolean.TRUE); - return instance; - } -} |
