You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by rg...@apache.org on 2014/10/10 11:59:58 UTC
svn commit: r1630749 [4/6] - in
/qpid/branches/QPID-6125-ProtocolRefactoring/java: ./
bdbstore/jmx/src/main/java/org/apache/qpid/server/store/berkeleydb/jmx/
bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/
bdbstore/src/main/java/org/apa...
Modified: qpid/branches/QPID-6125-ProtocolRefactoring/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ServerMethodDispatcherImpl.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6125-ProtocolRefactoring/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ServerMethodDispatcherImpl.java?rev=1630749&r1=1630748&r2=1630749&view=diff
==============================================================================
--- qpid/branches/QPID-6125-ProtocolRefactoring/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ServerMethodDispatcherImpl.java (original)
+++ qpid/branches/QPID-6125-ProtocolRefactoring/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ServerMethodDispatcherImpl.java Fri Oct 10 09:59:55 2014
@@ -1,2237 +1,2266 @@
-/*
- *
- * 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.security.AccessControlException;
-import java.util.Collection;
-import java.util.EnumSet;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.UUID;
-
-import javax.security.sasl.SaslException;
-import javax.security.sasl.SaslServer;
-
-import org.apache.log4j.Logger;
-
-import org.apache.qpid.AMQConnectionException;
-import org.apache.qpid.AMQException;
-import org.apache.qpid.exchange.ExchangeDefaults;
-import org.apache.qpid.framing.*;
-import org.apache.qpid.protocol.AMQConstant;
-import org.apache.qpid.server.consumer.ConsumerImpl;
-import org.apache.qpid.server.exchange.ExchangeImpl;
-import org.apache.qpid.server.filter.AMQInvalidArgumentException;
-import org.apache.qpid.server.flow.FlowCreditManager;
-import org.apache.qpid.server.flow.MessageOnlyCreditManager;
-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.MessageSource;
-import org.apache.qpid.server.message.ServerMessage;
-import org.apache.qpid.server.model.Broker;
-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.State;
-import org.apache.qpid.server.model.UnknownConfiguredObjectException;
-import org.apache.qpid.server.model.port.AmqpPort;
-import org.apache.qpid.server.protocol.AMQSessionModel;
-import org.apache.qpid.server.queue.AMQQueue;
-import org.apache.qpid.server.queue.QueueArgumentsConverter;
-import org.apache.qpid.server.security.SubjectCreator;
-import org.apache.qpid.server.security.auth.SubjectAuthenticationResult;
-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;
-
-public class ServerMethodDispatcherImpl implements MethodDispatcher
-{
- private static final Logger _logger = Logger.getLogger(ServerMethodDispatcherImpl.class);
-
- private final AMQProtocolSession<?> _connection;
-
-
- public static MethodDispatcher createMethodDispatcher(AMQProtocolSession<?> connection)
- {
- return new ServerMethodDispatcherImpl(connection);
- }
-
-
- public ServerMethodDispatcherImpl(AMQProtocolSession<?> connection)
- {
- _connection = connection;
- }
-
-
- protected final AMQProtocolSession<?> getConnection()
- {
- return _connection;
- }
-
- public boolean dispatchAccessRequest(AccessRequestBody body, int channelId) throws AMQException
- {
- final AMQChannel channel = _connection.getChannel(channelId);
- if (channel == null)
- {
- throw body.getChannelNotFoundException(channelId, _connection.getMethodRegistry());
- }
-
- MethodRegistry methodRegistry = _connection.getMethodRegistry();
-
- if(ProtocolVersion.v0_91.equals(_connection.getProtocolVersion()) )
- {
- throw new AMQException(AMQConstant.COMMAND_INVALID, "AccessRequest not present in AMQP versions other than 0-8, 0-9");
- }
-
- // 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);
- channel.sync();
- _connection.writeFrame(response.generateFrame(channelId));
- return true;
- }
-
- public boolean dispatchBasicAck(BasicAckBody body, int channelId) throws AMQException
- {
-
- if (_logger.isDebugEnabled())
- {
- _logger.debug("Ack(Tag:" + body.getDeliveryTag() + ":Mult:" + body.getMultiple() + ") received on channel " + channelId);
- }
-
- final AMQChannel channel = _connection.getChannel(channelId);
-
- if (channel == null)
- {
- throw body.getChannelNotFoundException(channelId, _connection.getMethodRegistry());
- }
-
- // this method throws an AMQException if the delivery tag is not known
- channel.acknowledgeMessage(body.getDeliveryTag(), body.getMultiple());
- return true;
- }
-
- public boolean dispatchBasicCancel(BasicCancelBody body, int channelId) throws AMQException
- {
- final AMQChannel channel = _connection.getChannel(channelId);
-
-
- if (channel == null)
- {
- throw body.getChannelNotFoundException(channelId, _connection.getMethodRegistry());
- }
-
- if (_logger.isDebugEnabled())
- {
- _logger.debug("BasicCancel: for:" + body.getConsumerTag() +
- " nowait:" + body.getNowait());
- }
-
- channel.unsubscribeConsumer(body.getConsumerTag());
- if (!body.getNowait())
- {
- MethodRegistry methodRegistry = _connection.getMethodRegistry();
- BasicCancelOkBody cancelOkBody = methodRegistry.createBasicCancelOkBody(body.getConsumerTag());
- channel.sync();
- _connection.writeFrame(cancelOkBody.generateFrame(channelId));
- }
- return true;
- }
-
- public boolean dispatchBasicConsume(BasicConsumeBody body, int channelId) throws AMQException
- {
- AMQChannel channel = _connection.getChannel(channelId);
- VirtualHostImpl<?,?,?> vHost = _connection.getVirtualHost();
-
- if (channel == null)
- {
- throw body.getChannelNotFoundException(channelId, _connection.getMethodRegistry());
- }
- else
- {
- channel.sync();
- String queueName = body.getQueue() == null ? null : body.getQueue().asString();
- if (_logger.isDebugEnabled())
- {
- _logger.debug("BasicConsume: from '" + queueName +
- "' for:" + body.getConsumerTag() +
- " nowait:" + body.getNowait() +
- " args:" + body.getArguments());
- }
-
- MessageSource queue = queueName == null ? channel.getDefaultQueue() : vHost.getQueue(queueName);
- final Collection<MessageSource> sources = new HashSet<>();
- if(queue != null)
- {
- sources.add(queue);
- }
- else if(vHost.getContextValue(Boolean.class, "qpid.enableMultiQueueConsumers")
- && body.getArguments() != null
- && body.getArguments().get("x-multiqueue") instanceof Collection)
- {
- for(Object object : (Collection<Object>) body.getArguments().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 = body.getArguments().get("x-multiqueue").toString();
- }
-
- if (sources.isEmpty())
- {
- if (_logger.isDebugEnabled())
- {
- _logger.debug("No queue for '" + queueName + "'");
- }
- if (queueName != null)
- {
- String msg = "No such queue, '" + queueName + "'";
- throw body.getChannelException(AMQConstant.NOT_FOUND, msg, _connection.getMethodRegistry());
- }
- else
- {
- String msg = "No queue name provided, no default queue defined.";
- throw body.getConnectionException(AMQConstant.NOT_ALLOWED, msg, _connection.getMethodRegistry());
- }
- }
- else
- {
- final AMQShortString consumerTagName;
-
- if (body.getConsumerTag() != null)
- {
- consumerTagName = body.getConsumerTag().intern(false);
- }
- else
- {
- consumerTagName = null;
- }
-
- try
- {
- if(consumerTagName == null || channel.getSubscription(consumerTagName) == null)
- {
-
- AMQShortString consumerTag = channel.consumeFromSource(consumerTagName,
- sources,
- !body.getNoAck(),
- body.getArguments(),
- body.getExclusive(),
- body.getNoLocal());
- if (!body.getNowait())
- {
- MethodRegistry methodRegistry = _connection.getMethodRegistry();
- AMQMethodBody responseBody = methodRegistry.createBasicConsumeOkBody(consumerTag);
- _connection.writeFrame(responseBody.generateFrame(channelId));
-
- }
- }
- else
- {
- AMQShortString msg = AMQShortString.validValueOf("Non-unique consumer tag, '" + body.getConsumerTag() + "'");
-
- MethodRegistry methodRegistry = _connection.getMethodRegistry();
- AMQMethodBody responseBody = methodRegistry.createConnectionCloseBody(AMQConstant.NOT_ALLOWED.getCode(), // replyCode
- msg, // replytext
- body.getClazz(),
- body.getMethod());
- _connection.writeFrame(responseBody.generateFrame(0));
- }
-
- }
- catch (AMQInvalidArgumentException ise)
- {
- _logger.debug("Closing connection due to invalid selector");
-
- MethodRegistry methodRegistry = _connection.getMethodRegistry();
- AMQMethodBody responseBody = methodRegistry.createChannelCloseBody(AMQConstant.ARGUMENT_INVALID.getCode(),
- AMQShortString.validValueOf(ise.getMessage()),
- body.getClazz(),
- body.getMethod());
- _connection.writeFrame(responseBody.generateFrame(channelId));
-
-
- }
- catch (AMQQueue.ExistingExclusiveConsumer e)
- {
- throw body.getConnectionException(AMQConstant.ACCESS_REFUSED,
- "Cannot subscribe to queue "
- + queue.getName()
- + " as it already has an existing exclusive consumer",
- _connection.getMethodRegistry());
- }
- catch (AMQQueue.ExistingConsumerPreventsExclusive e)
- {
- throw body.getConnectionException(AMQConstant.ACCESS_REFUSED,
- "Cannot subscribe to queue "
- + queue.getName()
- + " exclusively as it already has a consumer",
- _connection.getMethodRegistry());
- }
- catch (AccessControlException e)
- {
- throw body.getConnectionException(AMQConstant.ACCESS_REFUSED,
- "Cannot subscribe to queue "
- + queue.getName()
- + " permission denied", _connection.getMethodRegistry());
- }
- catch (MessageSource.ConsumerAccessRefused consumerAccessRefused)
- {
- throw body.getConnectionException(AMQConstant.ACCESS_REFUSED,
- "Cannot subscribe to queue "
- + queue.getName()
- + " as it already has an incompatible exclusivity policy",
- _connection.getMethodRegistry());
- }
-
- }
- }
- return true;
- }
-
- public boolean dispatchBasicGet(BasicGetBody body, int channelId) throws AMQException
- {
-
- VirtualHostImpl vHost = _connection.getVirtualHost();
-
- AMQChannel channel = _connection.getChannel(channelId);
- if (channel == null)
- {
- throw body.getChannelNotFoundException(channelId, _connection.getMethodRegistry());
- }
- else
- {
- channel.sync();
- AMQQueue queue = body.getQueue() == null ? channel.getDefaultQueue() : vHost.getQueue(body.getQueue().toString());
- if (queue == null)
- {
- _logger.info("No queue for '" + body.getQueue() + "'");
- if(body.getQueue()!=null)
- {
- throw body.getConnectionException(AMQConstant.NOT_FOUND,
- "No such queue, '" + body.getQueue() + "'",
- _connection.getMethodRegistry());
- }
- else
- {
- throw body.getConnectionException(AMQConstant.NOT_ALLOWED,
- "No queue name provided, no default queue defined.",
- _connection.getMethodRegistry());
- }
- }
- else
- {
-
- try
- {
- if (!performGet(queue, _connection, channel, !body.getNoAck()))
- {
- MethodRegistry methodRegistry = _connection.getMethodRegistry();
-
- BasicGetEmptyBody responseBody = methodRegistry.createBasicGetEmptyBody(null);
-
-
- _connection.writeFrame(responseBody.generateFrame(channelId));
- }
- }
- catch (AccessControlException e)
- {
- throw body.getConnectionException(AMQConstant.ACCESS_REFUSED,
- e.getMessage(), _connection.getMethodRegistry());
- }
- catch (MessageSource.ExistingExclusiveConsumer e)
- {
- throw body.getConnectionException(AMQConstant.NOT_ALLOWED,
- "Queue has an exclusive consumer",
- _connection.getMethodRegistry());
- }
- catch (MessageSource.ExistingConsumerPreventsExclusive e)
- {
- throw body.getConnectionException(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",
- _connection.getMethodRegistry());
- }
- catch (MessageSource.ConsumerAccessRefused consumerAccessRefused)
- {
- throw body.getConnectionException(AMQConstant.NOT_ALLOWED,
- "Queue has an incompatible exclusivit policy",
- _connection.getMethodRegistry());
- }
- }
- }
- return true;
- }
-
- public static boolean performGet(final AMQQueue queue,
- final AMQProtocolSession session,
- final AMQChannel channel,
- final boolean acks)
- throws AMQException, MessageSource.ExistingConsumerPreventsExclusive,
- MessageSource.ExistingExclusiveConsumer, MessageSource.ConsumerAccessRefused
- {
-
- final FlowCreditManager singleMessageCredit = new MessageOnlyCreditManager(1L);
-
- final GetDeliveryMethod getDeliveryMethod =
- new GetDeliveryMethod(singleMessageCredit, session, channel, queue);
- final RecordDeliveryMethod getRecordMethod = new RecordDeliveryMethod()
- {
-
- public void recordMessageDelivery(final ConsumerImpl sub, final MessageInstance entry, final long deliveryTag)
- {
- channel.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(channel,
- AMQShortString.EMPTY_STRING, null,
- singleMessageCredit, getDeliveryMethod, getRecordMethod);
- }
- else
- {
- target = ConsumerTarget_0_8.createGetNoAckTarget(channel,
- AMQShortString.EMPTY_STRING, null,
- singleMessageCredit, getDeliveryMethod, getRecordMethod);
- }
-
- ConsumerImpl sub = queue.addConsumer(target, null, AMQMessage.class, "", options);
- sub.flush();
- sub.close();
- return(getDeliveryMethod.hasDeliveredMessage());
-
-
- }
-
-
- private static class GetDeliveryMethod implements ClientDeliveryMethod
- {
-
- private final FlowCreditManager _singleMessageCredit;
- private final AMQProtocolSession _session;
- private final AMQChannel _channel;
- private final AMQQueue _queue;
- private boolean _deliveredMessage;
-
- public GetDeliveryMethod(final FlowCreditManager singleMessageCredit,
- final AMQProtocolSession session,
- final AMQChannel channel, final AMQQueue queue)
- {
- _singleMessageCredit = singleMessageCredit;
- _session = session;
- _channel = channel;
- _queue = queue;
- }
-
- @Override
- public long deliverToClient(final ConsumerImpl sub, final ServerMessage message,
- final InstanceProperties props, final long deliveryTag)
- {
- _singleMessageCredit.useCreditForMessage(message.getSize());
- long size =_session.getProtocolOutputConverter().writeGetOk(message,
- props,
- _channel.getChannelId(),
- deliveryTag,
- _queue.getQueueDepthMessages());
-
- _deliveredMessage = true;
- return size;
- }
-
- public boolean hasDeliveredMessage()
- {
- return _deliveredMessage;
- }
- }
-
- public boolean dispatchBasicPublish(BasicPublishBody body, int channelId) throws AMQException
- {
- if (_logger.isDebugEnabled())
- {
- _logger.debug("Publish received on channel " + channelId);
- }
-
- AMQShortString exchangeName = body.getExchange();
- VirtualHostImpl vHost = _connection.getVirtualHost();
-
- // TODO: check the delivery tag field details - is it unique across the broker or per subscriber?
-
- MessageDestination destination;
-
- if (exchangeName == null || AMQShortString.EMPTY_STRING.equals(exchangeName))
- {
- destination = vHost.getDefaultDestination();
- }
- else
- {
- destination = vHost.getMessageDestination(exchangeName.toString());
- }
-
- // if the exchange does not exist we raise a channel exception
- if (destination == null)
- {
- throw body.getChannelException(AMQConstant.NOT_FOUND, "Unknown exchange name",
- _connection.getMethodRegistry());
- }
- else
- {
- // The partially populated BasicDeliver frame plus the received route body
- // is stored in the channel. Once the final body frame has been received
- // it is routed to the exchange.
- AMQChannel channel = _connection.getChannel(channelId);
-
- if (channel == null)
- {
- throw body.getChannelNotFoundException(channelId, _connection.getMethodRegistry());
- }
-
- MessagePublishInfo info = new MessagePublishInfo(body.getExchange(),
- body.getImmediate(),
- body.getMandatory(),
- body.getRoutingKey());
- info.setExchange(exchangeName);
- try
- {
- channel.setPublishFrame(info, destination);
- }
- catch (AccessControlException e)
- {
- throw body.getConnectionException(AMQConstant.ACCESS_REFUSED,
- e.getMessage(),
- _connection.getMethodRegistry());
- }
- }
- return true;
- }
-
- public boolean dispatchBasicQos(BasicQosBody body, int channelId) throws AMQException
- {
- AMQChannel channel = _connection.getChannel(channelId);
- if (channel == null)
- {
- throw body.getChannelNotFoundException(channelId, _connection.getMethodRegistry());
- }
- channel.sync();
- channel.setCredit(body.getPrefetchSize(), body.getPrefetchCount());
-
-
- MethodRegistry methodRegistry = _connection.getMethodRegistry();
- AMQMethodBody responseBody = methodRegistry.createBasicQosOkBody();
- _connection.writeFrame(responseBody.generateFrame(channelId));
-
- return true;
- }
-
- public boolean dispatchBasicRecover(BasicRecoverBody body, int channelId) throws AMQException
- {
- _logger.debug("Recover received on protocol session " + _connection
- + " and channel " + channelId);
- AMQChannel channel = _connection.getChannel(channelId);
-
-
- if (channel == null)
- {
- throw body.getChannelNotFoundException(channelId, _connection.getMethodRegistry());
- }
-
- channel.resend();
-
- // Qpid 0-8 hacks a synchronous -ok onto recover.
- // In Qpid 0-9 we create a separate sync-recover, sync-recover-ok pair to be "more" compliant
- if(_connection.getProtocolVersion().equals(ProtocolVersion.v8_0))
- {
- MethodRegistry methodRegistry = _connection.getMethodRegistry();
- AMQMethodBody recoverOk = methodRegistry.createBasicRecoverSyncOkBody();
- channel.sync();
- _connection.writeFrame(recoverOk.generateFrame(channelId));
-
- }
-
- return true;
- }
-
- public boolean dispatchBasicReject(BasicRejectBody body, int channelId) throws AMQException
- {
-
- AMQChannel channel = _connection.getChannel(channelId);
-
- if (channel == null)
- {
- throw body.getChannelNotFoundException(channelId, _connection.getMethodRegistry());
- }
-
- if (_logger.isDebugEnabled())
- {
- _logger.debug("Rejecting:" + body.getDeliveryTag() +
- ": Requeue:" + body.getRequeue() +
- " on channel:" + channel.debugIdentity());
- }
-
- long deliveryTag = body.getDeliveryTag();
-
- MessageInstance message = channel.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:" + body.getRequeue() +
- " on channel:" + channel.debugIdentity());
- }
-
- if (body.getRequeue())
- {
- //this requeue represents a message rejected from the pre-dispatch queue
- //therefore we need to amend the delivery counter.
- message.decrementDeliveryCount();
-
- channel.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 = channel.isMaxDeliveryCountEnabled(deliveryTag);
- _logger.debug("maxDeliveryCountEnabled: "
- + maxDeliveryCountEnabled
- + " deliveryTag "
- + deliveryTag);
- if (maxDeliveryCountEnabled)
- {
- final boolean deliveredTooManyTimes = channel.isDeliveredTooManyTimes(deliveryTag);
- _logger.debug("deliveredTooManyTimes: "
- + deliveredTooManyTimes
- + " deliveryTag "
- + deliveryTag);
- if (deliveredTooManyTimes)
- {
- channel.deadLetter(body.getDeliveryTag());
- }
- 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
- {
- channel.requeue(deliveryTag);
- }
- }
- }
- }
- return true;
- }
-
- public boolean dispatchChannelOpen(ChannelOpenBody body, int channelId) throws AMQException
- {
- VirtualHostImpl virtualHost = _connection.getVirtualHost();
-
- // Protect the broker against out of order frame request.
- if (virtualHost == null)
- {
- throw new AMQException(AMQConstant.COMMAND_INVALID, "Virtualhost has not yet been set. ConnectionOpen has not been called.", null);
- }
- _logger.info("Connecting to: " + virtualHost.getName());
-
- final AMQChannel channel = new AMQChannel(_connection, channelId, virtualHost.getMessageStore());
-
- _connection.addChannel(channel);
-
- ChannelOpenOkBody response;
-
-
- response = _connection.getMethodRegistry().createChannelOpenOkBody();
-
-
- _connection.writeFrame(response.generateFrame(channelId));
- return true;
- }
-
-
- public boolean dispatchAccessRequestOk(AccessRequestOkBody body, int channelId) throws AMQException
- {
- throw new UnexpectedMethodException(body);
- }
-
-
- public boolean dispatchBasicCancelOk(BasicCancelOkBody body, int channelId) throws AMQException
- {
- throw new UnexpectedMethodException(body);
- }
-
- public boolean dispatchBasicConsumeOk(BasicConsumeOkBody body, int channelId) throws AMQException
- {
- throw new UnexpectedMethodException(body);
- }
-
- public boolean dispatchBasicDeliver(BasicDeliverBody body, int channelId) throws AMQException
- {
- throw new UnexpectedMethodException(body);
- }
-
- public boolean dispatchBasicGetEmpty(BasicGetEmptyBody body, int channelId) throws AMQException
- {
- throw new UnexpectedMethodException(body);
- }
-
- public boolean dispatchBasicGetOk(BasicGetOkBody body, int channelId) throws AMQException
- {
- throw new UnexpectedMethodException(body);
- }
-
- public boolean dispatchBasicQosOk(BasicQosOkBody body, int channelId) throws AMQException
- {
- throw new UnexpectedMethodException(body);
- }
-
- public boolean dispatchBasicReturn(BasicReturnBody body, int channelId) throws AMQException
- {
- throw new UnexpectedMethodException(body);
- }
-
- public boolean dispatchChannelClose(ChannelCloseBody body, int channelId) throws AMQException
- {
-
- if (_logger.isInfoEnabled())
- {
- _logger.info("Received channel close for id " + channelId
- + " citing class " + body.getClassId() +
- " and method " + body.getMethodId());
- }
-
-
- AMQChannel channel = _connection.getChannel(channelId);
-
- if (channel == null)
- {
- throw body.getConnectionException(AMQConstant.CHANNEL_ERROR,
- "Trying to close unknown channel",
- _connection.getMethodRegistry());
- }
- channel.sync();
- _connection.closeChannel(channelId);
- // Client requested closure so we don't wait for ok we send it
- _connection.closeChannelOk(channelId);
-
- MethodRegistry methodRegistry = _connection.getMethodRegistry();
- ChannelCloseOkBody responseBody = methodRegistry.createChannelCloseOkBody();
- _connection.writeFrame(responseBody.generateFrame(channelId));
- return true;
- }
-
-
- public boolean dispatchChannelCloseOk(ChannelCloseOkBody body, int channelId) throws AMQException
- {
-
- _logger.info("Received channel-close-ok for channel-id " + channelId);
-
- // Let the Protocol Session know the channel is now closed.
- _connection.closeChannelOk(channelId);
- return true;
- }
-
-
- public boolean dispatchChannelFlow(ChannelFlowBody body, int channelId) throws AMQException
- {
- final AMQProtocolSession<?> connection = getConnection();
-
-
- AMQChannel channel = connection.getChannel(channelId);
-
- if (channel == null)
- {
- throw body.getChannelNotFoundException(channelId, connection.getMethodRegistry());
- }
- channel.sync();
- channel.setSuspended(!body.getActive());
- _logger.debug("Channel.Flow for channel " + channelId + ", active=" + body.getActive());
-
- MethodRegistry methodRegistry = connection.getMethodRegistry();
- AMQMethodBody responseBody = methodRegistry.createChannelFlowOkBody(body.getActive());
- connection.writeFrame(responseBody.generateFrame(channelId));
- return true;
- }
-
- public boolean dispatchChannelFlowOk(ChannelFlowOkBody body, int channelId) throws AMQException
- {
- throw new UnexpectedMethodException(body);
- }
-
- public boolean dispatchChannelOpenOk(ChannelOpenOkBody body, int channelId) throws AMQException
- {
- throw new UnexpectedMethodException(body);
- }
-
-
- public boolean dispatchConnectionOpen(ConnectionOpenBody body, int channelId) throws AMQException
- {
-
- //ignore leading '/'
- String virtualHostName;
- if ((body.getVirtualHost() != null) && body.getVirtualHost().charAt(0) == '/')
- {
- virtualHostName = new StringBuilder(body.getVirtualHost().subSequence(1, body.getVirtualHost().length())).toString();
- }
- else
- {
- virtualHostName = body.getVirtualHost() == null ? null : String.valueOf(body.getVirtualHost());
- }
-
- VirtualHostImpl virtualHost = ((AmqpPort) _connection.getPort()).getVirtualHost(virtualHostName);
-
- if (virtualHost == null)
- {
- throw body.getConnectionException(AMQConstant.NOT_FOUND, "Unknown virtual host: '" + virtualHostName + "'",
- _connection.getMethodRegistry());
- }
- else
- {
- // Check virtualhost access
- if (virtualHost.getState() != State.ACTIVE)
- {
- throw body.getConnectionException(AMQConstant.CONNECTION_FORCED,
- "Virtual host '" + virtualHost.getName() + "' is not active",
- _connection.getMethodRegistry());
- }
-
- _connection.setVirtualHost(virtualHost);
- try
- {
- virtualHost.getSecurityManager().authoriseCreateConnection(_connection);
- }
- catch (AccessControlException e)
- {
- throw body.getConnectionException(AMQConstant.ACCESS_REFUSED,
- e.getMessage(),
- _connection.getMethodRegistry());
- }
-
- // See Spec (0.8.2). Section 3.1.2 Virtual Hosts
- if (_connection.getContextKey() == null)
- {
- _connection.setContextKey(new AMQShortString(Long.toString(System.currentTimeMillis())));
- }
-
- MethodRegistry methodRegistry = _connection.getMethodRegistry();
- AMQMethodBody responseBody = methodRegistry.createConnectionOpenOkBody(body.getVirtualHost());
-
- _connection.writeFrame(responseBody.generateFrame(channelId));
- }
- return true;
- }
-
-
- public boolean dispatchConnectionClose(ConnectionCloseBody body, int channelId) throws AMQException
- {
- if (_logger.isInfoEnabled())
- {
- _logger.info("ConnectionClose received with reply code/reply text " + body.getReplyCode() + "/" +
- body.getReplyText() + " for " + _connection);
- }
- try
- {
- _connection.closeSession();
- }
- catch (Exception e)
- {
- _logger.error("Error closing protocol session: " + e, e);
- }
-
- MethodRegistry methodRegistry = _connection.getMethodRegistry();
- ConnectionCloseOkBody responseBody = methodRegistry.createConnectionCloseOkBody();
- _connection.writeFrame(responseBody.generateFrame(channelId));
-
- _connection.closeProtocolSession();
-
- return true;
- }
-
-
- public boolean dispatchConnectionCloseOk(ConnectionCloseOkBody body, int channelId) throws AMQException
- {
- _logger.info("Received Connection-close-ok");
-
- try
- {
- _connection.closeSession();
- }
- catch (Exception e)
- {
- _logger.error("Error closing protocol session: " + e, e);
- }
- return true;
- }
-
- public boolean dispatchConnectionOpenOk(ConnectionOpenOkBody body, int channelId) throws AMQException
- {
- throw new UnexpectedMethodException(body);
- }
-
- public boolean dispatchConnectionRedirect(ConnectionRedirectBody body, int channelId) throws AMQException
- {
- throw new UnexpectedMethodException(body);
- }
-
- public boolean dispatchConnectionSecure(ConnectionSecureBody body, int channelId) throws AMQException
- {
- throw new UnexpectedMethodException(body);
- }
-
- public boolean dispatchConnectionStart(ConnectionStartBody body, int channelId) throws AMQException
- {
- throw new UnexpectedMethodException(body);
- }
-
- public boolean dispatchConnectionTune(ConnectionTuneBody body, int channelId) throws AMQException
- {
- throw new UnexpectedMethodException(body);
- }
-
-
- public boolean dispatchExchangeBoundOk(ExchangeBoundOkBody body, int channelId) throws AMQException
- {
- throw new UnexpectedMethodException(body);
- }
-
- public boolean dispatchExchangeDeclareOk(ExchangeDeclareOkBody body, int channelId) throws AMQException
- {
- throw new UnexpectedMethodException(body);
- }
-
- public boolean dispatchExchangeDeleteOk(ExchangeDeleteOkBody body, int channelId) throws AMQException
- {
- throw new UnexpectedMethodException(body);
- }
-
- public boolean dispatchQueueBindOk(QueueBindOkBody body, int channelId) throws AMQException
- {
- throw new UnexpectedMethodException(body);
- }
-
- public boolean dispatchQueueDeclareOk(QueueDeclareOkBody body, int channelId) throws AMQException
- {
- throw new UnexpectedMethodException(body);
- }
-
- public boolean dispatchQueueDeleteOk(QueueDeleteOkBody body, int channelId) throws AMQException
- {
- throw new UnexpectedMethodException(body);
- }
-
- public boolean dispatchQueuePurgeOk(QueuePurgeOkBody body, int channelId) throws AMQException
- {
- throw new UnexpectedMethodException(body);
- }
-
- public boolean dispatchTxCommitOk(TxCommitOkBody body, int channelId) throws AMQException
- {
- throw new UnexpectedMethodException(body);
- }
-
- public boolean dispatchTxRollbackOk(TxRollbackOkBody body, int channelId) throws AMQException
- {
- throw new UnexpectedMethodException(body);
- }
-
- public boolean dispatchTxSelectOk(TxSelectOkBody body, int channelId) throws AMQException
- {
- throw new UnexpectedMethodException(body);
- }
-
-
- public boolean dispatchConnectionSecureOk(ConnectionSecureOkBody body, int channelId) throws AMQException
- {
- Broker<?> broker = _connection.getBroker();
-
- SubjectCreator subjectCreator = _connection.getSubjectCreator();
-
- SaslServer ss = _connection.getSaslServer();
- if (ss == null)
- {
- throw new AMQException("No SASL context set up in session");
- }
- MethodRegistry methodRegistry = _connection.getMethodRegistry();
- SubjectAuthenticationResult authResult = subjectCreator.authenticate(ss, body.getResponse());
- switch (authResult.getStatus())
- {
- case ERROR:
- Exception cause = authResult.getCause();
-
- _logger.info("Authentication failed:" + (cause == null ? "" : cause.getMessage()));
-
- ConnectionCloseBody connectionCloseBody =
- methodRegistry.createConnectionCloseBody(AMQConstant.NOT_ALLOWED.getCode(),
- AMQConstant.NOT_ALLOWED.getName(),
- body.getClazz(),
- body.getMethod());
-
- _connection.writeFrame(connectionCloseBody.generateFrame(0));
- disposeSaslServer(_connection);
- 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());
- _connection.writeFrame(tuneBody.generateFrame(0));
- _connection.setAuthorizedSubject(authResult.getSubject());
- disposeSaslServer(_connection);
- break;
- case CONTINUE:
-
- ConnectionSecureBody
- secureBody = methodRegistry.createConnectionSecureBody(authResult.getChallenge());
- _connection.writeFrame(secureBody.generateFrame(0));
- }
- return true;
- }
-
- private void disposeSaslServer(AMQProtocolSession ps)
- {
- SaslServer ss = ps.getSaslServer();
- if (ss != null)
- {
- ps.setSaslServer(null);
- try
- {
- ss.dispose();
- }
- catch (SaslException e)
- {
- _logger.error("Error disposing of Sasl server: " + e);
- }
- }
- }
-
- public boolean dispatchConnectionStartOk(ConnectionStartOkBody body, int channelId) throws AMQException
- {
- Broker<?> broker = _connection.getBroker();
-
- _logger.info("SASL Mechanism selected: " + body.getMechanism());
- _logger.info("Locale selected: " + body.getLocale());
-
- SubjectCreator subjectCreator = _connection.getSubjectCreator();
- SaslServer ss = null;
- try
- {
- ss = subjectCreator.createSaslServer(String.valueOf(body.getMechanism()),
- _connection.getLocalFQDN(),
- _connection.getPeerPrincipal());
-
- if (ss == null)
- {
- throw body.getConnectionException(AMQConstant.RESOURCE_ERROR,
- "Unable to create SASL Server:" + body.getMechanism(),
- _connection.getMethodRegistry());
- }
-
- _connection.setSaslServer(ss);
-
- final SubjectAuthenticationResult authResult = subjectCreator.authenticate(ss, body.getResponse());
- //save clientProperties
- _connection.setClientProperties(body.getClientProperties());
-
- MethodRegistry methodRegistry = _connection.getMethodRegistry();
-
- switch (authResult.getStatus())
- {
- case ERROR:
- Exception cause = authResult.getCause();
-
- _logger.info("Authentication failed:" + (cause == null ? "" : cause.getMessage()));
-
- ConnectionCloseBody closeBody =
- methodRegistry.createConnectionCloseBody(AMQConstant.NOT_ALLOWED.getCode(), // replyCode
- AMQConstant.NOT_ALLOWED.getName(),
- body.getClazz(),
- body.getMethod());
-
- _connection.writeFrame(closeBody.generateFrame(0));
- disposeSaslServer(_connection);
- break;
-
- case SUCCESS:
- if (_logger.isInfoEnabled())
- {
- _logger.info("Connected as: " + authResult.getSubject());
- }
- _connection.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());
- _connection.writeFrame(tuneBody.generateFrame(0));
- break;
- case CONTINUE:
- ConnectionSecureBody
- secureBody = methodRegistry.createConnectionSecureBody(authResult.getChallenge());
- _connection.writeFrame(secureBody.generateFrame(0));
- }
- }
- catch (SaslException e)
- {
- disposeSaslServer(_connection);
- throw new AMQException("SASL error: " + e, e);
- }
- return true;
- }
-
- public boolean dispatchConnectionTuneOk(ConnectionTuneOkBody body, int channelId) throws AMQException
- {
- final AMQProtocolSession<?> connection = getConnection();
-
- if (_logger.isDebugEnabled())
- {
- _logger.debug(body);
- }
-
- connection.initHeartbeats(body.getHeartbeat());
-
- int brokerFrameMax = connection.getBroker().getContextValue(Integer.class, Broker.BROKER_FRAME_SIZE);
- if(brokerFrameMax <= 0)
- {
- brokerFrameMax = Integer.MAX_VALUE;
- }
-
- if(body.getFrameMax() > (long) brokerFrameMax)
- {
- throw new AMQConnectionException(AMQConstant.SYNTAX_ERROR,
- "Attempt to set max frame size to " + body.getFrameMax()
- + " greater than the broker will allow: "
- + brokerFrameMax,
- body.getClazz(), body.getMethod(),
- connection.getMethodRegistry(),null);
- }
- else if(body.getFrameMax() > 0 && body.getFrameMax() < AMQConstant.FRAME_MIN_SIZE.getCode())
- {
- throw new AMQConnectionException(AMQConstant.SYNTAX_ERROR,
- "Attempt to set max frame size to " + body.getFrameMax()
- + " which is smaller than the specification definined minimum: "
- + AMQConstant.FRAME_MIN_SIZE.getCode(),
- body.getClazz(), body.getMethod(),
- connection.getMethodRegistry(),null);
- }
- int frameMax = body.getFrameMax() == 0 ? brokerFrameMax : (int) body.getFrameMax();
- connection.setMaxFrameSize(frameMax);
-
- long maxChannelNumber = body.getChannelMax();
- //0 means no implied limit, except that forced by protocol limitations (0xFFFF)
- connection.setMaximumNumberOfChannels(maxChannelNumber == 0 ? 0xFFFFL : maxChannelNumber);
- return true;
- }
-
- public static final int OK = 0;
- public static final int EXCHANGE_NOT_FOUND = 1;
- public static final int QUEUE_NOT_FOUND = 2;
- public static final int NO_BINDINGS = 3;
- public static final int QUEUE_NOT_BOUND = 4;
- public static final int NO_QUEUE_BOUND_WITH_RK = 5;
- public static final int SPECIFIC_QUEUE_NOT_BOUND_WITH_RK = 6;
-
- public boolean dispatchExchangeBound(ExchangeBoundBody body, int channelId) throws AMQException
- {
- VirtualHostImpl virtualHost = _connection.getVirtualHost();
- MethodRegistry methodRegistry = _connection.getMethodRegistry();
-
- final AMQChannel channel = _connection.getChannel(channelId);
- if (channel == null)
- {
- throw body.getChannelNotFoundException(channelId, _connection.getMethodRegistry());
- }
- channel.sync();
-
-
- AMQShortString exchangeName = body.getExchange();
- AMQShortString queueName = body.getQueue();
- AMQShortString routingKey = body.getRoutingKey();
- ExchangeBoundOkBody response;
-
- if(isDefaultExchange(exchangeName))
- {
- if(routingKey == null)
- {
- if(queueName == null)
- {
- response = methodRegistry.createExchangeBoundOkBody(virtualHost.getQueues().isEmpty() ? NO_BINDINGS : OK, null);
- }
- else
- {
- AMQQueue queue = virtualHost.getQueue(queueName.toString());
- if (queue == null)
- {
-
- response = methodRegistry.createExchangeBoundOkBody(QUEUE_NOT_FOUND, // replyCode
- AMQShortString.validValueOf("Queue '" + queueName + "' not found")); // replyText
- }
- else
- {
- response = methodRegistry.createExchangeBoundOkBody(OK, null);
- }
- }
- }
- else
- {
- if(queueName == null)
- {
- response = methodRegistry.createExchangeBoundOkBody(virtualHost.getQueue(routingKey.toString()) == null ? NO_QUEUE_BOUND_WITH_RK : OK, null);
- }
- else
- {
- AMQQueue queue = virtualHost.getQueue(queueName.toString());
- if (queue == null)
- {
-
- response = methodRegistry.createExchangeBoundOkBody(QUEUE_NOT_FOUND, // replyCode
- AMQShortString.validValueOf("Queue '" + queueName + "' not found")); // replyText
- }
- else
- {
- response = methodRegistry.createExchangeBoundOkBody(queueName.equals(routingKey) ? OK : SPECIFIC_QUEUE_NOT_BOUND_WITH_RK, null);
- }
- }
- }
- }
- else
- {
- ExchangeImpl exchange = virtualHost.getExchange(exchangeName.toString());
- if (exchange == null)
- {
-
-
- response = methodRegistry.createExchangeBoundOkBody(EXCHANGE_NOT_FOUND,
- AMQShortString.validValueOf("Exchange '" + exchangeName + "' not found"));
- }
- else if (routingKey == null)
- {
- if (queueName == null)
- {
- if (exchange.hasBindings())
- {
- response = methodRegistry.createExchangeBoundOkBody(OK, null);
- }
- else
- {
-
- response = methodRegistry.createExchangeBoundOkBody(NO_BINDINGS, // replyCode
- null); // replyText
- }
- }
- else
- {
-
- AMQQueue queue = virtualHost.getQueue(queueName.toString());
- if (queue == null)
- {
-
- response = methodRegistry.createExchangeBoundOkBody(QUEUE_NOT_FOUND, // replyCode
- AMQShortString.validValueOf("Queue '" + queueName + "' not found")); // replyText
- }
- else
- {
- if (exchange.isBound(queue))
- {
-
- response = methodRegistry.createExchangeBoundOkBody(OK, // replyCode
- null); // replyText
- }
- else
- {
-
- response = methodRegistry.createExchangeBoundOkBody(QUEUE_NOT_BOUND, // replyCode
- AMQShortString.validValueOf("Queue '" + queueName + "' not bound to exchange '" + exchangeName + "'")); // replyText
- }
- }
- }
- }
- else if (queueName != null)
- {
- AMQQueue queue = virtualHost.getQueue(queueName.toString());
- if (queue == null)
- {
-
- response = methodRegistry.createExchangeBoundOkBody(QUEUE_NOT_FOUND, // replyCode
- AMQShortString.validValueOf("Queue '" + queueName + "' not found")); // replyText
- }
- else
- {
- String bindingKey = body.getRoutingKey() == null ? null : body.getRoutingKey().asString();
- if (exchange.isBound(bindingKey, queue))
- {
-
- response = methodRegistry.createExchangeBoundOkBody(OK, // replyCode
- null); // replyText
- }
- else
- {
-
- String message = "Queue '" + queueName + "' not bound with routing key '" +
- body.getRoutingKey() + "' to exchange '" + exchangeName + "'";
-
- response = methodRegistry.createExchangeBoundOkBody(SPECIFIC_QUEUE_NOT_BOUND_WITH_RK, // replyCode
- AMQShortString.validValueOf(message)); // replyText
- }
- }
- }
- else
- {
- if (exchange.isBound(body.getRoutingKey() == null ? "" : body.getRoutingKey().asString()))
- {
-
- response = methodRegistry.createExchangeBoundOkBody(OK, // replyCode
- null); // replyText
- }
- else
- {
-
- response = methodRegistry.createExchangeBoundOkBody(NO_QUEUE_BOUND_WITH_RK, // replyCode
- AMQShortString.validValueOf("No queue bound with routing key '" + body.getRoutingKey() +
- "' to exchange '" + exchangeName + "'")); // replyText
- }
- }
- }
- _connection.writeFrame(response.generateFrame(channelId));
- return true;
- }
-
- public boolean dispatchExchangeDeclare(ExchangeDeclareBody body, int channelId) throws AMQException
- {
- VirtualHostImpl virtualHost = _connection.getVirtualHost();
- final AMQChannel channel = _connection.getChannel(channelId);
- if (channel == null)
- {
- throw body.getChannelNotFoundException(channelId, _connection.getMethodRegistry());
- }
-
- final AMQShortString exchangeName = body.getExchange();
- if (_logger.isDebugEnabled())
- {
- _logger.debug("Request to declare exchange of type " + body.getType() + " with name " + exchangeName);
- }
-
- ExchangeImpl exchange;
-
- if(isDefaultExchange(exchangeName))
- {
- if(!new AMQShortString(ExchangeDefaults.DIRECT_EXCHANGE_CLASS).equals(body.getType()))
- {
- throw new AMQConnectionException(AMQConstant.NOT_ALLOWED, "Attempt to redeclare default exchange: "
- + " of type "
- + ExchangeDefaults.DIRECT_EXCHANGE_CLASS
- + " to " + body.getType() +".",
- body.getClazz(), body.getMethod(),
- _connection.getMethodRegistry(),null);
- }
- }
- else
- {
- if (body.getPassive())
- {
- exchange = virtualHost.getExchange(exchangeName.toString());
- if(exchange == null)
- {
- throw body.getChannelException(AMQConstant.NOT_FOUND, "Unknown exchange: " + exchangeName,
- _connection.getMethodRegistry());
- }
- else if (!(body.getType() == null || body.getType().length() ==0) && !exchange.getType().equals(body.getType().asString()))
- {
-
- throw new AMQConnectionException(AMQConstant.NOT_ALLOWED, "Attempt to redeclare exchange: " +
- exchangeName + " of type " + exchange.getType()
- + " to " + body.getType() +".",
- body.getClazz(), body.getMethod(),
- _connection.getMethodRegistry(),null);
- }
-
- }
- else
- {
- try
- {
- String name = exchangeName == null ? null : exchangeName.intern().toString();
- String type = body.getType() == null ? null : body.getType().intern().toString();
-
- Map<String,Object> attributes = new HashMap<String, Object>();
- if(body.getArguments() != null)
- {
- attributes.putAll(FieldTable.convertToMap(body.getArguments()));
- }
- attributes.put(org.apache.qpid.server.model.Exchange.ID, null);
- attributes.put(org.apache.qpid.server.model.Exchange.NAME,name);
- attributes.put(org.apache.qpid.server.model.Exchange.TYPE,type);
- attributes.put(org.apache.qpid.server.model.Exchange.DURABLE, body.getDurable());
- attributes.put(org.apache.qpid.server.model.Exchange.LIFETIME_POLICY,
- body.getAutoDelete() ? LifetimePolicy.DELETE_ON_NO_LINKS : LifetimePolicy.PERMANENT);
- if(!attributes.containsKey(org.apache.qpid.server.model.Exchange.ALTERNATE_EXCHANGE))
- {
- attributes.put(org.apache.qpid.server.model.Exchange.ALTERNATE_EXCHANGE, null);
- }
- exchange = virtualHost.createExchange(attributes);
-
- }
- catch(ReservedExchangeNameException e)
- {
- throw body.getConnectionException(AMQConstant.NOT_ALLOWED,
- "Attempt to declare exchange: " + exchangeName +
- " which begins with reserved prefix.",
- _connection.getMethodRegistry());
-
- }
- catch(ExchangeExistsException e)
- {
- exchange = e.getExistingExchange();
- if(!new AMQShortString(exchange.getType()).equals(body.getType()))
- {
- throw body.getConnectionException(AMQConstant.NOT_ALLOWED, "Attempt to redeclare exchange: "
- + exchangeName + " of type "
- + exchange.getType()
- + " to " + body.getType() + ".",
- _connection.getMethodRegistry());
- }
- }
- catch(NoFactoryForTypeException e)
- {
- throw body.getConnectionException(AMQConstant.COMMAND_INVALID,
- "Unknown exchange type '"
- + e.getType()
- + "' for exchange '"
- + exchangeName
- + "'",
- _connection.getMethodRegistry());
- }
- catch (AccessControlException e)
- {
- throw body.getConnectionException(AMQConstant.ACCESS_REFUSED,
- e.getMessage(),
- _connection.getMethodRegistry());
- }
- catch (UnknownConfiguredObjectException e)
- {
- // note - since 0-8/9/9-1 can't set the alt. exchange this exception should never occur
- throw body.getConnectionException(AMQConstant.NOT_FOUND,
- "Unknown alternate exchange "
- + (e.getName() != null
- ? "name: \"" + e.getName() + "\""
- : "id: " + e.getId()),
- _connection.getMethodRegistry());
- }
- catch (IllegalArgumentException e)
- {
- throw body.getConnectionException(AMQConstant.COMMAND_INVALID,
- "Error creating exchange '"
- + exchangeName
- + "': "
- + e.getMessage(),
- _connection.getMethodRegistry());
- }
- }
- }
-
- if(!body.getNowait())
- {
- MethodRegistry methodRegistry = _connection.getMethodRegistry();
- AMQMethodBody responseBody = methodRegistry.createExchangeDeclareOkBody();
- channel.sync();
- _connection.writeFrame(responseBody.generateFrame(channelId));
- }
- return true;
- }
-
- public boolean dispatchExchangeDelete(ExchangeDeleteBody body, int channelId) throws AMQException
- {
- VirtualHostImpl virtualHost = _connection.getVirtualHost();
- final AMQChannel channel = _connection.getChannel(channelId);
- if (channel == null)
- {
- throw body.getChannelNotFoundException(channelId, _connection.getMethodRegistry());
- }
- channel.sync();
- try
- {
-
- if(isDefaultExchange(body.getExchange()))
- {
- throw body.getConnectionException(AMQConstant.NOT_ALLOWED,
- "Default Exchange cannot be deleted",
- _connection.getMethodRegistry());
- }
-
- final String exchangeName = body.getExchange().toString();
-
- final ExchangeImpl exchange = virtualHost.getExchange(exchangeName);
- if(exchange == null)
- {
- throw body.getChannelException(AMQConstant.NOT_FOUND, "No such exchange: " + body.getExchange(),
- _connection.getMethodRegistry());
- }
-
- virtualHost.removeExchange(exchange, !body.getIfUnused());
-
- ExchangeDeleteOkBody responseBody = _connection.getMethodRegistry().createExchangeDeleteOkBody();
-
- _connection.writeFrame(responseBody.generateFrame(channelId));
- }
-
- catch (ExchangeIsAlternateException e)
- {
- throw body.getChannelException(AMQConstant.NOT_ALLOWED, "Exchange in use as an alternate exchange",
- _connection.getMethodRegistry());
-
- }
- catch (RequiredExchangeException e)
- {
- throw body.getChannelException(AMQConstant.NOT_ALLOWED,
- "Exchange '" + body.getExchange() + "' cannot be deleted",
- _connection.getMethodRegistry());
- }
- catch (AccessControlException e)
- {
- throw body.getConnectionException(AMQConstant.ACCESS_REFUSED,
- e.getMessage(),
- _connection.getMethodRegistry());
- }
- return true;
- }
-
- private boolean isDefaultExchange(final AMQShortString exchangeName)
- {
- return exchangeName == null || exchangeName.equals(AMQShortString.EMPTY_STRING);
- }
-
- public boolean dispatchQueueBind(QueueBindBody body, int channelId) throws AMQException
- {
- VirtualHostImpl virtualHost = _connection.getVirtualHost();
- AMQChannel channel = _connection.getChannel(channelId);
-
- if (channel == null)
- {
- throw body.getChannelNotFoundException(channelId, _connection.getMethodRegistry());
- }
-
- final AMQQueue queue;
- final AMQShortString routingKey;
-
- final AMQShortString queueName = body.getQueue();
-
- if (queueName == null)
- {
-
- queue = channel.getDefaultQueue();
-
- if (queue == null)
- {
- throw body.getChannelException(AMQConstant.NOT_FOUND,
- "No default queue defined on channel and queue was null",
- _connection.getMethodRegistry());
- }
-
- if (body.getRoutingKey() == null)
- {
- routingKey = AMQShortString.valueOf(queue.getName());
- }
- else
- {
- routingKey = body.getRoutingKey().intern();
- }
- }
- else
- {
- queue = virtualHost.getQueue(queueName.toString());
- routingKey = body.getRoutingKey() == null ? AMQShortString.EMPTY_STRING : body.getRoutingKey().intern();
- }
-
- if (queue == null)
- {
- throw body.getChannelException(AMQConstant.NOT_FOUND, "Queue " + queueName + " does not exist.",
- _connection.getMethodRegistry());
- }
-
- if(isDefaultExchange(body.getExchange()))
- {
- throw body.getConnectionException(AMQConstant.NOT_ALLOWED,
- "Cannot bind the queue " + queueName + " to the default exchange",
- _connection.getMethodRegistry());
- }
-
- final String exchangeName = body.getExchange().toString();
-
- final ExchangeImpl exch = virtualHost.getExchange(exchangeName);
- if (exch == null)
- {
- throw body.getChannelException(AMQConstant.NOT_FOUND, "Exchange " + exchangeName + " does not exist.",
- _connection.getMethodRegistry());
- }
-
-
- try
- {
-
- Map<String,Object> arguments = FieldTable.convertToMap(body.getArguments());
- 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);
- }
- }
- }
- catch (AccessControlException e)
- {
- throw body.getConnectionException(AMQConstant.ACCESS_REFUSED,
- e.getMessage(),
- _connection.getMethodRegistry());
- }
-
- if (_logger.isInfoEnabled())
- {
- _logger.info("Binding queue " + queue + " to exchange " + exch + " with routing key " + routingKey);
- }
- if (!body.getNowait())
- {
- channel.sync();
- MethodRegistry methodRegistry = _connection.getMethodRegistry();
- AMQMethodBody responseBody = methodRegistry.createQueueBindOkBody();
- _connection.writeFrame(responseBody.generateFrame(channelId));
-
- }
- return true;
- }
-
- public boolean dispatchQueueDeclare(QueueDeclareBody body, int channelId) throws AMQException
- {
- final AMQSessionModel session = _connection.getChannel(channelId);
- 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 ((body.getQueue() == null) || (body.getQueue().length() == 0))
- {
- queueName = new AMQShortString("tmp_" + UUID.randomUUID());
- }
- else
- {
- queueName = body.getQueue().intern();
- }
-
- AMQQueue queue;
-
- //TODO: do we need to check that the queue already exists with exactly the same "configuration"?
-
- AMQChannel channel = _connection.getChannel(channelId);
-
- if (channel == null)
- {
- throw body.getChannelNotFoundException(channelId, _connection.getMethodRegistry());
- }
-
- if(body.getPassive())
- {
- queue = virtualHost.getQueue(queueName.toString());
- if (queue == null)
- {
- String msg = "Queue: " + queueName + " not found on VirtualHost(" + virtualHost + ").";
- throw body.getChannelException(AMQConstant.NOT_FOUND, msg, _connection.getMethodRegistry());
- }
- else
- {
- if (!queue.verifySessionAccess(channel))
- {
- throw body.getConnectionException(AMQConstant.NOT_ALLOWED,
- "Queue "
- + queue.getName()
- + " is exclusive, but not created on this Connection.",
- _connection.getMethodRegistry());
- }
-
- //set this as the default queue on the channel:
- channel.setDefaultQueue(queue);
- }
- }
- else
- {
-
- try
- {
-
- queue = createQueue(channel, queueName, body, virtualHost, _connection);
-
- }
- catch(QueueExistsException qe)
- {
-
- queue = qe.getExistingQueue();
-
- if (!queue.verifySessionAccess(channel))
- {
- throw body.getConnectionException(AMQConstant.NOT_ALLOWED,
- "Queue "
- + queue.getName()
- + " is exclusive, but not created on this Connection.",
- _connection.getMethodRegistry());
- }
- else if(queue.isExclusive() != body.getExclusive())
- {
-
- throw body.getChannelException(AMQConstant.ALREADY_EXISTS,
- "Cannot re-declare queue '"
- + queue.getName()
- + "' with different exclusivity (was: "
- + queue.isExclusive()
- + " requested "
- + body.getExclusive()
- + ")",
- _connection.getMethodRegistry());
- }
- else if((body.getAutoDelete() && queue.getLifetimePolicy() != LifetimePolicy.DELETE_ON_NO_OUTBOUND_LINKS)
- || (!body.getAutoDelete() && queue.getLifetimePolicy() != ((body.getExclusive() && !body.getDurable()) ? LifetimePolicy.DELETE_ON_CONNECTION_CLOSE : LifetimePolicy.PERMANENT)))
- {
- throw body.getChannelException(AMQConstant.ALREADY_EXISTS,
- "Cannot re-declare queue '"
- + queue.getName()
- + "' with different lifetime policy (was: "
- + queue.getLifetimePolicy()
- + " requested autodelete: "
- + body.getAutoDelete()
- + ")",
- _connection.getMethodRegistry());
- }
- else if(queue.isDurable() != body.getDurable())
- {
- throw body.getChannelException(AMQConstant.ALREADY_EXISTS,
- "Cannot re-declare queue '"
- + queue.getName()
- + "' with different durability (was: "
- + queue.isDurable()
- + " requested "
- + body.getDurable()
- + ")",
- _connection.getMethodRegistry());
- }
-
- }
- catch (AccessControlException e)
- {
- throw body.getConnectionException(AMQConstant.ACCESS_REFUSED,
- e.getMessage(),
- _connection.getMethodRegistry());
- }
-
- //set this as the default queue on the channel:
- channel.setDefaultQueue(queue);
- }
-
- if (!body.getNowait())
- {
- channel.sync();
- MethodRegistry methodRegistry = _connection.getMethodRegistry();
- QueueDeclareOkBody responseBody =
- methodRegistry.createQueueDeclareOkBody(queueName,
- queue.getQueueDepthMessages(),
- queue.getConsumerCount());
- _connection.writeFrame(responseBody.generateFrame(channelId));
-
- _logger.info("Queue " + queueName + " declared successfully");
- }
- return true;
- }
-
- protected AMQQueue createQueue(final AMQChannel channel, final AMQShortString queueName,
- QueueDeclareBody body,
- final VirtualHostImpl virtualHost,
- final AMQProtocolSession session)
- throws AMQException, QueueExistsException
- {
-
- final boolean durable = body.getDurable();
- final boolean autoDelete = body.getAutoDelete();
- final boolean exclusive = body.getExclusive();
-
-
- Map<String, Object> attributes =
- QueueArgumentsConverter.convertWireArgsToModel(FieldTable.convertToMap(body.getArguments()));
- final String queueNameString = AMQShortString.toString(queueName);
- attributes.put(Queue.NAME, queueNameString);
- attributes.put(Queue.ID, UUID.randomUUID());
- 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);
-
-
- final AMQQueue queue = virtualHost.createQueue(attributes);
-
- return queue;
- }
-
- public boolean dispatchQueueDelete(QueueDeleteBody body, int channelId) throws AMQException
- {
- VirtualHostImpl virtualHost = _connection.getVirtualHost();
-
- AMQChannel channel = _connection.getChannel(channelId);
-
- if (channel == null)
- {
- throw body.getChannelNotFoundException(channelId, _connection.getMethodRegistry());
- }
- channel.sync();
- AMQQueue queue;
- if (body.getQueue() == null)
- {
-
- //get the default queue on the channel:
- queue = channel.getDefaultQueue();
- }
- else
- {
- queue = virtualHost.getQueue(body.getQueue().toString());
- }
-
- if (queue == null)
- {
- throw body.getChannelException(AMQConstant.NOT_FOUND, "Queue " + body.getQueue() + " does not exist.",
- _connection.getMethodRegistry());
-
- }
- else
- {
- if (body.getIfEmpty() && !queue.isEmpty())
- {
- throw body.getChannelException(AMQConstant.IN_USE, "Queue: " + body.getQueue() + " is not empty.",
- _connection.getMethodRegistry());
- }
- else if (body.getIfUnused() && !queue.isUnused())
- {
- // TODO - Error code
- throw body.getChannelException(AMQConstant.IN_USE, "Queue: " + body.getQueue() + " is still used.",
- _connection.getMethodRegistry());
- }
- else
- {
- if (!queue.verifySessionAccess(channel))
- {
- throw body.getConnectionException(AMQConstant.NOT_ALLOWED,
- "Queue "
- + queue.getName()
- + " is exclusive, but not created on this Connection.",
- _connection.getMethodRegistry());
- }
-
- int purged = 0;
- try
- {
- purged = virtualHost.removeQueue(queue);
- }
- catch (AccessControlException e)
- {
- throw body.getConnectionException(AMQConstant.ACCESS_REFUSED,
- e.getMessage(),
- _connection.getMethodRegistry());
- }
-
- MethodRegistry methodRegistry = _connection.getMethodRegistry();
- QueueDeleteOkBody responseBody = methodRegistry.createQueueDeleteOkBody(purged);
- _connection.writeFrame(responseBody.generateFrame(channelId));
- }
- }
- return true;
- }
-
- public boolean dispatchQueuePurge(QueuePurgeBody body, int channelId) throws AMQException
- {
- VirtualHostImpl virtualHost = _connection.getVirtualHost();
-
- AMQChannel channel = _connection.getChannel(channelId);
- if (channel == null)
- {
- throw body.getChannelNotFoundException(channelId, _connection.getMethodRegistry());
- }
- AMQQueue queue;
- if(body.getQueue() == null)
- {
-
- //get the default queue on the channel:
- queue = channel.getDefaultQueue();
-
- if(queue == null)
- {
- throw body.getConnectionException(AMQConstant.NOT_ALLOWED,
- "No queue specified.",
- _connection.getMethodRegistry());
- }
- }
- else
- {
- queue = virtualHost.getQueue(body.getQueue().toString());
- }
-
- if(queue == null)
- {
- throw body.getChannelException(AMQConstant.NOT_FOUND, "Queue " + body.getQueue() + " does not exist.",
- _connection.getMethodRegistry());
- }
- else
- {
- if (!queue.verifySessionAccess(channel))
- {
- throw body.getConnectionException(AMQConstant.NOT_ALLOWED,
- "Queue is exclusive, but not created on this Connection.",
- _connection.getMethodRegistry());
- }
-
- long purged = 0;
- try
- {
- purged = queue.clearQueue();
- }
- catch (AccessControlException e)
- {
- throw body.getConnectionException(AMQConstant.ACCESS_REFUSED,
- e.getMessage(),
- _connection.getMethodRegistry());
- }
-
-
- if(!body.getNowait())
- {
- channel.sync();
- MethodRegistry methodRegistry = _connection.getMethodRegistry();
- AMQMethodBody responseBody = methodRegistry.createQueuePurgeOkBody(purged);
- _connection.writeFrame(responseBody.generateFrame(channelId));
-
- }
- }
- return true;
- }
-
-
- public boolean dispatchTxCommit(TxCommitBody body, final int channelId) throws AMQException
- {
- try
- {
- if (_logger.isDebugEnabled())
- {
- _logger.debug("Commit received on channel " + channelId);
- }
- AMQChannel channel = _connection.getChannel(channelId);
-
- if (channel == null)
- {
- throw body.getChannelNotFoundException(channelId, _connection.getMethodRegistry());
- }
- channel.commit(new Runnable()
- {
-
- @Override
- public void run()
- {
- MethodRegistry methodRegistry = _connection.getMethodRegistry();
- AMQMethodBody responseBody = methodRegistry.createTxCommitOkBody();
- _connection.writeFrame(responseBody.generateFrame(channelId));
- }
- }, true);
-
-
-
- }
- catch (AMQException e)
- {
- throw body.getChannelException(e.getErrorCode(), "Failed to commit: " + e.getMessage(),
- _connection.getMethodRegistry());
- }
- return true;
- }
-
- public boolean dispatchTxRollback(TxRollbackBody body, final int channelId) throws AMQException
- {
- try
- {
[... 2456 lines stripped ...]
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org