diff options
| author | Robert Godfrey <rgodfrey@apache.org> | 2007-11-24 21:14:14 +0000 |
|---|---|---|
| committer | Robert Godfrey <rgodfrey@apache.org> | 2007-11-24 21:14:14 +0000 |
| commit | 66f97f32c78e0cf5914a441ae8277ee3aa659ce9 (patch) | |
| tree | f825c3d20483f741c06b9584ddb4e17d6a2404a9 /java/broker | |
| parent | 86996d5680a07acd82ffda2829dcdd6d6585e606 (diff) | |
| download | qpid-python-66f97f32c78e0cf5914a441ae8277ee3aa659ce9.tar.gz | |
QPID-567 : Add mutliversion support to Qpid/Java.
git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/branches/M2.1@597918 13f79535-47bb-0310-9956-ffa450edef68
Diffstat (limited to 'java/broker')
47 files changed, 1843 insertions, 662 deletions
diff --git a/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java b/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java index 17300d6b50..0907ea9df0 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java +++ b/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java @@ -266,8 +266,8 @@ public class AMQChannel // returns true iff the message was delivered (i.e. if all data was // received if (_currentMessage.addContentBodyFrame(_storeContext, - protocolSession.getRegistry().getProtocolVersionMethodConverter().convertToContentChunk( - contentBody))) + protocolSession.getMethodRegistry().getProtocolVersionMethodConverter().convertToContentChunk( + contentBody))) { // callback to allow the context to do any post message processing // primary use is to allow message return processing in the non-tx case diff --git a/java/broker/src/main/java/org/apache/qpid/server/handler/AccessRequestHandler.java b/java/broker/src/main/java/org/apache/qpid/server/handler/AccessRequestHandler.java new file mode 100644 index 0000000000..6ace626c28 --- /dev/null +++ b/java/broker/src/main/java/org/apache/qpid/server/handler/AccessRequestHandler.java @@ -0,0 +1,44 @@ +package org.apache.qpid.server.handler;
+
+import org.apache.qpid.framing.*;
+import org.apache.qpid.server.state.StateAwareMethodListener;
+import org.apache.qpid.server.state.AMQStateManager;
+import org.apache.qpid.server.protocol.AMQProtocolSession;
+import org.apache.qpid.server.virtualhost.VirtualHost;
+import org.apache.qpid.server.queue.QueueRegistry;
+import org.apache.qpid.server.queue.AMQQueue;
+import org.apache.qpid.server.exchange.Exchange;
+import org.apache.qpid.AMQException;
+
+/**
+ * @author Apache Software Foundation
+ *
+ *
+ */
+public class AccessRequestHandler implements StateAwareMethodListener<AccessRequestBody>
+{
+ private static final AccessRequestHandler _instance = new AccessRequestHandler();
+
+
+ public static AccessRequestHandler getInstance()
+ {
+ return _instance;
+ }
+
+ private AccessRequestHandler()
+ {
+ }
+
+ public void methodReceived(AMQStateManager stateManager, AccessRequestBody body, int channelId) throws AMQException
+ {
+ AMQProtocolSession session = stateManager.getProtocolSession();
+
+ MethodRegistry methodRegistry = session.getMethodRegistry();
+
+ // 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);
+
+ session.writeFrame(response.generateFrame(channelId));
+ }
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/handler/BasicAckMethodHandler.java b/java/broker/src/main/java/org/apache/qpid/server/handler/BasicAckMethodHandler.java index a6972475a6..f90e7c3dff 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/handler/BasicAckMethodHandler.java +++ b/java/broker/src/main/java/org/apache/qpid/server/handler/BasicAckMethodHandler.java @@ -44,24 +44,24 @@ public class BasicAckMethodHandler implements StateAwareMethodListener<BasicAckB { } - public void methodReceived(AMQStateManager stateManager, AMQMethodEvent<BasicAckBody> evt) throws AMQException + public void methodReceived(AMQStateManager stateManager, BasicAckBody body, int channelId) throws AMQException { AMQProtocolSession protocolSession = stateManager.getProtocolSession(); - BasicAckBody body = evt.getMethod(); + if (_log.isDebugEnabled()) { - _log.debug("Ack(Tag:" + body.deliveryTag + ":Mult:" + body.multiple + ") received on channel " + evt.getChannelId()); + _log.debug("Ack(Tag:" + body.getDeliveryTag() + ":Mult:" + body.getMultiple() + ") received on channel " + channelId); } - final AMQChannel channel = protocolSession.getChannel(evt.getChannelId()); + final AMQChannel channel = protocolSession.getChannel(channelId); if (channel == null) { - throw body.getChannelNotFoundException(evt.getChannelId()); + throw body.getChannelNotFoundException(channelId); } // this method throws an AMQException if the delivery tag is not known - channel.acknowledgeMessage(body.deliveryTag, body.multiple); + channel.acknowledgeMessage(body.getDeliveryTag(), body.getMultiple()); } } diff --git a/java/broker/src/main/java/org/apache/qpid/server/handler/BasicCancelMethodHandler.java b/java/broker/src/main/java/org/apache/qpid/server/handler/BasicCancelMethodHandler.java index 8bab96a11b..bda1c16cf6 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/handler/BasicCancelMethodHandler.java +++ b/java/broker/src/main/java/org/apache/qpid/server/handler/BasicCancelMethodHandler.java @@ -24,6 +24,7 @@ import org.apache.qpid.AMQException; import org.apache.qpid.framing.AMQFrame; import org.apache.qpid.framing.BasicCancelBody; import org.apache.qpid.framing.BasicCancelOkBody; +import org.apache.qpid.framing.MethodRegistry; import org.apache.qpid.protocol.AMQMethodEvent; import org.apache.qpid.server.AMQChannel; import org.apache.qpid.server.protocol.AMQProtocolSession; @@ -46,34 +47,30 @@ public class BasicCancelMethodHandler implements StateAwareMethodListener<BasicC { } - public void methodReceived(AMQStateManager stateManager, AMQMethodEvent<BasicCancelBody> evt) throws AMQException + public void methodReceived(AMQStateManager stateManager, BasicCancelBody body, int channelId) throws AMQException { - AMQProtocolSession protocolSession = stateManager.getProtocolSession(); + AMQProtocolSession session = stateManager.getProtocolSession(); + + final AMQChannel channel = session.getChannel(channelId); - final AMQChannel channel = protocolSession.getChannel(evt.getChannelId()); - final BasicCancelBody body = evt.getMethod(); if (channel == null) { - throw body.getChannelNotFoundException(evt.getChannelId()); + throw body.getChannelNotFoundException(channelId); } if (_log.isDebugEnabled()) { - _log.debug("BasicCancel: for:" + body.consumerTag + - " nowait:" + body.nowait); + _log.debug("BasicCancel: for:" + body.getConsumerTag() + + " nowait:" + body.getNowait()); } - channel.unsubscribeConsumer(protocolSession, body.consumerTag); - if (!body.nowait) + channel.unsubscribeConsumer(session, body.getConsumerTag()); + if (!body.getNowait()) { - // AMQP version change: Hardwire the version to 0-8 (major=8, minor=0) - // TODO: Connect this to the session version obtained from ProtocolInitiation for this session. - // Be aware of possible changes to parameter order as versions change. - final AMQFrame responseFrame = BasicCancelOkBody.createAMQFrame(evt.getChannelId(), - (byte) 8, (byte) 0, // AMQP version (major, minor) - body.consumerTag); // consumerTag - protocolSession.writeFrame(responseFrame); + MethodRegistry methodRegistry = session.getMethodRegistry(); + BasicCancelOkBody cancelOkBody = methodRegistry.createBasicCancelOkBody(body.getConsumerTag()); + session.writeFrame(cancelOkBody.generateFrame(channelId)); } } } diff --git a/java/broker/src/main/java/org/apache/qpid/server/handler/BasicConsumeMethodHandler.java b/java/broker/src/main/java/org/apache/qpid/server/handler/BasicConsumeMethodHandler.java index d1e4ae6aa6..bb48539f50 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/handler/BasicConsumeMethodHandler.java +++ b/java/broker/src/main/java/org/apache/qpid/server/handler/BasicConsumeMethodHandler.java @@ -22,11 +22,7 @@ package org.apache.qpid.server.handler; import org.apache.log4j.Logger; import org.apache.qpid.AMQException; -import org.apache.qpid.framing.AMQShortString; -import org.apache.qpid.framing.BasicConsumeBody; -import org.apache.qpid.framing.BasicConsumeOkBody; -import org.apache.qpid.framing.ChannelCloseBody; -import org.apache.qpid.framing.ConnectionCloseBody; +import org.apache.qpid.framing.*; import org.apache.qpid.protocol.AMQConstant; import org.apache.qpid.protocol.AMQMethodEvent; import org.apache.qpid.server.AMQChannel; @@ -52,12 +48,12 @@ public class BasicConsumeMethodHandler implements StateAwareMethodListener<Basic { } - public void methodReceived(AMQStateManager stateManager, AMQMethodEvent<BasicConsumeBody> evt) throws AMQException + public void methodReceived(AMQStateManager stateManager, BasicConsumeBody body, int channelId) throws AMQException { AMQProtocolSession session = stateManager.getProtocolSession(); - BasicConsumeBody body = evt.getMethod(); - final int channelId = evt.getChannelId(); + + AMQChannel channel = session.getChannel(channelId); @@ -65,29 +61,29 @@ public class BasicConsumeMethodHandler implements StateAwareMethodListener<Basic if (channel == null) { - throw body.getChannelNotFoundException(evt.getChannelId()); + throw body.getChannelNotFoundException(channelId); } else { if (_log.isDebugEnabled()) { - _log.debug("BasicConsume: from '" + body.queue + - "' for:" + body.consumerTag + - " nowait:" + body.nowait + - " args:" + body.arguments); + _log.debug("BasicConsume: from '" + body.getQueue() + + "' for:" + body.getConsumerTag() + + " nowait:" + body.getNowait() + + " args:" + body.getArguments()); } - AMQQueue queue = body.queue == null ? channel.getDefaultQueue() : vHost.getQueueRegistry().getQueue(body.queue); + AMQQueue queue = body.getQueue() == null ? channel.getDefaultQueue() : vHost.getQueueRegistry().getQueue(body.getQueue().intern()); if (queue == null) { if (_log.isTraceEnabled()) { - _log.trace("No queue for '" + body.queue + "'"); + _log.trace("No queue for '" + body.getQueue() + "'"); } - if (body.queue != null) + if (body.getQueue() != null) { - String msg = "No such queue, '" + body.queue + "'"; + String msg = "No such queue, '" + body.getQueue() + "'"; throw body.getChannelException(AMQConstant.NOT_FOUND, msg); } else @@ -99,23 +95,27 @@ public class BasicConsumeMethodHandler implements StateAwareMethodListener<Basic else { - if (body.consumerTag != null) + final AMQShortString consumerTagName; + + if (body.getConsumerTag() != null) + { + consumerTagName = body.getConsumerTag().intern(); + } + else { - body.consumerTag = body.consumerTag.intern(); + consumerTagName = null; } try { - AMQShortString consumerTag = channel.subscribeToQueue(body.consumerTag, queue, session, !body.noAck, - body.arguments, body.noLocal, body.exclusive); - if (!body.nowait) + AMQShortString consumerTag = channel.subscribeToQueue(consumerTagName, queue, session, !body.getNoAck(), + body.getArguments(), body.getNoLocal(), body.getExclusive()); + if (!body.getNowait()) { - // AMQP version change: Hardwire the version to 0-8 (major=8, minor=0) - // TODO: Connect this to the session version obtained from ProtocolInitiation for this session. - // Be aware of possible changes to parameter order as versions change. - session.writeFrame(BasicConsumeOkBody.createAMQFrame(channelId, - (byte) 8, (byte) 0, // AMQP version (major, minor) - consumerTag)); // consumerTag + MethodRegistry methodRegistry = session.getMethodRegistry(); + AMQMethodBody responseBody = methodRegistry.createBasicConsumeOkBody(consumerTag); + session.writeFrame(responseBody.generateFrame(channelId)); + } //now allow queue to start async processing of any backlog of messages @@ -124,33 +124,26 @@ public class BasicConsumeMethodHandler implements StateAwareMethodListener<Basic catch (org.apache.qpid.AMQInvalidArgumentException ise) { _log.debug("Closing connection due to invalid selector"); - // Why doesn't this ChannelException work. -// throw body.getChannelException(AMQConstant.INVALID_ARGUMENT, ise.getMessage()); - // AMQP version change: Hardwire the version to 0-8 (major=8, minor=0) - // TODO: Connect this to the session version obtained from ProtocolInitiation for this session. - // Be aware of possible changes to parameter order as versions change. - session.writeFrame(ChannelCloseBody.createAMQFrame(channelId, - (byte) 8, (byte) 0, // AMQP version (major, minor) - BasicConsumeBody.getClazz((byte) 8, (byte) 0), // classId - BasicConsumeBody.getMethod((byte) 8, (byte) 0), // methodId - AMQConstant.INVALID_ARGUMENT.getCode(), // replyCode - new AMQShortString(ise.getMessage()))); // replyText + + MethodRegistry methodRegistry = session.getMethodRegistry(); + AMQMethodBody responseBody = methodRegistry.createChannelCloseBody(AMQConstant.INVALID_ARGUMENT.getCode(), + new AMQShortString(ise.getMessage()), + body.getClazz(), + body.getMethod()); + session.writeFrame(responseBody.generateFrame(channelId)); + + } catch (ConsumerTagNotUniqueException e) { - AMQShortString msg = new AMQShortString("Non-unique consumer tag, '" + body.consumerTag + "'"); - // If the above doesn't work then perhaps this is wrong too. -// throw body.getConnectionException(AMQConstant.NOT_ALLOWED, -// "Non-unique consumer tag, '" + body.consumerTag + "'"); - // AMQP version change: Hardwire the version to 0-8 (major=8, minor=0) - // TODO: Connect this to the session version obtained from ProtocolInitiation for this session. - // Be aware of possible changes to parameter order as versions change. - session.writeFrame(ConnectionCloseBody.createAMQFrame(channelId, - (byte) 8, (byte) 0, // AMQP version (major, minor) - BasicConsumeBody.getClazz((byte) 8, (byte) 0), // classId - BasicConsumeBody.getMethod((byte) 8, (byte) 0), // methodId - AMQConstant.NOT_ALLOWED.getCode(), // replyCode - msg)); // replyText + AMQShortString msg = new AMQShortString("Non-unique consumer tag, '" + body.getConsumerTag() + "'"); + + MethodRegistry methodRegistry = session.getMethodRegistry(); + AMQMethodBody responseBody = methodRegistry.createConnectionCloseBody(AMQConstant.NOT_ALLOWED.getCode(), // replyCode + msg, // replytext + body.getClazz(), + body.getMethod()); + session.writeFrame(responseBody.generateFrame(0)); } catch (AMQQueue.ExistingExclusiveSubscription e) { diff --git a/java/broker/src/main/java/org/apache/qpid/server/handler/BasicGetMethodHandler.java b/java/broker/src/main/java/org/apache/qpid/server/handler/BasicGetMethodHandler.java index 782a89c704..8d69697350 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/handler/BasicGetMethodHandler.java +++ b/java/broker/src/main/java/org/apache/qpid/server/handler/BasicGetMethodHandler.java @@ -25,6 +25,7 @@ import org.apache.log4j.Logger; import org.apache.qpid.AMQException;
import org.apache.qpid.framing.BasicGetBody;
import org.apache.qpid.framing.BasicGetEmptyBody;
+import org.apache.qpid.framing.MethodRegistry;
import org.apache.qpid.protocol.AMQConstant;
import org.apache.qpid.protocol.AMQMethodEvent;
import org.apache.qpid.server.AMQChannel;
@@ -49,30 +50,29 @@ public class BasicGetMethodHandler implements StateAwareMethodListener<BasicGetB {
}
- public void methodReceived(AMQStateManager stateManager, AMQMethodEvent<BasicGetBody> evt) throws AMQException
+ public void methodReceived(AMQStateManager stateManager, BasicGetBody body, int channelId) throws AMQException
{
AMQProtocolSession session = stateManager.getProtocolSession();
- BasicGetBody body = evt.getMethod();
- final int channelId = evt.getChannelId();
+
VirtualHost vHost = session.getVirtualHost();
AMQChannel channel = session.getChannel(channelId);
if (channel == null)
{
- throw body.getChannelNotFoundException(evt.getChannelId());
+ throw body.getChannelNotFoundException(channelId);
}
else
{
- AMQQueue queue = body.queue == null ? channel.getDefaultQueue() : vHost.getQueueRegistry().getQueue(body.queue);
+ AMQQueue queue = body.getQueue() == null ? channel.getDefaultQueue() : vHost.getQueueRegistry().getQueue(body.getQueue());
if (queue == null)
{
- _log.info("No queue for '" + body.queue + "'");
- if(body.queue!=null)
+ _log.info("No queue for '" + body.getQueue() + "'");
+ if(body.getQueue()!=null)
{
throw body.getConnectionException(AMQConstant.NOT_FOUND,
- "No such queue, '" + body.queue + "'");
+ "No such queue, '" + body.getQueue()+ "'");
}
else
{
@@ -82,12 +82,14 @@ public class BasicGetMethodHandler implements StateAwareMethodListener<BasicGetB }
else
{
- if(!queue.performGet(session, channel, !body.noAck))
+ if(!queue.performGet(session, channel, !body.getNoAck()))
{
+ MethodRegistry methodRegistry = session.getMethodRegistry();
+ // TODO - set clusterId
+ BasicGetEmptyBody responseBody = methodRegistry.createBasicGetEmptyBody(null);
- // TODO - set clusterId
- session.writeFrame(BasicGetEmptyBody.createAMQFrame(channelId, body.getMajor(), body.getMinor(), null));
+ session.writeFrame(responseBody.generateFrame(channelId));
}
}
}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/handler/BasicPublishMethodHandler.java b/java/broker/src/main/java/org/apache/qpid/server/handler/BasicPublishMethodHandler.java index 68d4483df3..66afc61751 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/handler/BasicPublishMethodHandler.java +++ b/java/broker/src/main/java/org/apache/qpid/server/handler/BasicPublishMethodHandler.java @@ -24,6 +24,8 @@ import org.apache.log4j.Logger; import org.apache.qpid.AMQException; import org.apache.qpid.exchange.ExchangeDefaults; import org.apache.qpid.framing.BasicPublishBody; +import org.apache.qpid.framing.AMQShortString; +import org.apache.qpid.framing.MethodRegistry; import org.apache.qpid.protocol.AMQMethodEvent; import org.apache.qpid.protocol.AMQConstant; import org.apache.qpid.server.AMQChannel; @@ -50,29 +52,27 @@ public class BasicPublishMethodHandler implements StateAwareMethodListener<Basi { } - public void methodReceived(AMQStateManager stateManager, AMQMethodEvent<BasicPublishBody> evt) throws AMQException + public void methodReceived(AMQStateManager stateManager, BasicPublishBody body, int channelId) throws AMQException { AMQProtocolSession session = stateManager.getProtocolSession(); - final BasicPublishBody body = evt.getMethod(); + if (_log.isDebugEnabled()) { - _log.debug("Publish received on channel " + evt.getChannelId()); + _log.debug("Publish received on channel " + channelId); } + AMQShortString exchange = body.getExchange(); // TODO: check the delivery tag field details - is it unique across the broker or per subscriber? - if (body.exchange == null) + if (exchange == null) { - body.exchange = ExchangeDefaults.DEFAULT_EXCHANGE_NAME; + exchange = ExchangeDefaults.DEFAULT_EXCHANGE_NAME; } - else - { - body.exchange = body.exchange.intern(); - } + VirtualHost vHost = session.getVirtualHost(); - Exchange e = vHost.getExchangeRegistry().getExchange(body.exchange); + Exchange e = vHost.getExchangeRegistry().getExchange(exchange); // if the exchange does not exist we raise a channel exception if (e == null) { @@ -83,22 +83,18 @@ public class BasicPublishMethodHandler implements StateAwareMethodListener<Basi // 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 = session.getChannel(evt.getChannelId()); + AMQChannel channel = session.getChannel(channelId); if (channel == null) { - throw body.getChannelNotFoundException(evt.getChannelId()); + throw body.getChannelNotFoundException(channelId); } - if(body.routingKey != null) - { - body.routingKey = body.routingKey.intern(); - } - - MessagePublishInfo info = session.getRegistry().getProtocolVersionMethodConverter().convertToInfo(body); + MessagePublishInfo info = session.getMethodRegistry().getProtocolVersionMethodConverter().convertToInfo(body); channel.setPublishFrame(info, session); } } + } diff --git a/java/broker/src/main/java/org/apache/qpid/server/handler/BasicQosHandler.java b/java/broker/src/main/java/org/apache/qpid/server/handler/BasicQosHandler.java index 3cd6a87f64..3c95180dca 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/handler/BasicQosHandler.java +++ b/java/broker/src/main/java/org/apache/qpid/server/handler/BasicQosHandler.java @@ -23,6 +23,8 @@ package org.apache.qpid.server.handler; import org.apache.qpid.AMQException; import org.apache.qpid.framing.BasicQosBody; import org.apache.qpid.framing.BasicQosOkBody; +import org.apache.qpid.framing.MethodRegistry; +import org.apache.qpid.framing.AMQMethodBody; import org.apache.qpid.protocol.AMQMethodEvent; import org.apache.qpid.server.protocol.AMQProtocolSession; import org.apache.qpid.server.state.AMQStateManager; @@ -38,21 +40,21 @@ public class BasicQosHandler implements StateAwareMethodListener<BasicQosBody> return _instance; } - public void methodReceived(AMQStateManager stateManager, AMQMethodEvent<BasicQosBody> evt) throws AMQException + public void methodReceived(AMQStateManager stateManager, BasicQosBody body, int channelId) throws AMQException { AMQProtocolSession session = stateManager.getProtocolSession(); - AMQChannel channel = session.getChannel(evt.getChannelId()); + AMQChannel channel = session.getChannel(channelId); if (channel == null) { - throw evt.getMethod().getChannelNotFoundException(evt.getChannelId()); + throw body.getChannelNotFoundException(channelId); } - channel.setPrefetchCount(evt.getMethod().prefetchCount); - channel.setPrefetchSize(evt.getMethod().prefetchSize); + channel.setPrefetchCount(body.getPrefetchCount()); + channel.setPrefetchSize(body.getPrefetchSize()); - // AMQP version change: Hardwire the version to 0-8 (major=8, minor=0) - // TODO: Connect this to the session version obtained from ProtocolInitiation for this session. - // Be aware of possible changes to parameter order as versions change. - session.writeFrame(BasicQosOkBody.createAMQFrame(evt.getChannelId(), (byte) 8, (byte) 0)); + MethodRegistry methodRegistry = session.getMethodRegistry(); + AMQMethodBody responseBody = methodRegistry.createBasicQosOkBody(); + session.writeFrame(responseBody.generateFrame(channelId)); + } } diff --git a/java/broker/src/main/java/org/apache/qpid/server/handler/BasicRecoverMethodHandler.java b/java/broker/src/main/java/org/apache/qpid/server/handler/BasicRecoverMethodHandler.java index a436c35473..c7842cd643 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/handler/BasicRecoverMethodHandler.java +++ b/java/broker/src/main/java/org/apache/qpid/server/handler/BasicRecoverMethodHandler.java @@ -24,6 +24,9 @@ import org.apache.log4j.Logger; import org.apache.qpid.AMQException; import org.apache.qpid.framing.BasicRecoverBody; import org.apache.qpid.framing.BasicRecoverOkBody; +import org.apache.qpid.framing.ProtocolVersion; +import org.apache.qpid.framing.AMQMethodBody; +import org.apache.qpid.framing.amqp_8_0.MethodRegistry_8_0; import org.apache.qpid.protocol.AMQMethodEvent; import org.apache.qpid.server.AMQChannel; import org.apache.qpid.server.protocol.AMQProtocolSession; @@ -41,24 +44,30 @@ public class BasicRecoverMethodHandler implements StateAwareMethodListener<Basic return _instance; } - public void methodReceived(AMQStateManager stateManager, AMQMethodEvent<BasicRecoverBody> evt) throws AMQException + public void methodReceived(AMQStateManager stateManager, BasicRecoverBody body, int channelId) throws AMQException { AMQProtocolSession session = stateManager.getProtocolSession(); - _logger.debug("Recover received on protocol session " + session + " and channel " + evt.getChannelId()); - AMQChannel channel = session.getChannel(evt.getChannelId()); - BasicRecoverBody body = evt.getMethod(); + _logger.debug("Recover received on protocol session " + session + " and channel " + channelId); + AMQChannel channel = session.getChannel(channelId); + if (channel == null) { - throw body.getChannelNotFoundException(evt.getChannelId()); + throw body.getChannelNotFoundException(channelId); } - channel.resend(body.requeue); + channel.resend(body.getRequeue()); + + // 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(session.getProtocolVersion().equals(ProtocolVersion.v8_0)) + { + MethodRegistry_8_0 methodRegistry = (MethodRegistry_8_0) session.getMethodRegistry(); + AMQMethodBody recoverOk = methodRegistry.createBasicRecoverOkBody(); + session.writeFrame(recoverOk.generateFrame(channelId)); + + } - // AMQP version change: Hardwire the version to 0-8 (major=8, minor=0) - // TODO: Connect this to the session version obtained from ProtocolInitiation for this session. - // Be aware of possible changes to parameter order as versions change. - session.writeFrame(BasicRecoverOkBody.createAMQFrame(evt.getChannelId(), (byte) 8, (byte) 0)); } } diff --git a/java/broker/src/main/java/org/apache/qpid/server/handler/BasicRecoverSyncMethodHandler.java b/java/broker/src/main/java/org/apache/qpid/server/handler/BasicRecoverSyncMethodHandler.java new file mode 100644 index 0000000000..15484273c8 --- /dev/null +++ b/java/broker/src/main/java/org/apache/qpid/server/handler/BasicRecoverSyncMethodHandler.java @@ -0,0 +1,54 @@ +package org.apache.qpid.server.handler;
+
+import org.apache.log4j.Logger;
+
+import org.apache.qpid.framing.BasicRecoverBody;
+import org.apache.qpid.framing.ProtocolVersion;
+import org.apache.qpid.framing.AMQMethodBody;
+import org.apache.qpid.framing.BasicRecoverSyncBody;
+import org.apache.qpid.framing.amqp_0_9.MethodRegistry_0_9;
+import org.apache.qpid.framing.amqp_8_0.MethodRegistry_8_0;
+import org.apache.qpid.server.state.StateAwareMethodListener;
+import org.apache.qpid.server.state.AMQStateManager;
+import org.apache.qpid.server.protocol.AMQProtocolSession;
+import org.apache.qpid.server.AMQChannel;
+import org.apache.qpid.AMQException;
+
+public class BasicRecoverSyncMethodHandler implements StateAwareMethodListener<BasicRecoverSyncBody>
+{
+ private static final Logger _logger = Logger.getLogger(BasicRecoverSyncMethodHandler.class);
+
+ private static final BasicRecoverSyncMethodHandler _instance = new BasicRecoverSyncMethodHandler();
+
+ public static BasicRecoverSyncMethodHandler getInstance()
+ {
+ return _instance;
+ }
+
+ public void methodReceived(AMQStateManager stateManager, BasicRecoverSyncBody body, int channelId) throws AMQException
+ {
+ AMQProtocolSession session = stateManager.getProtocolSession();
+
+ _logger.debug("Recover received on protocol session " + session + " and channel " + channelId);
+ AMQChannel channel = session.getChannel(channelId);
+
+
+ if (channel == null)
+ {
+ throw body.getChannelNotFoundException(channelId);
+ }
+
+ channel.resend(body.getRequeue());
+
+ // 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(session.getProtocolVersion().equals(ProtocolVersion.v0_9))
+ {
+ MethodRegistry_0_9 methodRegistry = (MethodRegistry_0_9) session.getMethodRegistry();
+ AMQMethodBody recoverOk = methodRegistry.createBasicRecoverSyncOkBody();
+ session.writeFrame(recoverOk.generateFrame(channelId));
+
+ }
+
+ }
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/handler/BasicRejectMethodHandler.java b/java/broker/src/main/java/org/apache/qpid/server/handler/BasicRejectMethodHandler.java index aca485f62a..4f3aee0c93 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/handler/BasicRejectMethodHandler.java +++ b/java/broker/src/main/java/org/apache/qpid/server/handler/BasicRejectMethodHandler.java @@ -45,11 +45,11 @@ public class BasicRejectMethodHandler implements StateAwareMethodListener<BasicR { } - public void methodReceived(AMQStateManager stateManager, AMQMethodEvent<BasicRejectBody> evt) throws AMQException + public void methodReceived(AMQStateManager stateManager, BasicRejectBody body, int channelId) throws AMQException { AMQProtocolSession session = stateManager.getProtocolSession(); - int channelId = evt.getChannelId(); + // if (_logger.isDebugEnabled()) // { @@ -63,18 +63,18 @@ public class BasicRejectMethodHandler implements StateAwareMethodListener<BasicR if (channel == null) { - throw evt.getMethod().getChannelNotFoundException(channelId); + throw body.getChannelNotFoundException(channelId); } if (_logger.isDebugEnabled()) { - _logger.debug("Rejecting:" + evt.getMethod().deliveryTag + - ": Requeue:" + evt.getMethod().requeue + + _logger.debug("Rejecting:" + body.getDeliveryTag() + + ": Requeue:" + body.getRequeue() + //": Resend:" + evt.getMethod().resend + " on channel:" + channel.debugIdentity()); } - long deliveryTag = evt.getMethod().deliveryTag; + long deliveryTag = body.getDeliveryTag(); UnacknowledgedMessage message = channel.getUnacknowledgedMessageMap().get(deliveryTag); @@ -103,7 +103,7 @@ public class BasicRejectMethodHandler implements StateAwareMethodListener<BasicR if (_logger.isTraceEnabled()) { _logger.trace("Rejecting: DT:" + deliveryTag + "-" + message.message.debugIdentity() + - ": Requeue:" + evt.getMethod().requeue + + ": Requeue:" + body.getRequeue() + //": Resend:" + evt.getMethod().resend + " on channel:" + channel.debugIdentity()); } @@ -114,7 +114,7 @@ public class BasicRejectMethodHandler implements StateAwareMethodListener<BasicR message.message.reject(message.message.getDeliveredSubscription(message.queue)); } - if (evt.getMethod().requeue) + if (body.getRequeue()) { channel.requeue(deliveryTag); } diff --git a/java/broker/src/main/java/org/apache/qpid/server/handler/ChannelCloseHandler.java b/java/broker/src/main/java/org/apache/qpid/server/handler/ChannelCloseHandler.java index 1f4f1f9221..9133cce6b7 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/handler/ChannelCloseHandler.java +++ b/java/broker/src/main/java/org/apache/qpid/server/handler/ChannelCloseHandler.java @@ -25,6 +25,7 @@ import org.apache.qpid.AMQException; import org.apache.qpid.framing.AMQFrame; import org.apache.qpid.framing.ChannelCloseBody; import org.apache.qpid.framing.ChannelCloseOkBody; +import org.apache.qpid.framing.MethodRegistry; import org.apache.qpid.protocol.AMQMethodEvent; import org.apache.qpid.protocol.AMQConstant; import org.apache.qpid.server.protocol.AMQProtocolSession; @@ -47,16 +48,16 @@ public class ChannelCloseHandler implements StateAwareMethodListener<ChannelClos { } - public void methodReceived(AMQStateManager stateManager, AMQMethodEvent<ChannelCloseBody> evt) throws AMQException + public void methodReceived(AMQStateManager stateManager, ChannelCloseBody body, int channelId) throws AMQException { AMQProtocolSession session = stateManager.getProtocolSession(); - ChannelCloseBody body = evt.getMethod(); + if (_logger.isInfoEnabled()) { - _logger.info("Received channel close for id " + evt.getChannelId() + " citing class " + body.classId + - " and method " + body.methodId); + _logger.info("Received channel close for id " + channelId + " citing class " + body.getClassId() + + " and method " + body.getMethodId()); } - int channelId = evt.getChannelId(); + AMQChannel channel = session.getChannel(channelId); @@ -69,10 +70,8 @@ public class ChannelCloseHandler implements StateAwareMethodListener<ChannelClos // Client requested closure so we don't wait for ok we send it stateManager.getProtocolSession().closeChannelOk(channelId); - // AMQP version change: Hardwire the version to 0-8 (major=8, minor=0) - // TODO: Connect this to the session version obtained from ProtocolInitiation for this session. - // Be aware of possible changes to parameter order as versions change. - AMQFrame response = ChannelCloseOkBody.createAMQFrame(evt.getChannelId(), (byte) 8, (byte) 0); - session.writeFrame(response); + MethodRegistry methodRegistry = session.getMethodRegistry(); + ChannelCloseOkBody responseBody = methodRegistry.createChannelCloseOkBody(); + session.writeFrame(responseBody.generateFrame(channelId)); } } diff --git a/java/broker/src/main/java/org/apache/qpid/server/handler/ChannelCloseOkHandler.java b/java/broker/src/main/java/org/apache/qpid/server/handler/ChannelCloseOkHandler.java index ad5604e7ea..a857490e7e 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/handler/ChannelCloseOkHandler.java +++ b/java/broker/src/main/java/org/apache/qpid/server/handler/ChannelCloseOkHandler.java @@ -42,9 +42,9 @@ public class ChannelCloseOkHandler implements StateAwareMethodListener<ChannelCl { } - public void methodReceived(AMQStateManager stateManager, AMQMethodEvent<ChannelCloseOkBody> evt) throws AMQException + public void methodReceived(AMQStateManager stateManager, ChannelCloseOkBody body, int channelId) throws AMQException { - int channelId = evt.getChannelId(); + _logger.info("Received channel-close-ok for channel-id " + channelId); // Let the Protocol Session know the channel is now closed. diff --git a/java/broker/src/main/java/org/apache/qpid/server/handler/ChannelFlowHandler.java b/java/broker/src/main/java/org/apache/qpid/server/handler/ChannelFlowHandler.java index bfa170cfc5..696ca8a63b 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/handler/ChannelFlowHandler.java +++ b/java/broker/src/main/java/org/apache/qpid/server/handler/ChannelFlowHandler.java @@ -22,9 +22,7 @@ package org.apache.qpid.server.handler; import org.apache.log4j.Logger; import org.apache.qpid.AMQException; -import org.apache.qpid.framing.AMQFrame; -import org.apache.qpid.framing.ChannelFlowBody; -import org.apache.qpid.framing.ChannelFlowOkBody; +import org.apache.qpid.framing.*; import org.apache.qpid.protocol.AMQMethodEvent; import org.apache.qpid.server.AMQChannel; import org.apache.qpid.server.protocol.AMQProtocolSession; @@ -46,27 +44,23 @@ public class ChannelFlowHandler implements StateAwareMethodListener<ChannelFlowB { } - public void methodReceived(AMQStateManager stateManager, AMQMethodEvent<ChannelFlowBody> evt) throws AMQException + public void methodReceived(AMQStateManager stateManager, ChannelFlowBody body, int channelId) throws AMQException { AMQProtocolSession session = stateManager.getProtocolSession(); - ChannelFlowBody body = evt.getMethod(); - AMQChannel channel = session.getChannel(evt.getChannelId()); + + AMQChannel channel = session.getChannel(channelId); if (channel == null) { - throw body.getChannelNotFoundException(evt.getChannelId()); + throw body.getChannelNotFoundException(channelId); } - channel.setSuspended(!body.active); - _logger.debug("Channel.Flow for channel " + evt.getChannelId() + ", active=" + body.active); + channel.setSuspended(!body.getActive()); + _logger.debug("Channel.Flow for channel " + channelId + ", active=" + body.getActive()); - // AMQP version change: Hardwire the version to 0-8 (major=8, minor=0) - // TODO: Connect this to the session version obtained from ProtocolInitiation for this session. - // Be aware of possible changes to parameter order as versions change. - AMQFrame response = ChannelFlowOkBody.createAMQFrame(evt.getChannelId(), - (byte)8, (byte)0, // AMQP version (major, minor) - body.active); // active - session.writeFrame(response); + MethodRegistry methodRegistry = session.getMethodRegistry(); + AMQMethodBody responseBody = methodRegistry.createChannelFlowOkBody(body.getActive()); + session.writeFrame(responseBody.generateFrame(channelId)); } } diff --git a/java/broker/src/main/java/org/apache/qpid/server/handler/ChannelOpenHandler.java b/java/broker/src/main/java/org/apache/qpid/server/handler/ChannelOpenHandler.java index 03fc7a3926..3f604480b9 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/handler/ChannelOpenHandler.java +++ b/java/broker/src/main/java/org/apache/qpid/server/handler/ChannelOpenHandler.java @@ -20,11 +20,17 @@ */ package org.apache.qpid.server.handler; +import java.util.UUID; +import java.io.ByteArrayOutputStream; +import java.io.DataOutputStream; +import java.io.IOException; + import org.apache.qpid.AMQException; -import org.apache.qpid.framing.AMQFrame; -import org.apache.qpid.framing.ChannelOpenBody; -import org.apache.qpid.framing.ChannelOpenOkBody; +import org.apache.qpid.framing.*; +import org.apache.qpid.framing.amqp_0_9.MethodRegistry_0_9; +import org.apache.qpid.framing.amqp_8_0.MethodRegistry_8_0; import org.apache.qpid.protocol.AMQMethodEvent; +import org.apache.qpid.protocol.AMQConstant; import org.apache.qpid.server.AMQChannel; import org.apache.qpid.server.protocol.AMQProtocolSession; import org.apache.qpid.server.state.AMQStateManager; @@ -44,18 +50,55 @@ public class ChannelOpenHandler implements StateAwareMethodListener<ChannelOpenB { } - public void methodReceived(AMQStateManager stateManager, AMQMethodEvent<ChannelOpenBody> evt) throws AMQException + public void methodReceived(AMQStateManager stateManager, ChannelOpenBody body, int channelId) throws AMQException { AMQProtocolSession session = stateManager.getProtocolSession(); VirtualHost virtualHost = session.getVirtualHost(); - final AMQChannel channel = new AMQChannel(session,evt.getChannelId(), virtualHost.getMessageStore(), + final AMQChannel channel = new AMQChannel(session,channelId, virtualHost.getMessageStore(), virtualHost.getExchangeRegistry()); session.addChannel(channel); - // AMQP version change: Hardwire the version to 0-8 (major=8, minor=0) - // TODO: Connect this to the session version obtained from ProtocolInitiation for this session. - // Be aware of possible changes to parameter order as versions change. - AMQFrame response = ChannelOpenOkBody.createAMQFrame(evt.getChannelId(), (byte)8, (byte)0); - session.writeFrame(response); + + ChannelOpenOkBody response; + + ProtocolVersion pv = session.getProtocolVersion(); + + if(pv.equals(ProtocolVersion.v8_0)) + { + MethodRegistry_8_0 methodRegistry = (MethodRegistry_8_0) MethodRegistry.getMethodRegistry(ProtocolVersion.v8_0); + response = methodRegistry.createChannelOpenOkBody(); + + } + else if(pv.equals(ProtocolVersion.v0_9)) + { + MethodRegistry_0_9 methodRegistry = (MethodRegistry_0_9) MethodRegistry.getMethodRegistry(ProtocolVersion.v0_9); + UUID uuid = UUID.randomUUID(); + ByteArrayOutputStream output = new ByteArrayOutputStream(); + DataOutputStream dataOut = new DataOutputStream(output); + try + { + dataOut.writeLong(uuid.getMostSignificantBits()); + dataOut.writeLong(uuid.getLeastSignificantBits()); + dataOut.flush(); + dataOut.close(); + } + catch (IOException e) + { + // This *really* shouldn't happen as we're not doing any I/O + throw new RuntimeException("I/O exception when writing to byte array", e); + } + + // should really associate this channelId to the session + byte[] channelName = output.toByteArray(); + + response = methodRegistry.createChannelOpenOkBody(channelName); + } + else + { + throw new AMQException(AMQConstant.INTERNAL_ERROR, "Got channel open for protocol version not catered for: " + pv, null); + } + + + session.writeFrame(response.generateFrame(channelId)); } } diff --git a/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionCloseMethodHandler.java b/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionCloseMethodHandler.java index b086cad67f..dade5d5f54 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionCloseMethodHandler.java +++ b/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionCloseMethodHandler.java @@ -25,6 +25,7 @@ import org.apache.qpid.AMQException; import org.apache.qpid.framing.AMQFrame; import org.apache.qpid.framing.ConnectionCloseBody; import org.apache.qpid.framing.ConnectionCloseOkBody; +import org.apache.qpid.framing.MethodRegistry; import org.apache.qpid.protocol.AMQMethodEvent; import org.apache.qpid.server.protocol.AMQProtocolSession; import org.apache.qpid.server.state.AMQStateManager; @@ -45,14 +46,14 @@ public class ConnectionCloseMethodHandler implements StateAwareMethodListener<Co { } - public void methodReceived(AMQStateManager stateManager, AMQMethodEvent<ConnectionCloseBody> evt) throws AMQException + public void methodReceived(AMQStateManager stateManager, ConnectionCloseBody body, int channelId) throws AMQException { AMQProtocolSession session = stateManager.getProtocolSession(); - final ConnectionCloseBody body = evt.getMethod(); + if (_logger.isInfoEnabled()) { - _logger.info("ConnectionClose received with reply code/reply text " + body.replyCode + "/" + - body.replyText + " for " + session); + _logger.info("ConnectionClose received with reply code/reply text " + body.getReplyCode() + "/" + + body.getReplyText() + " for " + session); } try { @@ -62,10 +63,10 @@ public class ConnectionCloseMethodHandler implements StateAwareMethodListener<Co { _logger.error("Error closing protocol session: " + e, e); } - // AMQP version change: Hardwire the version to 0-8 (major=8, minor=0) - // TODO: Connect this to the session version obtained from ProtocolInitiation for this session. - // Be aware of possible changes to parameter order as versions change. - final AMQFrame response = ConnectionCloseOkBody.createAMQFrame(evt.getChannelId(), (byte) 8, (byte) 0); - session.writeFrame(response); + + MethodRegistry methodRegistry = session.getMethodRegistry(); + ConnectionCloseOkBody responseBody = methodRegistry.createConnectionCloseOkBody(); + session.writeFrame(responseBody.generateFrame(channelId)); + } } diff --git a/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionCloseOkMethodHandler.java b/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionCloseOkMethodHandler.java index 853f4df435..bc6e5ab403 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionCloseOkMethodHandler.java +++ b/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionCloseOkMethodHandler.java @@ -44,7 +44,7 @@ public class ConnectionCloseOkMethodHandler implements StateAwareMethodListener< { } - public void methodReceived(AMQStateManager stateManager, AMQMethodEvent<ConnectionCloseOkBody> evt) throws AMQException + public void methodReceived(AMQStateManager stateManager, ConnectionCloseOkBody body, int channelId) throws AMQException { AMQProtocolSession session = stateManager.getProtocolSession(); //todo should this not do more than just log the method? diff --git a/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionOpenMethodHandler.java b/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionOpenMethodHandler.java index 30a40c5a75..4f62e0b930 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionOpenMethodHandler.java +++ b/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionOpenMethodHandler.java @@ -21,10 +21,7 @@ package org.apache.qpid.server.handler; import org.apache.qpid.AMQException; -import org.apache.qpid.framing.AMQFrame; -import org.apache.qpid.framing.AMQShortString; -import org.apache.qpid.framing.ConnectionOpenBody; -import org.apache.qpid.framing.ConnectionOpenOkBody; +import org.apache.qpid.framing.*; import org.apache.qpid.protocol.AMQConstant; import org.apache.qpid.protocol.AMQMethodEvent; import org.apache.qpid.server.protocol.AMQProtocolSession; @@ -56,20 +53,20 @@ public class ConnectionOpenMethodHandler implements StateAwareMethodListener<Con return new AMQShortString(Long.toString(System.currentTimeMillis())); } - public void methodReceived(AMQStateManager stateManager, AMQMethodEvent<ConnectionOpenBody> evt) throws AMQException + public void methodReceived(AMQStateManager stateManager, ConnectionOpenBody body, int channelId) throws AMQException { AMQProtocolSession session = stateManager.getProtocolSession(); - ConnectionOpenBody body = evt.getMethod(); + //ignore leading '/' String virtualHostName; - if ((body.virtualHost != null) && body.virtualHost.charAt(0) == '/') + if ((body.getVirtualHost() != null) && body.getVirtualHost().charAt(0) == '/') { - virtualHostName = new StringBuilder(body.virtualHost.subSequence(1, body.virtualHost.length())).toString(); + virtualHostName = new StringBuilder(body.getVirtualHost().subSequence(1, body.getVirtualHost().length())).toString(); } else { - virtualHostName = body.virtualHost == null ? null : String.valueOf(body.virtualHost); + virtualHostName = body.getVirtualHost() == null ? null : String.valueOf(body.getVirtualHost()); } VirtualHost virtualHost = stateManager.getVirtualHostRegistry().getVirtualHost(virtualHostName); @@ -105,14 +102,14 @@ public class ConnectionOpenMethodHandler implements StateAwareMethodListener<Con session.setContextKey(generateClientID()); } - // AMQP version change: Hardwire the version to 0-8 (major=8, minor=0) - // TODO: Connect this to the session version obtained from ProtocolInitiation for this session. - // Be aware of possible changes to parameter order as versions change. - AMQFrame response = ConnectionOpenOkBody.createAMQFrame((short) 0, - (byte) 8, (byte) 0, // AMQP version (major, minor) - body.virtualHost); + MethodRegistry methodRegistry = session.getMethodRegistry(); + AMQMethodBody responseBody = methodRegistry.createConnectionOpenOkBody(body.getVirtualHost()); + stateManager.changeState(AMQState.CONNECTION_OPEN); - session.writeFrame(response); + + session.writeFrame(responseBody.generateFrame(channelId)); + + } } } diff --git a/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionSecureOkMethodHandler.java b/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionSecureOkMethodHandler.java index fef00942a0..193c3a088b 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionSecureOkMethodHandler.java +++ b/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionSecureOkMethodHandler.java @@ -25,11 +25,7 @@ import javax.security.sasl.SaslServer; import org.apache.log4j.Logger; import org.apache.qpid.AMQException; -import org.apache.qpid.framing.AMQFrame; -import org.apache.qpid.framing.ConnectionCloseBody; -import org.apache.qpid.framing.ConnectionSecureBody; -import org.apache.qpid.framing.ConnectionSecureOkBody; -import org.apache.qpid.framing.ConnectionTuneBody; +import org.apache.qpid.framing.*; import org.apache.qpid.protocol.AMQConstant; import org.apache.qpid.protocol.AMQMethodEvent; import org.apache.qpid.server.protocol.AMQProtocolSession; @@ -57,10 +53,10 @@ public class ConnectionSecureOkMethodHandler implements StateAwareMethodListener { } - public void methodReceived(AMQStateManager stateManager, AMQMethodEvent<ConnectionSecureOkBody> evt) throws AMQException + public void methodReceived(AMQStateManager stateManager, ConnectionSecureOkBody body, int channelId) throws AMQException { AMQProtocolSession session = stateManager.getProtocolSession(); - ConnectionSecureOkBody body = evt.getMethod(); + //fixme Vhost not defined yet //session.getVirtualHost().getAuthenticationManager(); @@ -71,8 +67,8 @@ public class ConnectionSecureOkMethodHandler implements StateAwareMethodListener { throw new AMQException("No SASL context set up in session"); } - - AuthenticationResult authResult = authMgr.authenticate(ss, body.response); + MethodRegistry methodRegistry = session.getMethodRegistry(); + AuthenticationResult authResult = authMgr.authenticate(ss, body.getResponse()); switch (authResult.status) { case ERROR: @@ -80,45 +76,34 @@ public class ConnectionSecureOkMethodHandler implements StateAwareMethodListener // throw new AMQException(AMQConstant.NOT_ALLOWED.getCode(), AMQConstant.NOT_ALLOWED.getName()); _logger.info("Authentication failed"); stateManager.changeState(AMQState.CONNECTION_CLOSING); - // AMQP version change: Hardwire the version to 0-8 (major=8, minor=0) - // TODO: Connect this to the session version obtained from ProtocolInitiation for this session. - // Be aware of possible changes to parameter order as versions change. - AMQFrame close = ConnectionCloseBody.createAMQFrame(0, - (byte)8, (byte)0, // AMQP version (major, minor) - ConnectionCloseBody.getClazz((byte)8, (byte)0), // classId - ConnectionCloseBody.getMethod((byte)8, (byte)0), // methodId - AMQConstant.NOT_ALLOWED.getCode(), // replyCode - AMQConstant.NOT_ALLOWED.getName()); // replyText - session.writeFrame(close); + + + ConnectionCloseBody connectionCloseBody = + methodRegistry.createConnectionCloseBody(AMQConstant.NOT_ALLOWED.getCode(), + AMQConstant.NOT_ALLOWED.getName(), + body.getClazz(), + body.getMethod()); + + session.writeFrame(connectionCloseBody.generateFrame(0) ); disposeSaslServer(session); break; case SUCCESS: _logger.info("Connected as: " + ss.getAuthorizationID()); stateManager.changeState(AMQState.CONNECTION_NOT_TUNED); - // TODO: Check the value of channelMax here: This should be the max - // value of a 2-byte unsigned integer (as channel is only 2 bytes on the wire), - // not Integer.MAX_VALUE (which is signed 4 bytes). - // AMQP version change: Hardwire the version to 0-8 (major=8, minor=0) - // TODO: Connect this to the session version obtained from ProtocolInitiation for this session. - // Be aware of possible changes to parameter order as versions change. - AMQFrame tune = ConnectionTuneBody.createAMQFrame(0, - (byte)8, (byte)0, // AMQP version (major, minor) - Integer.MAX_VALUE, // channelMax - ConnectionStartOkMethodHandler.getConfiguredFrameSize(), // frameMax - HeartbeatConfig.getInstance().getDelay()); // heartbeat - session.writeFrame(tune); + + ConnectionTuneBody tuneBody = + methodRegistry.createConnectionTuneBody(0xFFFF, + ConnectionStartOkMethodHandler.getConfiguredFrameSize(), + HeartbeatConfig.getInstance().getDelay()); + session.writeFrame(tuneBody.generateFrame(0)); session.setAuthorizedID(new UsernamePrincipal(ss.getAuthorizationID())); disposeSaslServer(session); break; case CONTINUE: stateManager.changeState(AMQState.CONNECTION_NOT_AUTH); - // AMQP version change: Hardwire the version to 0-8 (major=8, minor=0) - // TODO: Connect this to the session version obtained from ProtocolInitiation for this session. - // Be aware of possible changes to parameter order as versions change. - AMQFrame challenge = ConnectionSecureBody.createAMQFrame(0, - (byte)8, (byte)0, // AMQP version (major, minor) - authResult.challenge); // challenge - session.writeFrame(challenge); + + ConnectionSecureBody secureBody = methodRegistry.createConnectionSecureBody(authResult.challenge); + session.writeFrame(secureBody.generateFrame(0)); } } diff --git a/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionStartOkMethodHandler.java b/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionStartOkMethodHandler.java index bb9eeed1d4..f02121c89f 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionStartOkMethodHandler.java +++ b/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionStartOkMethodHandler.java @@ -26,11 +26,7 @@ import javax.security.sasl.SaslServer; import org.apache.commons.configuration.Configuration; import org.apache.log4j.Logger; import org.apache.qpid.AMQException; -import org.apache.qpid.framing.AMQFrame; -import org.apache.qpid.framing.ConnectionSecureBody; -import org.apache.qpid.framing.ConnectionStartOkBody; -import org.apache.qpid.framing.ConnectionTuneBody; -import org.apache.qpid.framing.ConnectionCloseBody; +import org.apache.qpid.framing.*; import org.apache.qpid.protocol.AMQMethodEvent; import org.apache.qpid.protocol.AMQConstant; import org.apache.qpid.server.protocol.AMQProtocolSession; @@ -53,7 +49,7 @@ public class ConnectionStartOkMethodHandler implements StateAwareMethodListener< private static final int DEFAULT_FRAME_SIZE = 65536; - public static StateAwareMethodListener<ConnectionStartOkBody> getInstance() + public static ConnectionStartOkMethodHandler getInstance() { return _instance; } @@ -62,51 +58,51 @@ public class ConnectionStartOkMethodHandler implements StateAwareMethodListener< { } - public void methodReceived(AMQStateManager stateManager, AMQMethodEvent<ConnectionStartOkBody> evt) throws AMQException + public void methodReceived(AMQStateManager stateManager, ConnectionStartOkBody body, int channelId) throws AMQException { AMQProtocolSession session = stateManager.getProtocolSession(); - final ConnectionStartOkBody body = evt.getMethod(); - _logger.info("SASL Mechanism selected: " + body.mechanism); - _logger.info("Locale selected: " + body.locale); + + _logger.info("SASL Mechanism selected: " + body.getMechanism()); + _logger.info("Locale selected: " + body.getLocale()); AuthenticationManager authMgr = ApplicationRegistry.getInstance().getAuthenticationManager();//session.getVirtualHost().getAuthenticationManager(); SaslServer ss = null; try { - ss = authMgr.createSaslServer(String.valueOf(body.mechanism), session.getLocalFQDN()); + ss = authMgr.createSaslServer(String.valueOf(body.getMechanism()), session.getLocalFQDN()); if (ss == null) { - throw body.getConnectionException(AMQConstant.RESOURCE_ERROR, "Unable to create SASL Server:" + body.mechanism + throw body.getConnectionException(AMQConstant.RESOURCE_ERROR, "Unable to create SASL Server:" + body.getMechanism() ); } session.setSaslServer(ss); - AuthenticationResult authResult = authMgr.authenticate(ss, body.response); + AuthenticationResult authResult = authMgr.authenticate(ss, body.getResponse()); //save clientProperties if (session.getClientProperties() == null) { - session.setClientProperties(body.clientProperties); + session.setClientProperties(body.getClientProperties()); } + MethodRegistry methodRegistry = session.getMethodRegistry(); + switch (authResult.status) { case ERROR: _logger.info("Authentication failed"); stateManager.changeState(AMQState.CONNECTION_CLOSING); - // AMQP version change: Hardwire the version to 0-8 (major=8, minor=0) - // TODO: Connect this to the session version obtained from ProtocolInitiation for this session. - // Be aware of possible changes to parameter order as versions change. - AMQFrame close = ConnectionCloseBody.createAMQFrame(0, - (byte) 8, (byte) 0, // AMQP version (major, minor) - ConnectionCloseBody.getClazz((byte) 8, (byte) 0), // classId - ConnectionCloseBody.getMethod((byte) 8, (byte) 0), // methodId - AMQConstant.NOT_ALLOWED.getCode(), // replyCode - AMQConstant.NOT_ALLOWED.getName()); // replyText - session.writeFrame(close); + + ConnectionCloseBody closeBody = + methodRegistry.createConnectionCloseBody(AMQConstant.NOT_ALLOWED.getCode(), // replyCode + AMQConstant.NOT_ALLOWED.getName(), + body.getClazz(), + body.getMethod()); + + session.writeFrame(closeBody.generateFrame(0)); disposeSaslServer(session); break; @@ -115,25 +111,17 @@ public class ConnectionStartOkMethodHandler implements StateAwareMethodListener< session.setAuthorizedID(new UsernamePrincipal(ss.getAuthorizationID())); stateManager.changeState(AMQState.CONNECTION_NOT_TUNED); - // AMQP version change: Hardwire the version to 0-8 (major=8, minor=0) - // TODO: Connect this to the session version obtained from ProtocolInitiation for this session. - // Be aware of possible changes to parameter order as versions change. - AMQFrame tune = ConnectionTuneBody.createAMQFrame(0, - (byte) 8, (byte) 0, // AMQP version (major, minor) - Integer.MAX_VALUE, // channelMax - getConfiguredFrameSize(), // frameMax - HeartbeatConfig.getInstance().getDelay()); // heartbeat - session.writeFrame(tune); + + ConnectionTuneBody tuneBody = methodRegistry.createConnectionTuneBody(0xFFFF, + getConfiguredFrameSize(), + HeartbeatConfig.getInstance().getDelay()); + session.writeFrame(tuneBody.generateFrame(0)); break; case CONTINUE: stateManager.changeState(AMQState.CONNECTION_NOT_AUTH); - // AMQP version change: Hardwire the version to 0-8 (major=8, minor=0) - // TODO: Connect this to the session version obtained from ProtocolInitiation for this session. - // Be aware of possible changes to parameter order as versions change. - AMQFrame challenge = ConnectionSecureBody.createAMQFrame(0, - (byte) 8, (byte) 0, // AMQP version (major, minor) - authResult.challenge); // challenge - session.writeFrame(challenge); + + ConnectionSecureBody secureBody = methodRegistry.createConnectionSecureBody(authResult.challenge); + session.writeFrame(secureBody.generateFrame(0)); } } catch (SaslException e) diff --git a/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionTuneOkMethodHandler.java b/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionTuneOkMethodHandler.java index ab7695955c..0fe8c5dc92 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionTuneOkMethodHandler.java +++ b/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionTuneOkMethodHandler.java @@ -40,15 +40,15 @@ public class ConnectionTuneOkMethodHandler implements StateAwareMethodListener<C return _instance; } - public void methodReceived(AMQStateManager stateManager, AMQMethodEvent<ConnectionTuneOkBody> evt) throws AMQException + public void methodReceived(AMQStateManager stateManager, ConnectionTuneOkBody body, int channelId) throws AMQException { AMQProtocolSession session = stateManager.getProtocolSession(); - ConnectionTuneOkBody body = evt.getMethod(); + if (_logger.isDebugEnabled()) { _logger.debug(body); } stateManager.changeState(AMQState.CONNECTION_NOT_OPENED); - session.initHeartbeats(body.heartbeat); + session.initHeartbeats(body.getHeartbeat()); } } diff --git a/java/broker/src/main/java/org/apache/qpid/server/handler/ExchangeBoundHandler.java b/java/broker/src/main/java/org/apache/qpid/server/handler/ExchangeBoundHandler.java index 162c03b025..491a2f80db 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/handler/ExchangeBoundHandler.java +++ b/java/broker/src/main/java/org/apache/qpid/server/handler/ExchangeBoundHandler.java @@ -21,10 +21,8 @@ package org.apache.qpid.server.handler; import org.apache.qpid.AMQException; -import org.apache.qpid.framing.AMQFrame; -import org.apache.qpid.framing.AMQShortString; -import org.apache.qpid.framing.ExchangeBoundBody; -import org.apache.qpid.framing.ExchangeBoundOkBody; +import org.apache.qpid.framing.*; +import org.apache.qpid.framing.amqp_8_0.MethodRegistry_8_0; import org.apache.qpid.protocol.AMQMethodEvent; import org.apache.qpid.server.exchange.Exchange; import org.apache.qpid.server.protocol.AMQProtocolSession; @@ -36,6 +34,8 @@ import org.apache.qpid.server.virtualhost.VirtualHost; /** * @author Apache Software Foundation + * + * */ public class ExchangeBoundHandler implements StateAwareMethodListener<ExchangeBoundBody> { @@ -64,35 +64,32 @@ public class ExchangeBoundHandler implements StateAwareMethodListener<ExchangeBo { } - public void methodReceived(AMQStateManager stateManager, AMQMethodEvent<ExchangeBoundBody> evt) throws AMQException + public void methodReceived(AMQStateManager stateManager, ExchangeBoundBody body, int channelId) throws AMQException { AMQProtocolSession session = stateManager.getProtocolSession(); + VirtualHost virtualHost = session.getVirtualHost(); QueueRegistry queueRegistry = virtualHost.getQueueRegistry(); + MethodRegistry methodRegistry = session.getMethodRegistry(); - // AMQP version change: Hardwire the version to 0-8 (major=8, minor=0) - // TODO: Connect this to the session version obtained from ProtocolInitiation for this session. - byte major = (byte)8; - byte minor = (byte)0; - ExchangeBoundBody body = evt.getMethod(); - AMQShortString exchangeName = body.exchange; - AMQShortString queueName = body.queue; - AMQShortString routingKey = body.routingKey; + + AMQShortString exchangeName = body.getExchange(); + AMQShortString queueName = body.getQueue(); + AMQShortString routingKey = body.getRoutingKey(); if (exchangeName == null) { throw new AMQException("Exchange exchange must not be null"); } Exchange exchange = virtualHost.getExchangeRegistry().getExchange(exchangeName); - AMQFrame response; + ExchangeBoundOkBody response; if (exchange == null) { - // AMQP version change: Be aware of possible changes to parameter order as versions change. - response = ExchangeBoundOkBody.createAMQFrame(evt.getChannelId(), - major, minor, // AMQP version (major, minor) - EXCHANGE_NOT_FOUND, // replyCode - new AMQShortString("Exchange " + exchangeName + " not found")); // replyText + + + response = methodRegistry.createExchangeBoundOkBody(EXCHANGE_NOT_FOUND, + new AMQShortString("Exchange " + exchangeName + " not found")); } else if (routingKey == null) { @@ -100,18 +97,12 @@ public class ExchangeBoundHandler implements StateAwareMethodListener<ExchangeBo { if (exchange.hasBindings()) { - // AMQP version change: Be aware of possible changes to parameter order as versions change. - response = ExchangeBoundOkBody.createAMQFrame(evt.getChannelId(), - major, minor, // AMQP version (major, minor) - OK, // replyCode - null); // replyText + response = methodRegistry.createExchangeBoundOkBody(OK, null); } else { - // AMQP version change: Be aware of possible changes to parameter order as versions change. - response = ExchangeBoundOkBody.createAMQFrame(evt.getChannelId(), - major, minor, // AMQP version (major, minor) - NO_BINDINGS, // replyCode + + response = methodRegistry.createExchangeBoundOkBody(NO_BINDINGS, // replyCode null); // replyText } } @@ -121,28 +112,22 @@ public class ExchangeBoundHandler implements StateAwareMethodListener<ExchangeBo AMQQueue queue = queueRegistry.getQueue(queueName); if (queue == null) { - // AMQP version change: Be aware of possible changes to parameter order as versions change. - response = ExchangeBoundOkBody.createAMQFrame(evt.getChannelId(), - major, minor, // AMQP version (major, minor) - QUEUE_NOT_FOUND, // replyCode + + response = methodRegistry.createExchangeBoundOkBody(QUEUE_NOT_FOUND, // replyCode new AMQShortString("Queue " + queueName + " not found")); // replyText } else { if (exchange.isBound(queue)) { - // AMQP version change: Be aware of possible changes to parameter order as versions change. - response = ExchangeBoundOkBody.createAMQFrame(evt.getChannelId(), - major, minor, // AMQP version (major, minor) - OK, // replyCode + + response = methodRegistry.createExchangeBoundOkBody(OK, // replyCode null); // replyText } else { - // AMQP version change: Be aware of possible changes to parameter order as versions change. - response = ExchangeBoundOkBody.createAMQFrame(evt.getChannelId(), - major, minor, // AMQP version (major, minor) - QUEUE_NOT_BOUND, // replyCode + + response = methodRegistry.createExchangeBoundOkBody(QUEUE_NOT_BOUND, // replyCode new AMQShortString("Queue " + queueName + " not bound to exchange " + exchangeName)); // replyText } } @@ -153,53 +138,43 @@ public class ExchangeBoundHandler implements StateAwareMethodListener<ExchangeBo AMQQueue queue = queueRegistry.getQueue(queueName); if (queue == null) { - // AMQP version change: Be aware of possible changes to parameter order as versions change. - response = ExchangeBoundOkBody.createAMQFrame(evt.getChannelId(), - major, minor, // AMQP version (major, minor) - QUEUE_NOT_FOUND, // replyCode + + response = methodRegistry.createExchangeBoundOkBody(QUEUE_NOT_FOUND, // replyCode new AMQShortString("Queue " + queueName + " not found")); // replyText } else { - if (exchange.isBound(body.routingKey, queue)) + if (exchange.isBound(body.getRoutingKey(), queue)) { - // AMQP version change: Be aware of possible changes to parameter order as versions change. - response = ExchangeBoundOkBody.createAMQFrame(evt.getChannelId(), - major, minor, // AMQP version (major, minor) - OK, // replyCode + + response = methodRegistry.createExchangeBoundOkBody(OK, // replyCode null); // replyText } else { - // AMQP version change: Be aware of possible changes to parameter order as versions change. - response = ExchangeBoundOkBody.createAMQFrame(evt.getChannelId(), - major, minor, // AMQP version (major, minor) - SPECIFIC_QUEUE_NOT_BOUND_WITH_RK, // replyCode + + response = methodRegistry.createExchangeBoundOkBody(SPECIFIC_QUEUE_NOT_BOUND_WITH_RK, // replyCode new AMQShortString("Queue " + queueName + " not bound with routing key " + - body.routingKey + " to exchange " + exchangeName)); // replyText + body.getRoutingKey() + " to exchange " + exchangeName)); // replyText } } } else { - if (exchange.isBound(body.routingKey)) + if (exchange.isBound(body.getRoutingKey())) { - // AMQP version change: Be aware of possible changes to parameter order as versions change. - response = ExchangeBoundOkBody.createAMQFrame(evt.getChannelId(), - major, minor, // AMQP version (major, minor) - OK, // replyCode + + response = methodRegistry.createExchangeBoundOkBody(OK, // replyCode null); // replyText } else { - // AMQP version change: Be aware of possible changes to parameter order as versions change. - response = ExchangeBoundOkBody.createAMQFrame(evt.getChannelId(), - major, minor, // AMQP version (major, minor) - NO_QUEUE_BOUND_WITH_RK, // replyCode - new AMQShortString("No queue bound with routing key " + body.routingKey + + + response = methodRegistry.createExchangeBoundOkBody(NO_QUEUE_BOUND_WITH_RK, // replyCode + new AMQShortString("No queue bound with routing key " + body.getRoutingKey() + " to exchange " + exchangeName)); // replyText } } - session.writeFrame(response); + session.writeFrame(response.generateFrame(channelId)); } } diff --git a/java/broker/src/main/java/org/apache/qpid/server/handler/ExchangeDeclareHandler.java b/java/broker/src/main/java/org/apache/qpid/server/handler/ExchangeDeclareHandler.java index 8ab8366d9e..c4a57f0286 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/handler/ExchangeDeclareHandler.java +++ b/java/broker/src/main/java/org/apache/qpid/server/handler/ExchangeDeclareHandler.java @@ -24,9 +24,7 @@ import org.apache.log4j.Logger; import org.apache.qpid.AMQConnectionException; import org.apache.qpid.AMQException; import org.apache.qpid.AMQUnknownExchangeType; -import org.apache.qpid.framing.AMQFrame; -import org.apache.qpid.framing.ExchangeDeclareBody; -import org.apache.qpid.framing.ExchangeDeclareOkBody; +import org.apache.qpid.framing.*; import org.apache.qpid.protocol.AMQConstant; import org.apache.qpid.protocol.AMQMethodEvent; import org.apache.qpid.server.exchange.Exchange; @@ -54,61 +52,60 @@ public class ExchangeDeclareHandler implements StateAwareMethodListener<Exchange { } - public void methodReceived(AMQStateManager stateManager, AMQMethodEvent<ExchangeDeclareBody> evt) throws AMQException + public void methodReceived(AMQStateManager stateManager, ExchangeDeclareBody body, int channelId) throws AMQException { AMQProtocolSession session = stateManager.getProtocolSession(); VirtualHost virtualHost = session.getVirtualHost(); ExchangeRegistry exchangeRegistry = virtualHost.getExchangeRegistry(); ExchangeFactory exchangeFactory = virtualHost.getExchangeFactory(); - final ExchangeDeclareBody body = evt.getMethod(); + if (_logger.isDebugEnabled()) { - _logger.debug("Request to declare exchange of type " + body.type + " with name " + body.exchange); + _logger.debug("Request to declare exchange of type " + body.getType() + " with name " + body.getExchange()); } synchronized(exchangeRegistry) { - Exchange exchange = exchangeRegistry.getExchange(body.exchange); + Exchange exchange = exchangeRegistry.getExchange(body.getExchange()); if (exchange == null) { - if(body.passive && ((body.type == null) || body.type.length() ==0)) + if(body.getPassive() && ((body.getType() == null) || body.getType().length() ==0)) { - throw body.getChannelException(AMQConstant.NOT_FOUND, "Unknown exchange: " + body.exchange); + throw body.getChannelException(AMQConstant.NOT_FOUND, "Unknown exchange: " + body.getExchange()); } else { try { - exchange = exchangeFactory.createExchange(body.exchange == null ? null : body.exchange.intern(), - body.type == null ? null : body.type.intern(), - body.durable, - body.passive, body.ticket); + exchange = exchangeFactory.createExchange(body.getExchange() == null ? null : body.getExchange().intern(), + body.getType() == null ? null : body.getType().intern(), + body.getDurable(), + body.getPassive(), body.getTicket()); exchangeRegistry.registerExchange(exchange); } catch(AMQUnknownExchangeType e) { - throw body.getConnectionException(AMQConstant.COMMAND_INVALID, "Unknown exchange: " + body.exchange,e); + throw body.getConnectionException(AMQConstant.COMMAND_INVALID, "Unknown exchange: " + body.getExchange(),e); } } } - else if (!exchange.getType().equals(body.type)) + else if (!exchange.getType().equals(body.getType())) { - throw new AMQConnectionException(AMQConstant.NOT_ALLOWED, "Attempt to redeclare exchange: " + body.exchange + " of type " + exchange.getType() + " to " + body.type +".",body.getClazz(), body.getMethod(),body.getMajor(),body.getMinor()); + throw new AMQConnectionException(AMQConstant.NOT_ALLOWED, "Attempt to redeclare exchange: " + body.getExchange() + " of type " + exchange.getType() + " to " + body.getType() +".",body.getClazz(), body.getMethod(),body.getMajor(),body.getMinor()); } } - if(!body.nowait) + if(!body.getNowait()) { - // AMQP version change: Hardwire the version to 0-8 (major=8, minor=0) - // TODO: Connect this to the session version obtained from ProtocolInitiation for this session. - // Be aware of possible changes to parameter order as versions change. - AMQFrame response = ExchangeDeclareOkBody.createAMQFrame(evt.getChannelId(), (byte)8, (byte)0); - session.writeFrame(response); + MethodRegistry methodRegistry = session.getMethodRegistry(); + AMQMethodBody responseBody = methodRegistry.createExchangeDeclareOkBody(); + session.writeFrame(responseBody.generateFrame(channelId)); + } } } diff --git a/java/broker/src/main/java/org/apache/qpid/server/handler/ExchangeDeleteHandler.java b/java/broker/src/main/java/org/apache/qpid/server/handler/ExchangeDeleteHandler.java index f9926c399c..8f36e6c767 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/handler/ExchangeDeleteHandler.java +++ b/java/broker/src/main/java/org/apache/qpid/server/handler/ExchangeDeleteHandler.java @@ -45,21 +45,20 @@ public class ExchangeDeleteHandler implements StateAwareMethodListener<ExchangeD { } - public void methodReceived(AMQStateManager stateManager, AMQMethodEvent<ExchangeDeleteBody> evt) throws AMQException + public void methodReceived(AMQStateManager stateManager, ExchangeDeleteBody body, int channelId) throws AMQException { AMQProtocolSession session = stateManager.getProtocolSession(); VirtualHost virtualHost = session.getVirtualHost(); ExchangeRegistry exchangeRegistry = virtualHost.getExchangeRegistry(); - ExchangeDeleteBody body = evt.getMethod(); + try { - exchangeRegistry.unregisterExchange(body.exchange, body.ifUnused); - // AMQP version change: Hardwire the version to 0-8 (major=8, minor=0) - // TODO: Connect this to the session version obtained from ProtocolInitiation for this session. - // Be aware of possible changes to parameter order as versions change. - AMQFrame response = ExchangeDeleteOkBody.createAMQFrame(evt.getChannelId(), (byte)8, (byte)0); - session.writeFrame(response); + exchangeRegistry.unregisterExchange(body.getExchange(), body.getIfUnused()); + + ExchangeDeleteOkBody responseBody = session.getMethodRegistry().createExchangeDeleteOkBody(); + + session.writeFrame(responseBody.generateFrame(channelId)); } catch (ExchangeInUseException e) { diff --git a/java/broker/src/main/java/org/apache/qpid/server/handler/QueueBindHandler.java b/java/broker/src/main/java/org/apache/qpid/server/handler/QueueBindHandler.java index 3e68069838..057586a143 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/handler/QueueBindHandler.java +++ b/java/broker/src/main/java/org/apache/qpid/server/handler/QueueBindHandler.java @@ -23,9 +23,7 @@ package org.apache.qpid.server.handler; import org.apache.log4j.Logger; import org.apache.qpid.AMQException; import org.apache.qpid.AMQInvalidRoutingKeyException; -import org.apache.qpid.framing.AMQFrame; -import org.apache.qpid.framing.QueueBindBody; -import org.apache.qpid.framing.QueueBindOkBody; +import org.apache.qpid.framing.*; import org.apache.qpid.protocol.AMQConstant; import org.apache.qpid.protocol.AMQMethodEvent; import org.apache.qpid.server.AMQChannel; @@ -53,22 +51,24 @@ public class QueueBindHandler implements StateAwareMethodListener<QueueBindBody> { } - public void methodReceived(AMQStateManager stateManager, AMQMethodEvent<QueueBindBody> evt) throws AMQException + public void methodReceived(AMQStateManager stateManager, QueueBindBody body, int channelId) throws AMQException { AMQProtocolSession session = stateManager.getProtocolSession(); VirtualHost virtualHost = session.getVirtualHost(); ExchangeRegistry exchangeRegistry = virtualHost.getExchangeRegistry(); QueueRegistry queueRegistry = virtualHost.getQueueRegistry(); - final QueueBindBody body = evt.getMethod(); + final AMQQueue queue; - if (body.queue == null) + final AMQShortString routingKey; + + if (body.getQueue() == null) { - AMQChannel channel = session.getChannel(evt.getChannelId()); + AMQChannel channel = session.getChannel(channelId); if (channel == null) { - throw body.getChannelNotFoundException(evt.getChannelId()); + throw body.getChannelNotFoundException(channelId); } queue = channel.getDefaultQueue(); @@ -78,41 +78,42 @@ public class QueueBindHandler implements StateAwareMethodListener<QueueBindBody> throw body.getChannelException(AMQConstant.NOT_FOUND, "No default queue defined on channel and queue was null"); } - if (body.routingKey == null) + if (body.getRoutingKey() == null) + { + routingKey = queue.getName(); + } + else { - body.routingKey = queue.getName(); + routingKey = body.getRoutingKey().intern(); } } else { - queue = queueRegistry.getQueue(body.queue); + queue = queueRegistry.getQueue(body.getQueue()); + routingKey = body.getRoutingKey() == null ? null : body.getRoutingKey().intern(); } if (queue == null) { - throw body.getChannelException(AMQConstant.NOT_FOUND, "Queue " + body.queue + " does not exist."); + throw body.getChannelException(AMQConstant.NOT_FOUND, "Queue " + body.getQueue() + " does not exist."); } - final Exchange exch = exchangeRegistry.getExchange(body.exchange); + final Exchange exch = exchangeRegistry.getExchange(body.getExchange()); if (exch == null) { - throw body.getChannelException(AMQConstant.NOT_FOUND, "Exchange " + body.exchange + " does not exist."); + throw body.getChannelException(AMQConstant.NOT_FOUND, "Exchange " + body.getExchange() + " does not exist."); } - if (body.routingKey != null) - { - body.routingKey = body.routingKey.intern(); - } try { - if (!exch.isBound(body.routingKey, body.arguments, queue)) + if (!exch.isBound(routingKey, body.getArguments(), queue)) { - queue.bind(body.routingKey, body.arguments, exch); + queue.bind(routingKey, body.getArguments(), exch); } } catch (AMQInvalidRoutingKeyException rke) { - throw body.getChannelException(AMQConstant.INVALID_ROUTING_KEY, body.routingKey.toString()); + throw body.getChannelException(AMQConstant.INVALID_ROUTING_KEY, routingKey.toString()); } catch (AMQException e) { @@ -121,15 +122,14 @@ public class QueueBindHandler implements StateAwareMethodListener<QueueBindBody> if (_log.isInfoEnabled()) { - _log.info("Binding queue " + queue + " to exchange " + exch + " with routing key " + body.routingKey); + _log.info("Binding queue " + queue + " to exchange " + exch + " with routing key " + routingKey); } - if (!body.nowait) + if (!body.getNowait()) { - // AMQP version change: Hardwire the version to 0-8 (major=8, minor=0) - // TODO: Connect this to the session version obtained from ProtocolInitiation for this session. - // Be aware of possible changes to parameter order as versions change. - final AMQFrame response = QueueBindOkBody.createAMQFrame(evt.getChannelId(), (byte) 8, (byte) 0); - session.writeFrame(response); + MethodRegistry methodRegistry = session.getMethodRegistry(); + AMQMethodBody responseBody = methodRegistry.createQueueBindOkBody(); + session.writeFrame(responseBody.generateFrame(channelId)); + } } } diff --git a/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeclareHandler.java b/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeclareHandler.java index 29697542be..f59fcf2a12 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeclareHandler.java +++ b/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeclareHandler.java @@ -28,10 +28,7 @@ import org.apache.log4j.Logger; import org.apache.qpid.AMQException; import org.apache.qpid.configuration.Configured; import org.apache.qpid.exchange.ExchangeDefaults; -import org.apache.qpid.framing.AMQFrame; -import org.apache.qpid.framing.AMQShortString; -import org.apache.qpid.framing.QueueDeclareBody; -import org.apache.qpid.framing.QueueDeclareOkBody; +import org.apache.qpid.framing.*; import org.apache.qpid.protocol.AMQConstant; import org.apache.qpid.protocol.AMQMethodEvent; import org.apache.qpid.server.configuration.Configurator; @@ -70,7 +67,7 @@ public class QueueDeclareHandler implements StateAwareMethodListener<QueueDeclar Configurator.configure(this); } - public void methodReceived(AMQStateManager stateManager, AMQMethodEvent<QueueDeclareBody> evt) throws AMQException + public void methodReceived(AMQStateManager stateManager, QueueDeclareBody body, int channelId) throws AMQException { AMQProtocolSession session = stateManager.getProtocolSession(); VirtualHost virtualHost = session.getVirtualHost(); @@ -78,12 +75,20 @@ public class QueueDeclareHandler implements StateAwareMethodListener<QueueDeclar QueueRegistry queueRegistry = virtualHost.getQueueRegistry(); MessageStore store = virtualHost.getMessageStore(); - QueueDeclareBody body = evt.getMethod(); + + + + final AMQShortString queueName; // if we aren't given a queue name, we create one which we return to the client - if (body.queue == null) + + if (body.getQueue() == null) { - body.queue = createName(); + queueName = createName(); + } + else + { + queueName = body.getQueue().intern(); } AMQQueue queue; @@ -94,16 +99,12 @@ public class QueueDeclareHandler implements StateAwareMethodListener<QueueDeclar - if (((queue = queueRegistry.getQueue(body.queue)) == null)) + if (((queue = queueRegistry.getQueue(queueName)) == null)) { - if(body.queue != null) - { - body.queue = body.queue.intern(); - } - if (body.passive) + if (body.getPassive()) { - String msg = "Queue: " + body.queue + " not found on VirtualHost(" + virtualHost + ")."; + String msg = "Queue: " + queueName + " not found on VirtualHost(" + virtualHost + ")."; throw body.getChannelException(AMQConstant.NOT_FOUND, msg); } else @@ -118,42 +119,40 @@ public class QueueDeclareHandler implements StateAwareMethodListener<QueueDeclar { Exchange defaultExchange = exchangeRegistry.getDefaultExchange(); - queue.bind(body.queue, null, defaultExchange); - _log.info("Queue " + body.queue + " bound to default exchange(" + defaultExchange.getName() + ")"); + queue.bind(queueName, null, defaultExchange); + _log.info("Queue " + queueName + " bound to default exchange(" + defaultExchange.getName() + ")"); } } } else if (queue.getOwner() != null && !session.getContextKey().equals(queue.getOwner())) { - throw body.getChannelException(AMQConstant.ALREADY_EXISTS, "Cannot declare queue('" + body.queue + "')," + throw body.getChannelException(AMQConstant.ALREADY_EXISTS, "Cannot declare queue('" + queueName + "')," + " as exclusive queue with same name " + "declared on another client ID('" + queue.getOwner() + "')"); } - AMQChannel channel = session.getChannel(evt.getChannelId()); + AMQChannel channel = session.getChannel(channelId); if (channel == null) { - throw body.getChannelNotFoundException(evt.getChannelId()); + throw body.getChannelNotFoundException(channelId); } //set this as the default queue on the channel: channel.setDefaultQueue(queue); } - if (!body.nowait) + if (!body.getNowait()) { - // AMQP version change: Hardwire the version to 0-8 (major=8, minor=0) - // TODO: Connect this to the session version obtained from ProtocolInitiation for this session. - // Be aware of possible changes to parameter order as versions change. - AMQFrame response = QueueDeclareOkBody.createAMQFrame(evt.getChannelId(), - (byte) 8, (byte) 0, // AMQP version (major, minor) - queue.getConsumerCount(), // consumerCount - queue.getMessageCount(), // messageCount - body.queue); // queue - _log.info("Queue " + body.queue + " declared successfully"); - session.writeFrame(response); + MethodRegistry methodRegistry = session.getMethodRegistry(); + QueueDeclareOkBody responseBody = + methodRegistry.createQueueDeclareOkBody(queueName, + queue.getMessageCount(), + queue.getConsumerCount()); + session.writeFrame(responseBody.generateFrame(channelId)); + + _log.info("Queue " + queueName + " declared successfully"); } } @@ -166,11 +165,11 @@ public class QueueDeclareHandler implements StateAwareMethodListener<QueueDeclar throws AMQException { final QueueRegistry registry = virtualHost.getQueueRegistry(); - AMQShortString owner = body.exclusive ? session.getContextKey() : null; - final AMQQueue queue = new AMQQueue(body.queue, body.durable, owner, body.autoDelete, virtualHost); + AMQShortString owner = body.getExclusive() ? session.getContextKey() : null; + final AMQQueue queue = new AMQQueue(body.getQueue(), body.getDurable(), owner, body.getAutoDelete(), virtualHost); final AMQShortString queueName = queue.getName(); - if (body.exclusive && !body.durable) + if (body.getExclusive() && !body.getDurable()) { final AMQProtocolSession.Task deleteQueueTask = new AMQProtocolSession.Task() diff --git a/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeleteHandler.java b/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeleteHandler.java index eb7089afdc..09b7de3a5c 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeleteHandler.java +++ b/java/broker/src/main/java/org/apache/qpid/server/handler/QueueDeleteHandler.java @@ -23,6 +23,7 @@ package org.apache.qpid.server.handler; import org.apache.qpid.AMQException; import org.apache.qpid.framing.QueueDeleteBody; import org.apache.qpid.framing.QueueDeleteOkBody; +import org.apache.qpid.framing.MethodRegistry; import org.apache.qpid.protocol.AMQMethodEvent; import org.apache.qpid.protocol.AMQConstant; import org.apache.qpid.server.protocol.AMQProtocolSession; @@ -56,22 +57,21 @@ public class QueueDeleteHandler implements StateAwareMethodListener<QueueDeleteB } - public void methodReceived(AMQStateManager stateManager, AMQMethodEvent<QueueDeleteBody> evt) throws AMQException + public void methodReceived(AMQStateManager stateManager, QueueDeleteBody body, int channelId) throws AMQException { AMQProtocolSession session = stateManager.getProtocolSession(); VirtualHost virtualHost = session.getVirtualHost(); QueueRegistry queueRegistry = virtualHost.getQueueRegistry(); MessageStore store = virtualHost.getMessageStore(); - QueueDeleteBody body = evt.getMethod(); AMQQueue queue; - if (body.queue == null) + if (body.getQueue() == null) { - AMQChannel channel = session.getChannel(evt.getChannelId()); + AMQChannel channel = session.getChannel(channelId); if (channel == null) { - throw body.getChannelNotFoundException(evt.getChannelId()); + throw body.getChannelNotFoundException(channelId); } //get the default queue on the channel: @@ -79,43 +79,40 @@ public class QueueDeleteHandler implements StateAwareMethodListener<QueueDeleteB } else { - queue = queueRegistry.getQueue(body.queue); + queue = queueRegistry.getQueue(body.getQueue()); } if (queue == null) { if (_failIfNotFound) { - throw body.getChannelException(AMQConstant.NOT_FOUND, "Queue " + body.queue + " does not exist."); + throw body.getChannelException(AMQConstant.NOT_FOUND, "Queue " + body.getQueue() + " does not exist."); } } else { - if (body.ifEmpty && !queue.isEmpty()) + if (body.getIfEmpty() && !queue.isEmpty()) { - throw body.getChannelException(AMQConstant.IN_USE, "Queue: " + body.queue + " is not empty."); + throw body.getChannelException(AMQConstant.IN_USE, "Queue: " + body.getQueue() + " is not empty."); } - else if (body.ifUnused && !queue.isUnused()) + else if (body.getIfUnused() && !queue.isUnused()) { // TODO - Error code - throw body.getChannelException(AMQConstant.IN_USE, "Queue: " + body.queue + " is still used."); + throw body.getChannelException(AMQConstant.IN_USE, "Queue: " + body.getQueue() + " is still used."); } else { - int purged = queue.delete(body.ifUnused, body.ifEmpty); + int purged = queue.delete(body.getIfUnused(), body.getIfEmpty()); if (queue.isDurable()) { store.removeQueue(queue.getName()); } - - // AMQP version change: Hardwire the version to 0-8 (major=8, minor=0) - // TODO: Connect this to the session version obtained from ProtocolInitiation for this session. - // Be aware of possible changes to parameter order as versions change. - session.writeFrame(QueueDeleteOkBody.createAMQFrame(evt.getChannelId(), - (byte) 8, (byte) 0, // AMQP version (major, minor) - purged)); // messageCount + + MethodRegistry methodRegistry = session.getMethodRegistry(); + QueueDeleteOkBody responseBody = methodRegistry.createQueueDeleteOkBody(purged); + session.writeFrame(responseBody.generateFrame(channelId)); } } } diff --git a/java/broker/src/main/java/org/apache/qpid/server/handler/QueuePurgeHandler.java b/java/broker/src/main/java/org/apache/qpid/server/handler/QueuePurgeHandler.java index 3e1937bb43..5bdca93bc6 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/handler/QueuePurgeHandler.java +++ b/java/broker/src/main/java/org/apache/qpid/server/handler/QueuePurgeHandler.java @@ -24,6 +24,8 @@ package org.apache.qpid.server.handler; import org.apache.qpid.AMQException;
import org.apache.qpid.framing.QueuePurgeBody;
import org.apache.qpid.framing.QueuePurgeOkBody;
+import org.apache.qpid.framing.MethodRegistry;
+import org.apache.qpid.framing.AMQMethodBody;
import org.apache.qpid.protocol.AMQConstant;
import org.apache.qpid.protocol.AMQMethodEvent;
import org.apache.qpid.server.protocol.AMQProtocolSession;
@@ -55,22 +57,22 @@ public class QueuePurgeHandler implements StateAwareMethodListener<QueuePurgeBod _failIfNotFound = failIfNotFound;
}
- public void methodReceived(AMQStateManager stateManager, AMQMethodEvent<QueuePurgeBody> evt) throws AMQException
+ public void methodReceived(AMQStateManager stateManager, QueuePurgeBody body, int channelId) throws AMQException
{
AMQProtocolSession session = stateManager.getProtocolSession();
VirtualHost virtualHost = session.getVirtualHost();
QueueRegistry queueRegistry = virtualHost.getQueueRegistry();
- AMQChannel channel = session.getChannel(evt.getChannelId());
+ AMQChannel channel = session.getChannel(channelId);
+
- QueuePurgeBody body = evt.getMethod();
AMQQueue queue;
- if(body.queue == null)
+ if(body.getQueue() == null)
{
if (channel == null)
{
- throw body.getChannelNotFoundException(evt.getChannelId());
+ throw body.getChannelNotFoundException(channelId);
}
//get the default queue on the channel:
@@ -86,14 +88,14 @@ public class QueuePurgeHandler implements StateAwareMethodListener<QueuePurgeBod }
else
{
- queue = queueRegistry.getQueue(body.queue);
+ queue = queueRegistry.getQueue(body.getQueue());
}
if(queue == null)
{
if(_failIfNotFound)
{
- throw body.getChannelException(AMQConstant.NOT_FOUND, "Queue " + body.queue + " does not exist.");
+ throw body.getChannelException(AMQConstant.NOT_FOUND, "Queue " + body.getQueue() + " does not exist.");
}
}
else
@@ -101,14 +103,13 @@ public class QueuePurgeHandler implements StateAwareMethodListener<QueuePurgeBod long purged = queue.clearQueue(channel.getStoreContext());
- if(!body.nowait)
+ if(!body.getNowait())
{
- // AMQP version change: Hardwire the version to 0-8 (major=8, minor=0)
- // TODO: Connect this to the session version obtained from ProtocolInitiation for this session.
- // Be aware of possible changes to parameter order as versions change.
- session.writeFrame(QueuePurgeOkBody.createAMQFrame(evt.getChannelId(),
- (byte)8, (byte)0, // AMQP version (major, minor)
- purged)); // messageCount
+
+ MethodRegistry methodRegistry = session.getMethodRegistry();
+ AMQMethodBody responseBody = methodRegistry.createQueuePurgeOkBody(purged);
+ session.writeFrame(responseBody.generateFrame(channelId));
+
}
}
}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/handler/ServerMethodDispatcherImpl.java b/java/broker/src/main/java/org/apache/qpid/server/handler/ServerMethodDispatcherImpl.java new file mode 100644 index 0000000000..9475b83c8f --- /dev/null +++ b/java/broker/src/main/java/org/apache/qpid/server/handler/ServerMethodDispatcherImpl.java @@ -0,0 +1,566 @@ +/*
+ *
+ * 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.handler;
+
+import java.util.Map;
+import java.util.HashMap;
+
+import org.apache.qpid.server.state.AMQStateManager;
+import org.apache.qpid.framing.*;
+import org.apache.qpid.AMQException;
+
+public class ServerMethodDispatcherImpl implements MethodDispatcher
+{
+ private final AMQStateManager _stateManager;
+
+ private static interface DispatcherFactory
+ {
+ public MethodDispatcher createMethodDispatcher(AMQStateManager stateManager);
+ }
+
+ private static final Map<ProtocolVersion, DispatcherFactory> _dispatcherFactories =
+ new HashMap<ProtocolVersion, DispatcherFactory>();
+
+
+ static
+ {
+ _dispatcherFactories.put(ProtocolVersion.v8_0,
+ new DispatcherFactory()
+ {
+ public MethodDispatcher createMethodDispatcher(AMQStateManager stateManager)
+ {
+ return new ServerMethodDispatcherImpl_8_0(stateManager);
+ }
+ });
+
+ _dispatcherFactories.put(ProtocolVersion.v0_9,
+ new DispatcherFactory()
+ {
+ public MethodDispatcher createMethodDispatcher(AMQStateManager stateManager)
+ {
+ return new ServerMethodDispatcherImpl_0_9(stateManager);
+ }
+ });
+
+ }
+
+
+ private static final AccessRequestHandler _accessRequestHandler = AccessRequestHandler.getInstance();
+ private static final ChannelCloseHandler _channelCloseHandler = ChannelCloseHandler.getInstance();
+ private static final ChannelOpenHandler _channelOpenHandler = ChannelOpenHandler.getInstance();
+ private static final ChannelCloseOkHandler _channelCloseOkHandler = ChannelCloseOkHandler.getInstance();
+ private static final ConnectionCloseMethodHandler _connectionCloseMethodHandler = ConnectionCloseMethodHandler.getInstance();
+ private static final ConnectionCloseOkMethodHandler _connectionCloseOkMethodHandler = ConnectionCloseOkMethodHandler.getInstance();
+ private static final ConnectionOpenMethodHandler _connectionOpenMethodHandler = ConnectionOpenMethodHandler.getInstance();
+ private static final ConnectionTuneOkMethodHandler _connectionTuneOkMethodHandler = ConnectionTuneOkMethodHandler.getInstance();
+ private static final ConnectionSecureOkMethodHandler _connectionSecureOkMethodHandler = ConnectionSecureOkMethodHandler.getInstance();
+ private static final ConnectionStartOkMethodHandler _connectionStartOkMethodHandler = ConnectionStartOkMethodHandler.getInstance();
+ private static final ExchangeDeclareHandler _exchangeDeclareHandler = ExchangeDeclareHandler.getInstance();
+ private static final ExchangeDeleteHandler _exchangeDeleteHandler = ExchangeDeleteHandler.getInstance();
+ private static final ExchangeBoundHandler _exchangeBoundHandler = ExchangeBoundHandler.getInstance();
+ private static final BasicAckMethodHandler _basicAckMethodHandler = BasicAckMethodHandler.getInstance();
+ private static final BasicRecoverMethodHandler _basicRecoverMethodHandler = BasicRecoverMethodHandler.getInstance();
+ private static final BasicConsumeMethodHandler _basicConsumeMethodHandler = BasicConsumeMethodHandler.getInstance();
+ private static final BasicGetMethodHandler _basicGetMethodHandler = BasicGetMethodHandler.getInstance();
+ private static final BasicCancelMethodHandler _basicCancelMethodHandler = BasicCancelMethodHandler.getInstance();
+ private static final BasicPublishMethodHandler _basicPublishMethodHandler = BasicPublishMethodHandler.getInstance();
+ private static final BasicQosHandler _basicQosHandler = BasicQosHandler.getInstance();
+ private static final QueueBindHandler _queueBindHandler = QueueBindHandler.getInstance();
+ private static final QueueDeclareHandler _queueDeclareHandler = QueueDeclareHandler.getInstance();
+ private static final QueueDeleteHandler _queueDeleteHandler = QueueDeleteHandler.getInstance();
+ private static final QueuePurgeHandler _queuePurgeHandler = QueuePurgeHandler.getInstance();
+ private static final ChannelFlowHandler _channelFlowHandler = ChannelFlowHandler.getInstance();
+ private static final TxSelectHandler _txSelectHandler = TxSelectHandler.getInstance();
+ private static final TxCommitHandler _txCommitHandler = TxCommitHandler.getInstance();
+ private static final TxRollbackHandler _txRollbackHandler = TxRollbackHandler.getInstance();
+ private static final BasicRejectMethodHandler _basicRejectMethodHandler = BasicRejectMethodHandler.getInstance();
+
+
+
+ public static MethodDispatcher createMethodDispatcher(AMQStateManager stateManager, ProtocolVersion protocolVersion)
+ {
+ return _dispatcherFactories.get(protocolVersion).createMethodDispatcher(stateManager);
+ }
+
+
+ public ServerMethodDispatcherImpl(AMQStateManager stateManager)
+ {
+ _stateManager = stateManager;
+ }
+
+
+ protected AMQStateManager getStateManager()
+ {
+ return _stateManager;
+ }
+
+
+
+ public boolean dispatchAccessRequest(AccessRequestBody body, int channelId) throws AMQException
+ {
+ _accessRequestHandler.methodReceived(_stateManager, body, channelId);
+ return true;
+ }
+
+ public boolean dispatchBasicAck(BasicAckBody body, int channelId) throws AMQException
+ {
+ _basicAckMethodHandler.methodReceived(_stateManager, body, channelId);
+ return true;
+ }
+
+ public boolean dispatchBasicCancel(BasicCancelBody body, int channelId) throws AMQException
+ {
+ _basicCancelMethodHandler.methodReceived(_stateManager, body, channelId);
+ return true;
+ }
+
+ public boolean dispatchBasicConsume(BasicConsumeBody body, int channelId) throws AMQException
+ {
+ _basicConsumeMethodHandler.methodReceived(_stateManager, body, channelId);
+ return true;
+ }
+
+ public boolean dispatchBasicGet(BasicGetBody body, int channelId) throws AMQException
+ {
+ _basicGetMethodHandler.methodReceived(_stateManager, body, channelId);
+ return true;
+ }
+
+ public boolean dispatchBasicPublish(BasicPublishBody body, int channelId) throws AMQException
+ {
+ _basicPublishMethodHandler.methodReceived(_stateManager, body, channelId);
+ return true;
+ }
+
+ public boolean dispatchBasicQos(BasicQosBody body, int channelId) throws AMQException
+ {
+ _basicQosHandler.methodReceived(_stateManager, body, channelId);
+ return true;
+ }
+
+ public boolean dispatchBasicRecover(BasicRecoverBody body, int channelId) throws AMQException
+ {
+ _basicRecoverMethodHandler.methodReceived(_stateManager, body, channelId);
+ return true;
+ }
+
+ public boolean dispatchBasicReject(BasicRejectBody body, int channelId) throws AMQException
+ {
+ _basicRejectMethodHandler.methodReceived(_stateManager, body, channelId);
+ return true;
+ }
+
+ public boolean dispatchChannelOpen(ChannelOpenBody body, int channelId) throws AMQException
+ {
+ _channelOpenHandler.methodReceived(_stateManager, body, 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
+ {
+ _channelCloseHandler.methodReceived(_stateManager, body, channelId);
+ return true;
+ }
+
+
+ public boolean dispatchChannelCloseOk(ChannelCloseOkBody body, int channelId) throws AMQException
+ {
+ _channelCloseOkHandler.methodReceived(_stateManager, body, channelId);
+ return true;
+ }
+
+
+ public boolean dispatchChannelFlow(ChannelFlowBody body, int channelId) throws AMQException
+ {
+ _channelFlowHandler.methodReceived(_stateManager, body, 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
+ {
+ _connectionOpenMethodHandler.methodReceived(_stateManager, body, channelId);
+ return true;
+ }
+
+
+ public boolean dispatchConnectionClose(ConnectionCloseBody body, int channelId) throws AMQException
+ {
+ _connectionCloseMethodHandler.methodReceived(_stateManager, body, channelId);
+ return true;
+ }
+
+
+ public boolean dispatchConnectionCloseOk(ConnectionCloseOkBody body, int channelId) throws AMQException
+ {
+ _connectionCloseOkMethodHandler.methodReceived(_stateManager, body, channelId);
+ 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 dispatchDtxSelectOk(DtxSelectOkBody body, int channelId) throws AMQException
+ {
+ throw new UnexpectedMethodException(body);
+ }
+
+ public boolean dispatchDtxStartOk(DtxStartOkBody 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 dispatchFileCancelOk(FileCancelOkBody body, int channelId) throws AMQException
+ {
+ throw new UnexpectedMethodException(body);
+ }
+
+ public boolean dispatchFileConsumeOk(FileConsumeOkBody body, int channelId) throws AMQException
+ {
+ throw new UnexpectedMethodException(body);
+ }
+
+ public boolean dispatchFileDeliver(FileDeliverBody body, int channelId) throws AMQException
+ {
+ throw new UnexpectedMethodException(body);
+ }
+
+ public boolean dispatchFileOpen(FileOpenBody body, int channelId) throws AMQException
+ {
+ throw new UnexpectedMethodException(body);
+ }
+
+ public boolean dispatchFileOpenOk(FileOpenOkBody body, int channelId) throws AMQException
+ {
+ throw new UnexpectedMethodException(body);
+ }
+
+ public boolean dispatchFileQosOk(FileQosOkBody body, int channelId) throws AMQException
+ {
+ throw new UnexpectedMethodException(body);
+ }
+
+ public boolean dispatchFileReturn(FileReturnBody body, int channelId) throws AMQException
+ {
+ throw new UnexpectedMethodException(body);
+ }
+
+ public boolean dispatchFileStage(FileStageBody 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 dispatchStreamCancelOk(StreamCancelOkBody body, int channelId) throws AMQException
+ {
+ throw new UnexpectedMethodException(body);
+ }
+
+ public boolean dispatchStreamConsumeOk(StreamConsumeOkBody body, int channelId) throws AMQException
+ {
+ throw new UnexpectedMethodException(body);
+ }
+
+ public boolean dispatchStreamDeliver(StreamDeliverBody body, int channelId) throws AMQException
+ {
+ throw new UnexpectedMethodException(body);
+ }
+
+ public boolean dispatchStreamQosOk(StreamQosOkBody body, int channelId) throws AMQException
+ {
+ throw new UnexpectedMethodException(body);
+ }
+
+ public boolean dispatchStreamReturn(StreamReturnBody 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
+ {
+ _connectionSecureOkMethodHandler.methodReceived(_stateManager, body, channelId);
+ return true;
+ }
+
+ public boolean dispatchConnectionStartOk(ConnectionStartOkBody body, int channelId) throws AMQException
+ {
+ _connectionStartOkMethodHandler.methodReceived(_stateManager, body, channelId);
+ return true;
+ }
+
+ public boolean dispatchConnectionTuneOk(ConnectionTuneOkBody body, int channelId) throws AMQException
+ {
+ _connectionTuneOkMethodHandler.methodReceived(_stateManager, body, channelId);
+ return true;
+ }
+
+ public boolean dispatchDtxSelect(DtxSelectBody body, int channelId) throws AMQException
+ {
+ return false;
+ }
+
+ public boolean dispatchDtxStart(DtxStartBody body, int channelId) throws AMQException
+ {
+ return false;
+ }
+
+ public boolean dispatchExchangeBound(ExchangeBoundBody body, int channelId) throws AMQException
+ {
+ _exchangeBoundHandler.methodReceived(_stateManager, body, channelId);
+ return true;
+ }
+
+ public boolean dispatchExchangeDeclare(ExchangeDeclareBody body, int channelId) throws AMQException
+ {
+ _exchangeDeclareHandler.methodReceived(_stateManager, body, channelId);
+ return true;
+ }
+
+ public boolean dispatchExchangeDelete(ExchangeDeleteBody body, int channelId) throws AMQException
+ {
+ _exchangeDeleteHandler.methodReceived(_stateManager, body, channelId);
+ return true;
+ }
+
+ public boolean dispatchFileAck(FileAckBody body, int channelId) throws AMQException
+ {
+ return false;
+ }
+
+ public boolean dispatchFileCancel(FileCancelBody body, int channelId) throws AMQException
+ {
+ return false;
+ }
+
+ public boolean dispatchFileConsume(FileConsumeBody body, int channelId) throws AMQException
+ {
+ return false;
+ }
+
+ public boolean dispatchFilePublish(FilePublishBody body, int channelId) throws AMQException
+ {
+ return false;
+ }
+
+ public boolean dispatchFileQos(FileQosBody body, int channelId) throws AMQException
+ {
+ return false;
+ }
+
+ public boolean dispatchFileReject(FileRejectBody body, int channelId) throws AMQException
+ {
+ return false;
+ }
+
+ public boolean dispatchQueueBind(QueueBindBody body, int channelId) throws AMQException
+ {
+ _queueBindHandler.methodReceived(_stateManager, body, channelId);
+ return true;
+ }
+
+ public boolean dispatchQueueDeclare(QueueDeclareBody body, int channelId) throws AMQException
+ {
+ _queueDeclareHandler.methodReceived(_stateManager, body, channelId);
+ return true;
+ }
+
+ public boolean dispatchQueueDelete(QueueDeleteBody body, int channelId) throws AMQException
+ {
+ _queueDeleteHandler.methodReceived(_stateManager, body, channelId);
+ return true;
+ }
+
+ public boolean dispatchQueuePurge(QueuePurgeBody body, int channelId) throws AMQException
+ {
+ _queuePurgeHandler.methodReceived(_stateManager, body, channelId);
+ return true;
+ }
+
+ public boolean dispatchStreamCancel(StreamCancelBody body, int channelId) throws AMQException
+ {
+ return false;
+ }
+
+ public boolean dispatchStreamConsume(StreamConsumeBody body, int channelId) throws AMQException
+ {
+ return false;
+ }
+
+ public boolean dispatchStreamPublish(StreamPublishBody body, int channelId) throws AMQException
+ {
+ return false;
+ }
+
+ public boolean dispatchStreamQos(StreamQosBody body, int channelId) throws AMQException
+ {
+ return false;
+ }
+
+ public boolean dispatchTunnelRequest(TunnelRequestBody body, int channelId) throws AMQException
+ {
+ return false;
+ }
+
+ public boolean dispatchTxCommit(TxCommitBody body, int channelId) throws AMQException
+ {
+ _txCommitHandler.methodReceived(_stateManager, body, channelId);
+ return true;
+ }
+
+ public boolean dispatchTxRollback(TxRollbackBody body, int channelId) throws AMQException
+ {
+ _txRollbackHandler.methodReceived(_stateManager, body, channelId);
+ return true;
+ }
+
+ public boolean dispatchTxSelect(TxSelectBody body, int channelId) throws AMQException
+ {
+ _txSelectHandler.methodReceived(_stateManager, body, channelId);
+ return true;
+ }
+
+
+
+
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/handler/ServerMethodDispatcherImpl_0_9.java b/java/broker/src/main/java/org/apache/qpid/server/handler/ServerMethodDispatcherImpl_0_9.java new file mode 100644 index 0000000000..97646a015f --- /dev/null +++ b/java/broker/src/main/java/org/apache/qpid/server/handler/ServerMethodDispatcherImpl_0_9.java @@ -0,0 +1,160 @@ +/*
+ *
+ * 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.handler;
+
+
+import org.apache.qpid.framing.amqp_0_9.MethodDispatcher_0_9;
+import org.apache.qpid.framing.*;
+import org.apache.qpid.server.state.AMQStateManager;
+import org.apache.qpid.AMQException;
+
+
+
+public class ServerMethodDispatcherImpl_0_9
+ extends ServerMethodDispatcherImpl
+ implements MethodDispatcher_0_9
+
+{
+
+ private static final BasicRecoverSyncMethodHandler _basicRecoverSyncMethodHandler =
+ BasicRecoverSyncMethodHandler.getInstance();
+
+ public ServerMethodDispatcherImpl_0_9(AMQStateManager stateManager)
+ {
+ super(stateManager);
+ }
+
+ public boolean dispatchBasicRecoverSync(BasicRecoverSyncBody body, int channelId) throws AMQException
+ {
+ _basicRecoverSyncMethodHandler.methodReceived(getStateManager(), body, channelId);
+ return true;
+ }
+
+ public boolean dispatchBasicRecoverSyncOk(BasicRecoverSyncOkBody body, int channelId) throws AMQException
+ {
+ throw new UnexpectedMethodException(body);
+ }
+
+ public boolean dispatchChannelOk(ChannelOkBody body, int channelId) throws AMQException
+ {
+ return false;
+ }
+
+ public boolean dispatchChannelPing(ChannelPingBody body, int channelId) throws AMQException
+ {
+ return false;
+ }
+
+ public boolean dispatchChannelPong(ChannelPongBody body, int channelId) throws AMQException
+ {
+ return false;
+ }
+
+ public boolean dispatchChannelResume(ChannelResumeBody body, int channelId) throws AMQException
+ {
+ return false;
+ }
+
+ public boolean dispatchMessageAppend(MessageAppendBody body, int channelId) throws AMQException
+ {
+ return false;
+ }
+
+ public boolean dispatchMessageCancel(MessageCancelBody body, int channelId) throws AMQException
+ {
+ return false;
+ }
+
+ public boolean dispatchMessageCheckpoint(MessageCheckpointBody body, int channelId) throws AMQException
+ {
+ return false;
+ }
+
+ public boolean dispatchMessageClose(MessageCloseBody body, int channelId) throws AMQException
+ {
+ return false;
+ }
+
+ public boolean dispatchMessageConsume(MessageConsumeBody body, int channelId) throws AMQException
+ {
+ return false;
+ }
+
+ public boolean dispatchMessageEmpty(MessageEmptyBody body, int channelId) throws AMQException
+ {
+ return false;
+ }
+
+ public boolean dispatchMessageGet(MessageGetBody body, int channelId) throws AMQException
+ {
+ return false;
+ }
+
+ public boolean dispatchMessageOffset(MessageOffsetBody body, int channelId) throws AMQException
+ {
+ return false;
+ }
+
+ public boolean dispatchMessageOk(MessageOkBody body, int channelId) throws AMQException
+ {
+ return false;
+ }
+
+ public boolean dispatchMessageOpen(MessageOpenBody body, int channelId) throws AMQException
+ {
+ return false;
+ }
+
+ public boolean dispatchMessageQos(MessageQosBody body, int channelId) throws AMQException
+ {
+ return false;
+ }
+
+ public boolean dispatchMessageRecover(MessageRecoverBody body, int channelId) throws AMQException
+ {
+ return false;
+ }
+
+ public boolean dispatchMessageReject(MessageRejectBody body, int channelId) throws AMQException
+ {
+ return false;
+ }
+
+ public boolean dispatchMessageResume(MessageResumeBody body, int channelId) throws AMQException
+ {
+ return false;
+ }
+
+ public boolean dispatchMessageTransfer(MessageTransferBody body, int channelId) throws AMQException
+ {
+ return false;
+ }
+
+ public boolean dispatchQueueUnbindOk(QueueUnbindOkBody body, int channelId) throws AMQException
+ {
+ throw new UnexpectedMethodException(body);
+ }
+
+ public boolean dispatchQueueUnbind(QueueUnbindBody body, int channelId) throws AMQException
+ {
+ return false;
+ }
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/handler/ServerMethodDispatcherImpl_8_0.java b/java/broker/src/main/java/org/apache/qpid/server/handler/ServerMethodDispatcherImpl_8_0.java new file mode 100644 index 0000000000..d599ca3d4e --- /dev/null +++ b/java/broker/src/main/java/org/apache/qpid/server/handler/ServerMethodDispatcherImpl_8_0.java @@ -0,0 +1,86 @@ +/*
+ *
+ * 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.handler;
+
+import org.apache.qpid.framing.amqp_8_0.MethodDispatcher_8_0;
+import org.apache.qpid.framing.*;
+import org.apache.qpid.server.state.AMQStateManager;
+import org.apache.qpid.AMQException;
+
+public class ServerMethodDispatcherImpl_8_0
+ extends ServerMethodDispatcherImpl
+ implements MethodDispatcher_8_0
+{
+ public ServerMethodDispatcherImpl_8_0(AMQStateManager stateManager)
+ {
+ super(stateManager);
+ }
+
+ public boolean dispatchBasicRecoverOk(BasicRecoverOkBody body, int channelId) throws AMQException
+ {
+ throw new UnexpectedMethodException(body);
+ }
+
+ public boolean dispatchChannelAlert(ChannelAlertBody body, int channelId) throws AMQException
+ {
+ throw new UnexpectedMethodException(body);
+ }
+
+ public boolean dispatchTestContent(TestContentBody body, int channelId) throws AMQException
+ {
+ return false;
+ }
+
+ public boolean dispatchTestContentOk(TestContentOkBody body, int channelId) throws AMQException
+ {
+ return false;
+ }
+
+ public boolean dispatchTestInteger(TestIntegerBody body, int channelId) throws AMQException
+ {
+ return false;
+ }
+
+ public boolean dispatchTestIntegerOk(TestIntegerOkBody body, int channelId) throws AMQException
+ {
+ return false;
+ }
+
+ public boolean dispatchTestString(TestStringBody body, int channelId) throws AMQException
+ {
+ return false;
+ }
+
+ public boolean dispatchTestStringOk(TestStringOkBody body, int channelId) throws AMQException
+ {
+ return false;
+ }
+
+ public boolean dispatchTestTable(TestTableBody body, int channelId) throws AMQException
+ {
+ return false;
+ }
+
+ public boolean dispatchTestTableOk(TestTableOkBody body, int channelId) throws AMQException
+ {
+ return false;
+ }
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/handler/TxCommitHandler.java b/java/broker/src/main/java/org/apache/qpid/server/handler/TxCommitHandler.java index 3d7ec286f9..79cc722e0e 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/handler/TxCommitHandler.java +++ b/java/broker/src/main/java/org/apache/qpid/server/handler/TxCommitHandler.java @@ -24,6 +24,8 @@ import org.apache.log4j.Logger; import org.apache.qpid.AMQException; import org.apache.qpid.framing.TxCommitBody; import org.apache.qpid.framing.TxCommitOkBody; +import org.apache.qpid.framing.MethodRegistry; +import org.apache.qpid.framing.AMQMethodBody; import org.apache.qpid.protocol.AMQMethodEvent; import org.apache.qpid.server.AMQChannel; import org.apache.qpid.server.protocol.AMQProtocolSession; @@ -45,7 +47,7 @@ public class TxCommitHandler implements StateAwareMethodListener<TxCommitBody> { } - public void methodReceived(AMQStateManager stateManager, AMQMethodEvent<TxCommitBody> evt) throws AMQException + public void methodReceived(AMQStateManager stateManager, TxCommitBody body, int channelId) throws AMQException { AMQProtocolSession session = stateManager.getProtocolSession(); @@ -53,25 +55,26 @@ public class TxCommitHandler implements StateAwareMethodListener<TxCommitBody> { if (_log.isDebugEnabled()) { - _log.debug("Commit received on channel " + evt.getChannelId()); + _log.debug("Commit received on channel " + channelId); } - AMQChannel channel = session.getChannel(evt.getChannelId()); + AMQChannel channel = session.getChannel(channelId); if (channel == null) { - throw evt.getMethod().getChannelNotFoundException(evt.getChannelId()); + throw body.getChannelNotFoundException(channelId); } channel.commit(); - // AMQP version change: Hardwire the version to 0-8 (major=8, minor=0) - // TODO: Connect this to the session version obtained from ProtocolInitiation for this session. - // Be aware of possible changes to parameter order as versions change. - session.writeFrame(TxCommitOkBody.createAMQFrame(evt.getChannelId(), (byte) 8, (byte) 0)); + + MethodRegistry methodRegistry = session.getMethodRegistry(); + AMQMethodBody responseBody = methodRegistry.createTxCommitOkBody(); + session.writeFrame(responseBody.generateFrame(channelId)); + channel.processReturns(session); } catch (AMQException e) { - throw evt.getMethod().getChannelException(e.getErrorCode(), "Failed to commit: " + e.getMessage()); + throw body.getChannelException(e.getErrorCode(), "Failed to commit: " + e.getMessage()); } } } diff --git a/java/broker/src/main/java/org/apache/qpid/server/handler/TxRollbackHandler.java b/java/broker/src/main/java/org/apache/qpid/server/handler/TxRollbackHandler.java index f747f7a840..5f402f3fda 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/handler/TxRollbackHandler.java +++ b/java/broker/src/main/java/org/apache/qpid/server/handler/TxRollbackHandler.java @@ -23,6 +23,8 @@ package org.apache.qpid.server.handler; import org.apache.qpid.AMQException; import org.apache.qpid.framing.TxRollbackBody; import org.apache.qpid.framing.TxRollbackOkBody; +import org.apache.qpid.framing.MethodRegistry; +import org.apache.qpid.framing.AMQMethodBody; import org.apache.qpid.protocol.AMQMethodEvent; import org.apache.qpid.server.AMQChannel; import org.apache.qpid.server.protocol.AMQProtocolSession; @@ -42,24 +44,26 @@ public class TxRollbackHandler implements StateAwareMethodListener<TxRollbackBod { } - public void methodReceived(AMQStateManager stateManager, AMQMethodEvent<TxRollbackBody> evt) throws AMQException + public void methodReceived(AMQStateManager stateManager, TxRollbackBody body, int channelId) throws AMQException { AMQProtocolSession session = stateManager.getProtocolSession(); try { - AMQChannel channel = session.getChannel(evt.getChannelId()); + AMQChannel channel = session.getChannel(channelId); if (channel == null) { - throw evt.getMethod().getChannelNotFoundException(evt.getChannelId()); + throw body.getChannelNotFoundException(channelId); } channel.rollback(); - // AMQP version change: Hardwire the version to 0-8 (major=8, minor=0) - // TODO: Connect this to the session version obtained from ProtocolInitiation for this session. - // Be aware of possible changes to parameter order as versions change. - session.writeFrame(TxRollbackOkBody.createAMQFrame(evt.getChannelId(), (byte) 8, (byte) 0)); + + MethodRegistry methodRegistry = session.getMethodRegistry(); + AMQMethodBody responseBody = methodRegistry.createTxRollbackOkBody(); + session.writeFrame(responseBody.generateFrame(channelId)); + + //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? @@ -67,7 +71,7 @@ public class TxRollbackHandler implements StateAwareMethodListener<TxRollbackBod } catch (AMQException e) { - throw evt.getMethod().getChannelException(e.getErrorCode(), "Failed to rollback: " + e.getMessage()); + throw body.getChannelException(e.getErrorCode(), "Failed to rollback: " + e.getMessage()); } } } diff --git a/java/broker/src/main/java/org/apache/qpid/server/handler/TxSelectHandler.java b/java/broker/src/main/java/org/apache/qpid/server/handler/TxSelectHandler.java index a9e478e301..308f5b73cf 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/handler/TxSelectHandler.java +++ b/java/broker/src/main/java/org/apache/qpid/server/handler/TxSelectHandler.java @@ -23,6 +23,7 @@ package org.apache.qpid.server.handler; import org.apache.qpid.AMQException; import org.apache.qpid.framing.TxSelectBody; import org.apache.qpid.framing.TxSelectOkBody; +import org.apache.qpid.framing.MethodRegistry; import org.apache.qpid.protocol.AMQMethodEvent; import org.apache.qpid.server.protocol.AMQProtocolSession; import org.apache.qpid.server.state.AMQStateManager; @@ -42,22 +43,21 @@ public class TxSelectHandler implements StateAwareMethodListener<TxSelectBody> { } - public void methodReceived(AMQStateManager stateManager, AMQMethodEvent<TxSelectBody> evt) throws AMQException + public void methodReceived(AMQStateManager stateManager, TxSelectBody body, int channelId) throws AMQException { AMQProtocolSession session = stateManager.getProtocolSession(); - AMQChannel channel = session.getChannel(evt.getChannelId()); + AMQChannel channel = session.getChannel(channelId); if (channel == null) { - throw evt.getMethod().getChannelNotFoundException(evt.getChannelId()); + throw body.getChannelNotFoundException(channelId); } channel.setLocalTransactional(); - // AMQP version change: Hardwire the version to 0-8 (major=8, minor=0) - // TODO: Connect this to the session version obtained from ProtocolInitiation for this session. - // Be aware of possible changes to parameter order as versions change. - session.writeFrame(TxSelectOkBody.createAMQFrame(evt.getChannelId(), (byte) 8, (byte) 0)); + MethodRegistry methodRegistry = session.getMethodRegistry(); + TxSelectOkBody responseBody = methodRegistry.createTxSelectOkBody(); + session.writeFrame(responseBody.generateFrame(channelId)); } } diff --git a/java/broker/src/main/java/org/apache/qpid/server/handler/UnexpectedMethodException.java b/java/broker/src/main/java/org/apache/qpid/server/handler/UnexpectedMethodException.java new file mode 100644 index 0000000000..fb18519fe1 --- /dev/null +++ b/java/broker/src/main/java/org/apache/qpid/server/handler/UnexpectedMethodException.java @@ -0,0 +1,33 @@ +/*
+ *
+ * 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.handler;
+
+
+import org.apache.qpid.framing.AMQMethodBody;
+import org.apache.qpid.AMQException;
+
+public class UnexpectedMethodException extends AMQException
+{
+ public UnexpectedMethodException(AMQMethodBody body)
+ {
+ super("Unexpected method recevied: " + body.getClass().getName());
+ }
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/output/ProtocolOutputConverterRegistry.java b/java/broker/src/main/java/org/apache/qpid/server/output/ProtocolOutputConverterRegistry.java index 8366c426dd..36e7e88fd6 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/output/ProtocolOutputConverterRegistry.java +++ b/java/broker/src/main/java/org/apache/qpid/server/output/ProtocolOutputConverterRegistry.java @@ -27,8 +27,8 @@ package org.apache.qpid.server.output;
import org.apache.qpid.server.output.ProtocolOutputConverter.Factory;
-import org.apache.qpid.server.output.amqp0_8.ProtocolOutputConverterImpl;
import org.apache.qpid.server.protocol.AMQProtocolSession;
+import org.apache.qpid.framing.ProtocolVersion;
import java.util.Map;
import java.util.HashMap;
@@ -36,27 +36,26 @@ import java.util.HashMap; public class ProtocolOutputConverterRegistry
{
- private static final Map<Byte, Map<Byte, Factory>> _registry =
- new HashMap<Byte, Map<Byte, Factory>>();
+ private static final Map<ProtocolVersion, Factory> _registry =
+ new HashMap<ProtocolVersion, Factory>();
static
{
- register((byte) 8, (byte) 0, ProtocolOutputConverterImpl.getInstanceFactory());
+ register(ProtocolVersion.v8_0, org.apache.qpid.server.output.amqp0_8.ProtocolOutputConverterImpl.getInstanceFactory());
+ register(ProtocolVersion.v0_9, org.apache.qpid.server.output.amqp0_9.ProtocolOutputConverterImpl.getInstanceFactory());
+
}
- private static void register(byte major, byte minor, Factory converter)
+ private static void register(ProtocolVersion version, Factory converter)
{
- if(!_registry.containsKey(major))
- {
- _registry.put(major, new HashMap<Byte, Factory>());
- }
- _registry.get(major).put(minor, converter);
+
+ _registry.put(version,converter);
}
public static ProtocolOutputConverter getConverter(AMQProtocolSession session)
{
- return _registry.get(session.getProtocolMajorVersion()).get(session.getProtocolMinorVersion()).newInstance(session);
+ return _registry.get(session.getProtocolVersion()).newInstance(session);
}
}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/output/amqp0_8/ProtocolOutputConverterImpl.java b/java/broker/src/main/java/org/apache/qpid/server/output/amqp0_8/ProtocolOutputConverterImpl.java index 8462ed9557..1bfe1e3d35 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/output/amqp0_8/ProtocolOutputConverterImpl.java +++ b/java/broker/src/main/java/org/apache/qpid/server/output/amqp0_8/ProtocolOutputConverterImpl.java @@ -43,6 +43,7 @@ import java.util.Iterator; public class ProtocolOutputConverterImpl implements ProtocolOutputConverter
{
+
public static Factory getInstanceFactory()
{
return new Factory()
@@ -98,7 +99,7 @@ public class ProtocolOutputConverterImpl implements ProtocolOutputConverter //
ContentChunk cb = messageHandle.getContentChunk(storeContext,messageId, 0);
- AMQDataBlock firstContentBody = new AMQFrame(channelId, getProtocolSession().getRegistry().getProtocolVersionMethodConverter().convertToBody(cb));
+ AMQDataBlock firstContentBody = new AMQFrame(channelId, getProtocolSession().getMethodRegistry().getProtocolVersionMethodConverter().convertToBody(cb));
AMQDataBlock[] headerAndFirstContent = new AMQDataBlock[]{contentHeader, firstContentBody};
CompositeAMQDataBlock compositeBlock = new CompositeAMQDataBlock(deliver, headerAndFirstContent);
writeFrame(compositeBlock);
@@ -109,7 +110,7 @@ public class ProtocolOutputConverterImpl implements ProtocolOutputConverter for(int i = 1; i < bodyCount; i++)
{
cb = messageHandle.getContentChunk(storeContext,messageId, i);
- writeFrame(new AMQFrame(channelId, getProtocolSession().getRegistry().getProtocolVersionMethodConverter().convertToBody(cb)));
+ writeFrame(new AMQFrame(channelId, getProtocolSession().getMethodRegistry().getProtocolVersionMethodConverter().convertToBody(cb)));
}
@@ -149,7 +150,7 @@ public class ProtocolOutputConverterImpl implements ProtocolOutputConverter //
ContentChunk cb = messageHandle.getContentChunk(storeContext,messageId, 0);
- AMQDataBlock firstContentBody = new AMQFrame(channelId, getProtocolSession().getRegistry().getProtocolVersionMethodConverter().convertToBody(cb));
+ AMQDataBlock firstContentBody = new AMQFrame(channelId, getProtocolSession().getMethodRegistry().getProtocolVersionMethodConverter().convertToBody(cb));
AMQDataBlock[] headerAndFirstContent = new AMQDataBlock[]{contentHeader, firstContentBody};
CompositeAMQDataBlock compositeBlock = new CompositeAMQDataBlock(deliver, headerAndFirstContent);
writeFrame(compositeBlock);
@@ -160,7 +161,7 @@ public class ProtocolOutputConverterImpl implements ProtocolOutputConverter for(int i = 1; i < bodyCount; i++)
{
cb = messageHandle.getContentChunk(storeContext, messageId, i);
- writeFrame(new AMQFrame(channelId, getProtocolSession().getRegistry().getProtocolVersionMethodConverter().convertToBody(cb)));
+ writeFrame(new AMQFrame(channelId, getProtocolSession().getMethodRegistry().getProtocolVersionMethodConverter().convertToBody(cb)));
}
@@ -176,11 +177,14 @@ public class ProtocolOutputConverterImpl implements ProtocolOutputConverter final MessagePublishInfo pb = message.getMessagePublishInfo();
final AMQMessageHandle messageHandle = message.getMessageHandle();
- AMQFrame deliverFrame = BasicDeliverBody.createAMQFrame(channelId, getProtocolMajorVersion(),
- getProtocolMinorVersion(),
- consumerTag,
- deliveryTag, pb.getExchange(), messageHandle.isRedelivered(),
- pb.getRoutingKey());
+ MethodRegistry methodRegistry = MethodRegistry.getMethodRegistry(ProtocolVersion.v8_0);
+ BasicDeliverBody deliverBody =
+ methodRegistry.createBasicDeliverBody(consumerTag,
+ deliveryTag,
+ messageHandle.isRedelivered(),
+ pb.getExchange(),
+ pb.getRoutingKey());
+ AMQFrame deliverFrame = deliverBody.generateFrame(channelId);
return deliverFrame.toByteBuffer();
@@ -192,13 +196,14 @@ public class ProtocolOutputConverterImpl implements ProtocolOutputConverter final MessagePublishInfo pb = message.getMessagePublishInfo();
final AMQMessageHandle messageHandle = message.getMessageHandle();
- AMQFrame getOkFrame = BasicGetOkBody.createAMQFrame(channelId,
- getProtocolMajorVersion(),
- getProtocolMinorVersion(),
- deliveryTag, pb.getExchange(),
- queueSize,
- messageHandle.isRedelivered(),
- pb.getRoutingKey());
+ MethodRegistry methodRegistry = MethodRegistry.getMethodRegistry(ProtocolVersion.v8_0);
+ BasicGetOkBody getOkBody =
+ methodRegistry.createBasicGetOkBody(deliveryTag,
+ messageHandle.isRedelivered(),
+ pb.getExchange(),
+ pb.getRoutingKey(),
+ queueSize);
+ AMQFrame getOkFrame = getOkBody.generateFrame(channelId);
return getOkFrame.toByteBuffer();
}
@@ -215,12 +220,13 @@ public class ProtocolOutputConverterImpl implements ProtocolOutputConverter private ByteBuffer createEncodedReturnFrame(AMQMessage message, int channelId, int replyCode, AMQShortString replyText) throws AMQException
{
- AMQFrame returnFrame = BasicReturnBody.createAMQFrame(channelId,
- getProtocolMajorVersion(),
- getProtocolMinorVersion(),
- message.getMessagePublishInfo().getExchange(),
- replyCode, replyText,
- message.getMessagePublishInfo().getRoutingKey());
+ MethodRegistry methodRegistry = MethodRegistry.getMethodRegistry(ProtocolVersion.v8_0);
+ BasicReturnBody basicReturnBody =
+ methodRegistry.createBasicReturnBody(replyCode,
+ replyText,
+ message.getMessagePublishInfo().getExchange(),
+ message.getMessagePublishInfo().getRoutingKey());
+ AMQFrame returnFrame = basicReturnBody.generateFrame(channelId);
return returnFrame.toByteBuffer();
}
@@ -272,11 +278,9 @@ public class ProtocolOutputConverterImpl implements ProtocolOutputConverter public void confirmConsumerAutoClose(int channelId, AMQShortString consumerTag)
{
+ MethodRegistry methodRegistry = MethodRegistry.getMethodRegistry(ProtocolVersion.v8_0);
+ BasicCancelOkBody basicCancelOkBody = methodRegistry.createBasicCancelOkBody(consumerTag);
+ writeFrame(basicCancelOkBody.generateFrame(channelId));
- writeFrame(BasicCancelOkBody.createAMQFrame(channelId,
- getProtocolMajorVersion(),
- getProtocolMinorVersion(),
- consumerTag // consumerTag
- ));
}
}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/output/amqp0_9/ProtocolOutputConverterImpl.java b/java/broker/src/main/java/org/apache/qpid/server/output/amqp0_9/ProtocolOutputConverterImpl.java new file mode 100644 index 0000000000..0bc2fcf6f7 --- /dev/null +++ b/java/broker/src/main/java/org/apache/qpid/server/output/amqp0_9/ProtocolOutputConverterImpl.java @@ -0,0 +1,260 @@ +package org.apache.qpid.server.output.amqp0_9;
+
+import org.apache.mina.common.ByteBuffer;
+
+import java.util.Iterator;
+
+import org.apache.qpid.server.output.ProtocolOutputConverter;
+import org.apache.qpid.server.protocol.AMQProtocolSession;
+import org.apache.qpid.server.queue.AMQMessage;
+import org.apache.qpid.server.queue.AMQMessageHandle;
+import org.apache.qpid.server.store.StoreContext;
+import org.apache.qpid.framing.*;
+import org.apache.qpid.framing.abstraction.ContentChunk;
+import org.apache.qpid.framing.abstraction.MessagePublishInfo;
+import org.apache.qpid.AMQException;
+
+public class ProtocolOutputConverterImpl implements ProtocolOutputConverter
+{
+
+
+ public static Factory getInstanceFactory()
+ {
+ return new Factory()
+ {
+
+ public ProtocolOutputConverter newInstance(AMQProtocolSession session)
+ {
+ return new ProtocolOutputConverterImpl(session);
+ }
+ };
+ }
+
+ private final AMQProtocolSession _protocolSession;
+
+ private ProtocolOutputConverterImpl(AMQProtocolSession session)
+ {
+ _protocolSession = session;
+ }
+
+
+ public AMQProtocolSession getProtocolSession()
+ {
+ return _protocolSession;
+ }
+
+ public void writeDeliver(AMQMessage message, int channelId, long deliveryTag, AMQShortString consumerTag)
+ throws AMQException
+ {
+ ByteBuffer deliver = createEncodedDeliverFrame(message, channelId, deliveryTag, consumerTag);
+ AMQDataBlock contentHeader = ContentHeaderBody.createAMQFrame(channelId,
+ message.getContentHeaderBody());
+
+ final AMQMessageHandle messageHandle = message.getMessageHandle();
+ final StoreContext storeContext = message.getStoreContext();
+ final Long messageId = message.getMessageId();
+
+ final int bodyCount = messageHandle.getBodyCount(storeContext,messageId);
+
+ if(bodyCount == 0)
+ {
+ SmallCompositeAMQDataBlock compositeBlock = new SmallCompositeAMQDataBlock(deliver,
+ contentHeader);
+
+ writeFrame(compositeBlock);
+ }
+ else
+ {
+
+
+ //
+ // Optimise the case where we have a single content body. In that case we create a composite block
+ // so that we can writeDeliver out the deliver, header and body with a single network writeDeliver.
+ //
+ ContentChunk cb = messageHandle.getContentChunk(storeContext,messageId, 0);
+
+ AMQDataBlock firstContentBody = new AMQFrame(channelId, getProtocolSession().getMethodRegistry().getProtocolVersionMethodConverter().convertToBody(cb));
+ AMQDataBlock[] headerAndFirstContent = new AMQDataBlock[]{contentHeader, firstContentBody};
+ CompositeAMQDataBlock compositeBlock = new CompositeAMQDataBlock(deliver, headerAndFirstContent);
+ writeFrame(compositeBlock);
+
+ //
+ // Now start writing out the other content bodies
+ //
+ for(int i = 1; i < bodyCount; i++)
+ {
+ cb = messageHandle.getContentChunk(storeContext,messageId, i);
+ writeFrame(new AMQFrame(channelId, getProtocolSession().getMethodRegistry().getProtocolVersionMethodConverter().convertToBody(cb)));
+ }
+
+
+ }
+
+
+ }
+
+
+ public void writeGetOk(AMQMessage message, int channelId, long deliveryTag, int queueSize) throws AMQException
+ {
+
+ final AMQMessageHandle messageHandle = message.getMessageHandle();
+ final StoreContext storeContext = message.getStoreContext();
+ final long messageId = message.getMessageId();
+
+ ByteBuffer deliver = createEncodedGetOkFrame(message, channelId, deliveryTag, queueSize);
+
+
+ AMQDataBlock contentHeader = ContentHeaderBody.createAMQFrame(channelId,
+ message.getContentHeaderBody());
+
+ final int bodyCount = messageHandle.getBodyCount(storeContext,messageId);
+ if(bodyCount == 0)
+ {
+ SmallCompositeAMQDataBlock compositeBlock = new SmallCompositeAMQDataBlock(deliver,
+ contentHeader);
+ writeFrame(compositeBlock);
+ }
+ else
+ {
+
+
+ //
+ // Optimise the case where we have a single content body. In that case we create a composite block
+ // so that we can writeDeliver out the deliver, header and body with a single network writeDeliver.
+ //
+ ContentChunk cb = messageHandle.getContentChunk(storeContext,messageId, 0);
+
+ AMQDataBlock firstContentBody = new AMQFrame(channelId, getProtocolSession().getMethodRegistry().getProtocolVersionMethodConverter().convertToBody(cb));
+ AMQDataBlock[] headerAndFirstContent = new AMQDataBlock[]{contentHeader, firstContentBody};
+ CompositeAMQDataBlock compositeBlock = new CompositeAMQDataBlock(deliver, headerAndFirstContent);
+ writeFrame(compositeBlock);
+
+ //
+ // Now start writing out the other content bodies
+ //
+ for(int i = 1; i < bodyCount; i++)
+ {
+ cb = messageHandle.getContentChunk(storeContext, messageId, i);
+ writeFrame(new AMQFrame(channelId, getProtocolSession().getMethodRegistry().getProtocolVersionMethodConverter().convertToBody(cb)));
+ }
+
+
+ }
+
+
+ }
+
+
+ private ByteBuffer createEncodedDeliverFrame(AMQMessage message, int channelId, long deliveryTag, AMQShortString consumerTag)
+ throws AMQException
+ {
+ final MessagePublishInfo pb = message.getMessagePublishInfo();
+ final AMQMessageHandle messageHandle = message.getMessageHandle();
+
+ MethodRegistry methodRegistry = MethodRegistry.getMethodRegistry(ProtocolVersion.v0_9);
+ BasicDeliverBody deliverBody =
+ methodRegistry.createBasicDeliverBody(consumerTag,
+ deliveryTag,
+ messageHandle.isRedelivered(),
+ pb.getExchange(),
+ pb.getRoutingKey());
+ AMQFrame deliverFrame = deliverBody.generateFrame(channelId);
+
+
+ return deliverFrame.toByteBuffer();
+ }
+
+ private ByteBuffer createEncodedGetOkFrame(AMQMessage message, int channelId, long deliveryTag, int queueSize)
+ throws AMQException
+ {
+ final MessagePublishInfo pb = message.getMessagePublishInfo();
+ final AMQMessageHandle messageHandle = message.getMessageHandle();
+
+ MethodRegistry methodRegistry = MethodRegistry.getMethodRegistry(ProtocolVersion.v0_9);
+ BasicGetOkBody getOkBody =
+ methodRegistry.createBasicGetOkBody(deliveryTag,
+ messageHandle.isRedelivered(),
+ pb.getExchange(),
+ pb.getRoutingKey(),
+ queueSize);
+ AMQFrame getOkFrame = getOkBody.generateFrame(channelId);
+
+ return getOkFrame.toByteBuffer();
+ }
+
+ public byte getProtocolMinorVersion()
+ {
+ return getProtocolSession().getProtocolMinorVersion();
+ }
+
+ public byte getProtocolMajorVersion()
+ {
+ return getProtocolSession().getProtocolMajorVersion();
+ }
+
+ private ByteBuffer createEncodedReturnFrame(AMQMessage message, int channelId, int replyCode, AMQShortString replyText) throws AMQException
+ {
+ MethodRegistry methodRegistry = MethodRegistry.getMethodRegistry(ProtocolVersion.v0_9);
+ BasicReturnBody basicReturnBody =
+ methodRegistry.createBasicReturnBody(replyCode,
+ replyText,
+ message.getMessagePublishInfo().getExchange(),
+ message.getMessagePublishInfo().getRoutingKey());
+ AMQFrame returnFrame = basicReturnBody.generateFrame(channelId);
+
+ return returnFrame.toByteBuffer();
+ }
+
+ public void writeReturn(AMQMessage message, int channelId, int replyCode, AMQShortString replyText)
+ throws AMQException
+ {
+ ByteBuffer returnFrame = createEncodedReturnFrame(message, channelId, replyCode, replyText);
+
+ AMQDataBlock contentHeader = ContentHeaderBody.createAMQFrame(channelId,
+ message.getContentHeaderBody());
+
+ Iterator<AMQDataBlock> bodyFrameIterator = message.getBodyFrameIterator(getProtocolSession(), channelId);
+ //
+ // Optimise the case where we have a single content body. In that case we create a composite block
+ // so that we can writeDeliver out the deliver, header and body with a single network writeDeliver.
+ //
+ if (bodyFrameIterator.hasNext())
+ {
+ AMQDataBlock firstContentBody = bodyFrameIterator.next();
+ AMQDataBlock[] headerAndFirstContent = new AMQDataBlock[]{contentHeader, firstContentBody};
+ CompositeAMQDataBlock compositeBlock = new CompositeAMQDataBlock(returnFrame, headerAndFirstContent);
+ writeFrame(compositeBlock);
+ }
+ else
+ {
+ CompositeAMQDataBlock compositeBlock = new CompositeAMQDataBlock(returnFrame,
+ new AMQDataBlock[]{contentHeader});
+
+ writeFrame(compositeBlock);
+ }
+
+ //
+ // Now start writing out the other content bodies
+ // TODO: MINA needs to be fixed so the the pending writes buffer is not unbounded
+ //
+ while (bodyFrameIterator.hasNext())
+ {
+ writeFrame(bodyFrameIterator.next());
+ }
+ }
+
+
+ public void writeFrame(AMQDataBlock block)
+ {
+ getProtocolSession().writeFrame(block);
+ }
+
+
+ public void confirmConsumerAutoClose(int channelId, AMQShortString consumerTag)
+ {
+ MethodRegistry methodRegistry = MethodRegistry.getMethodRegistry(ProtocolVersion.v0_9);
+ BasicCancelOkBody basicCancelOkBody = methodRegistry.createBasicCancelOkBody(consumerTag);
+ writeFrame(basicCancelOkBody.generateFrame(channelId));
+
+ }
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQMinaProtocolSession.java b/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQMinaProtocolSession.java index 5bfd47b469..0fe6d3636e 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQMinaProtocolSession.java +++ b/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQMinaProtocolSession.java @@ -39,6 +39,7 @@ import org.apache.qpid.protocol.AMQConstant; import org.apache.qpid.protocol.AMQMethodEvent; import org.apache.qpid.protocol.AMQMethodListener; import org.apache.qpid.server.AMQChannel; +import org.apache.qpid.server.handler.ServerMethodDispatcherImpl; import org.apache.qpid.server.management.Managable; import org.apache.qpid.server.management.ManagedObject; import org.apache.qpid.server.output.ProtocolOutputConverter; @@ -107,10 +108,11 @@ public class AMQMinaProtocolSession implements AMQProtocolSession, Managable private FieldTable _clientProperties; private final List<Task> _taskList = new CopyOnWriteArrayList<Task>(); - private VersionSpecificRegistry _registry = MainRegistry.getVersionSpecificRegistry(_protocolVersion); + private List<Integer> _closingChannelsList = new ArrayList<Integer>(); private ProtocolOutputConverter _protocolOutputConverter; private Principal _authorizedID; + private MethodDispatcher _dispatcher; public ManagedObject getManagedObject() { @@ -235,24 +237,24 @@ public class AMQMinaProtocolSession implements AMQProtocolSession, Managable ((AMQDecoder) _codecFactory.getDecoder()).setExpectProtocolInitiation(false); try { - pi.checkVersion(); // Fails if not correct + ProtocolVersion pv = pi.checkVersion(); // Fails if not correct // This sets the protocol version (and hence framing classes) for this session. - setProtocolVersion(pi._protocolMajor, pi._protocolMinor); + setProtocolVersion(pv); String mechanisms = ApplicationRegistry.getInstance().getAuthenticationManager().getMechanisms(); String locales = "en_US"; - // Interfacing with generated code - be aware of possible changes to parameter order as versions change. - AMQFrame response = - ConnectionStartBody.createAMQFrame((short) 0, getProtocolMajorVersion(), getProtocolMinorVersion(), // AMQP version (major, minor) - locales.getBytes(), // locales - mechanisms.getBytes(), // mechanisms - null, // serverProperties - (short) getProtocolMajorVersion(), // versionMajor - (short) getProtocolMinorVersion()); // versionMinor - _minaProtocolSession.write(response); + + AMQMethodBody responseBody = getMethodRegistry().createConnectionStartBody((short) getProtocolMajorVersion(), + (short) getProtocolMinorVersion(), + null, + mechanisms.getBytes(), + locales.getBytes()); + _minaProtocolSession.write(responseBody.generateFrame(0)); + + } catch (AMQException e) { @@ -548,6 +550,7 @@ public class AMQMinaProtocolSession implements AMQProtocolSession, Managable public void closeChannelOk(int channelId) { + removeChannel(channelId); _closingChannelsList.remove(new Integer(channelId)); } @@ -695,13 +698,12 @@ public class AMQMinaProtocolSession implements AMQProtocolSession, Managable } } - private void setProtocolVersion(byte major, byte minor) + private void setProtocolVersion(ProtocolVersion pv) { - _protocolVersion = new ProtocolVersion(major, minor); - - _registry = MainRegistry.getVersionSpecificRegistry(_protocolVersion); + _protocolVersion = pv; _protocolOutputConverter = ProtocolOutputConverterRegistry.getConverter(this); + _dispatcher = ServerMethodDispatcherImpl.createMethodDispatcher(_stateManager, _protocolVersion); } public byte getProtocolMajorVersion() @@ -709,6 +711,11 @@ public class AMQMinaProtocolSession implements AMQProtocolSession, Managable return _protocolVersion.getMajorVersion(); } + public ProtocolVersion getProtocolVersion() + { + return _protocolVersion; + } + public byte getProtocolMinorVersion() { return _protocolVersion.getMinorVersion(); @@ -719,9 +726,9 @@ public class AMQMinaProtocolSession implements AMQProtocolSession, Managable return (getProtocolMajorVersion() == major) && (getProtocolMinorVersion() == minor); } - public VersionSpecificRegistry getRegistry() + public MethodRegistry getRegistry() { - return _registry; + return getMethodRegistry(); } public Object getClientIdentifier() @@ -766,6 +773,16 @@ public class AMQMinaProtocolSession implements AMQProtocolSession, Managable return _authorizedID; } + public MethodRegistry getMethodRegistry() + { + return MethodRegistry.getMethodRegistry(getProtocolVersion()); + } + + public MethodDispatcher getMethodDispatcher() + { + return _dispatcher; + } + public String getClientVersion() { return (_clientVersion == null) ? null : _clientVersion.toString(); diff --git a/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQPFastProtocolHandler.java b/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQPFastProtocolHandler.java index 476e608b01..06f2fbcfd7 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQPFastProtocolHandler.java +++ b/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQPFastProtocolHandler.java @@ -30,13 +30,7 @@ import org.apache.mina.filter.codec.ProtocolCodecFilter; import org.apache.mina.util.SessionUtil; import org.apache.qpid.AMQException; import org.apache.qpid.codec.AMQCodecFactory; -import org.apache.qpid.framing.AMQDataBlock; -import org.apache.qpid.framing.AMQProtocolHeaderException; -import org.apache.qpid.framing.AMQShortString; -import org.apache.qpid.framing.ConnectionCloseBody; -import org.apache.qpid.framing.HeartbeatBody; -import org.apache.qpid.framing.ProtocolInitiation; -import org.apache.qpid.framing.ProtocolVersion; +import org.apache.qpid.framing.*; import org.apache.qpid.server.registry.ApplicationRegistry; import org.apache.qpid.server.registry.IApplicationRegistry; import org.apache.qpid.server.transport.ConnectorConfiguration; @@ -177,15 +171,12 @@ public class AMQPFastProtocolHandler extends IoHandlerAdapter { _logger.error("Exception caught in" + session + ", closing session explictly: " + throwable, throwable); - // Be aware of possible changes to parameter order as versions change. - protocolSession.write(ConnectionCloseBody.createAMQFrame(0, - session.getProtocolMajorVersion(), - session.getProtocolMinorVersion(), // AMQP version (major, minor) - 0, // classId - 0, // methodId - 200, // replyCode - new AMQShortString(throwable.getMessage()) // replyText - )); + + MethodRegistry methodRegistry = MethodRegistry.getMethodRegistry(session.getProtocolVersion()); + ConnectionCloseBody closeBody = methodRegistry.createConnectionCloseBody(200,new AMQShortString(throwable.getMessage()),0,0); + + protocolSession.write(closeBody.generateFrame(0)); + protocolSession.close(); } } diff --git a/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSession.java b/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSession.java index 390117acf6..c9316f7405 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSession.java +++ b/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSession.java @@ -23,9 +23,7 @@ package org.apache.qpid.server.protocol; import javax.security.sasl.SaslServer; import org.apache.qpid.AMQException; -import org.apache.qpid.framing.AMQDataBlock; -import org.apache.qpid.framing.AMQShortString; -import org.apache.qpid.framing.FieldTable; +import org.apache.qpid.framing.*; import org.apache.qpid.protocol.AMQVersionAwareProtocolSession; import org.apache.qpid.server.AMQChannel; import org.apache.qpid.server.output.ProtocolOutputConverter; @@ -37,6 +35,8 @@ import java.security.Principal; public interface AMQProtocolSession extends AMQVersionAwareProtocolSession { + + public static interface Task { public void doTask(AMQProtocolSession session) throws AMQException; @@ -172,4 +172,8 @@ public interface AMQProtocolSession extends AMQVersionAwareProtocolSession /** @return a Principal that was used to authorized this session */ Principal getAuthorizedID(); + public MethodRegistry getMethodRegistry(); + + public MethodDispatcher getMethodDispatcher(); + } diff --git a/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSessionMBean.java b/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSessionMBean.java index 66f928a70e..bd072985c4 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSessionMBean.java +++ b/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSessionMBean.java @@ -61,6 +61,7 @@ import org.apache.qpid.AMQException; import org.apache.qpid.framing.AMQFrame; import org.apache.qpid.framing.AMQShortString; import org.apache.qpid.framing.ConnectionCloseBody; +import org.apache.qpid.framing.MethodRegistry; import org.apache.qpid.protocol.AMQConstant; import org.apache.qpid.server.AMQChannel; import org.apache.qpid.server.management.AMQManagedObject; @@ -261,17 +262,17 @@ public class AMQProtocolSessionMBean extends AMQManagedObject implements Managed */ public void closeConnection() throws JMException { - // AMQP version change: Hardwire the version to 0-8 (major=8, minor=0) - // TODO: Connect this to the session version obtained from ProtocolInitiation for this session. - // Be aware of possible changes to parameter order as versions change. - final AMQFrame response = - ConnectionCloseBody.createAMQFrame(0, _session.getProtocolMajorVersion(), _session.getProtocolMinorVersion(), // AMQP version (major, minor) - 0, // classId - 0, // methodId - AMQConstant.REPLY_SUCCESS.getCode(), // replyCode - BROKER_MANAGEMENT_CONSOLE_HAS_CLOSED_THE_CONNECTION // replyText - ); - _session.writeFrame(response); + + MethodRegistry methodRegistry = _session.getMethodRegistry(); + ConnectionCloseBody responseBody = + methodRegistry.createConnectionCloseBody(AMQConstant.REPLY_SUCCESS.getCode(), + // replyCode + BROKER_MANAGEMENT_CONSOLE_HAS_CLOSED_THE_CONNECTION, + // replyText, + 0, + 0); + + _session.writeFrame(responseBody.generateFrame(0)); try { diff --git a/java/broker/src/main/java/org/apache/qpid/server/queue/AMQMessage.java b/java/broker/src/main/java/org/apache/qpid/server/queue/AMQMessage.java index dd9f32a306..307a4c8d21 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/queue/AMQMessage.java +++ b/java/broker/src/main/java/org/apache/qpid/server/queue/AMQMessage.java @@ -187,7 +187,7 @@ public class AMQMessage private ProtocolVersionMethodConverter getProtocolVersionMethodConverter() { - return _protocolSession.getRegistry().getProtocolVersionMethodConverter(); + return _protocolSession.getMethodRegistry().getProtocolVersionMethodConverter(); } public void remove() diff --git a/java/broker/src/main/java/org/apache/qpid/server/state/AMQStateManager.java b/java/broker/src/main/java/org/apache/qpid/server/state/AMQStateManager.java index f96900d0a9..c5b3099f58 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/state/AMQStateManager.java +++ b/java/broker/src/main/java/org/apache/qpid/server/state/AMQStateManager.java @@ -28,37 +28,11 @@ import java.util.concurrent.CopyOnWriteArraySet; import org.apache.log4j.Logger; import org.apache.qpid.AMQException; -import org.apache.qpid.framing.AMQMethodBody; -import org.apache.qpid.framing.BasicAckBody; -import org.apache.qpid.framing.BasicCancelBody; -import org.apache.qpid.framing.BasicConsumeBody; -import org.apache.qpid.framing.BasicGetBody; -import org.apache.qpid.framing.BasicPublishBody; -import org.apache.qpid.framing.BasicQosBody; -import org.apache.qpid.framing.BasicRecoverBody; -import org.apache.qpid.framing.BasicRejectBody; -import org.apache.qpid.framing.ChannelCloseBody; -import org.apache.qpid.framing.ChannelCloseOkBody; -import org.apache.qpid.framing.ChannelFlowBody; -import org.apache.qpid.framing.ChannelOpenBody; -import org.apache.qpid.framing.ConnectionCloseBody; -import org.apache.qpid.framing.ConnectionCloseOkBody; -import org.apache.qpid.framing.ConnectionOpenBody; -import org.apache.qpid.framing.ConnectionSecureOkBody; -import org.apache.qpid.framing.ConnectionStartOkBody; -import org.apache.qpid.framing.ConnectionTuneOkBody; -import org.apache.qpid.framing.ExchangeBoundBody; -import org.apache.qpid.framing.ExchangeDeclareBody; -import org.apache.qpid.framing.ExchangeDeleteBody; -import org.apache.qpid.framing.QueueBindBody; -import org.apache.qpid.framing.QueueDeclareBody; -import org.apache.qpid.framing.QueueDeleteBody; -import org.apache.qpid.framing.QueuePurgeBody; -import org.apache.qpid.framing.TxCommitBody; -import org.apache.qpid.framing.TxRollbackBody; -import org.apache.qpid.framing.TxSelectBody; +import org.apache.qpid.AMQConnectionException; +import org.apache.qpid.framing.*; import org.apache.qpid.protocol.AMQMethodEvent; import org.apache.qpid.protocol.AMQMethodListener; +import org.apache.qpid.protocol.AMQConstant; import org.apache.qpid.server.handler.BasicAckMethodHandler; import org.apache.qpid.server.handler.BasicCancelMethodHandler; import org.apache.qpid.server.handler.BasicConsumeMethodHandler; @@ -107,43 +81,35 @@ public class AMQStateManager implements AMQMethodListener * Maps from an AMQState instance to a Map from Class to StateTransitionHandler. The class must be a subclass of * AMQFrame. */ - private final EnumMap<AMQState, Map<Class<? extends AMQMethodBody>, StateAwareMethodListener<? extends AMQMethodBody>>> _state2HandlersMap = +/* private final EnumMap<AMQState, Map<Class<? extends AMQMethodBody>, StateAwareMethodListener<? extends AMQMethodBody>>> _state2HandlersMap = new EnumMap<AMQState, Map<Class<? extends AMQMethodBody>, StateAwareMethodListener<? extends AMQMethodBody>>>( AMQState.class); + */ + private CopyOnWriteArraySet<StateListener> _stateListeners = new CopyOnWriteArraySet<StateListener>(); public AMQStateManager(VirtualHostRegistry virtualHostRegistry, AMQProtocolSession protocolSession) { - this(AMQState.CONNECTION_NOT_STARTED, true, virtualHostRegistry, protocolSession); - } - protected AMQStateManager(AMQState initial, boolean register, VirtualHostRegistry virtualHostRegistry, - AMQProtocolSession protocolSession) - { _virtualHostRegistry = virtualHostRegistry; _protocolSession = protocolSession; - _currentState = initial; - if (register) - { - registerListeners(); - } + _currentState = AMQState.CONNECTION_NOT_STARTED; + } + /* protected void registerListeners() { Map<Class<? extends AMQMethodBody>, StateAwareMethodListener<? extends AMQMethodBody>> frame2handlerMap; frame2handlerMap = new HashMap<Class<? extends AMQMethodBody>, StateAwareMethodListener<? extends AMQMethodBody>>(); - frame2handlerMap.put(ConnectionStartOkBody.class, ConnectionStartOkMethodHandler.getInstance()); _state2HandlersMap.put(AMQState.CONNECTION_NOT_STARTED, frame2handlerMap); frame2handlerMap = new HashMap<Class<? extends AMQMethodBody>, StateAwareMethodListener<? extends AMQMethodBody>>(); - frame2handlerMap.put(ConnectionSecureOkBody.class, ConnectionSecureOkMethodHandler.getInstance()); _state2HandlersMap.put(AMQState.CONNECTION_NOT_AUTH, frame2handlerMap); frame2handlerMap = new HashMap<Class<? extends AMQMethodBody>, StateAwareMethodListener<? extends AMQMethodBody>>(); - frame2handlerMap.put(ConnectionTuneOkBody.class, ConnectionTuneOkMethodHandler.getInstance()); _state2HandlersMap.put(AMQState.CONNECTION_NOT_TUNED, frame2handlerMap); frame2handlerMap = new HashMap<Class<? extends AMQMethodBody>, StateAwareMethodListener<? extends AMQMethodBody>>(); @@ -154,37 +120,41 @@ public class AMQStateManager implements AMQMethodListener // ConnectionOpen handlers // frame2handlerMap = new HashMap<Class<? extends AMQMethodBody>, StateAwareMethodListener<? extends AMQMethodBody>>(); - frame2handlerMap.put(ChannelOpenBody.class, ChannelOpenHandler.getInstance()); - frame2handlerMap.put(ChannelCloseBody.class, ChannelCloseHandler.getInstance()); - frame2handlerMap.put(ChannelCloseOkBody.class, ChannelCloseOkHandler.getInstance()); - frame2handlerMap.put(ConnectionCloseBody.class, ConnectionCloseMethodHandler.getInstance()); - frame2handlerMap.put(ExchangeDeclareBody.class, ExchangeDeclareHandler.getInstance()); - frame2handlerMap.put(ExchangeDeleteBody.class, ExchangeDeleteHandler.getInstance()); - frame2handlerMap.put(ExchangeBoundBody.class, ExchangeBoundHandler.getInstance()); - frame2handlerMap.put(BasicAckBody.class, BasicAckMethodHandler.getInstance()); - frame2handlerMap.put(BasicRecoverBody.class, BasicRecoverMethodHandler.getInstance()); - frame2handlerMap.put(BasicConsumeBody.class, BasicConsumeMethodHandler.getInstance()); - frame2handlerMap.put(BasicGetBody.class, BasicGetMethodHandler.getInstance()); - frame2handlerMap.put(BasicCancelBody.class, BasicCancelMethodHandler.getInstance()); - frame2handlerMap.put(BasicPublishBody.class, BasicPublishMethodHandler.getInstance()); - frame2handlerMap.put(BasicQosBody.class, BasicQosHandler.getInstance()); - frame2handlerMap.put(QueueBindBody.class, QueueBindHandler.getInstance()); - frame2handlerMap.put(QueueDeclareBody.class, QueueDeclareHandler.getInstance()); - frame2handlerMap.put(QueueDeleteBody.class, QueueDeleteHandler.getInstance()); - frame2handlerMap.put(QueuePurgeBody.class, QueuePurgeHandler.getInstance()); - frame2handlerMap.put(ChannelFlowBody.class, ChannelFlowHandler.getInstance()); - frame2handlerMap.put(TxSelectBody.class, TxSelectHandler.getInstance()); - frame2handlerMap.put(TxCommitBody.class, TxCommitHandler.getInstance()); - frame2handlerMap.put(TxRollbackBody.class, TxRollbackHandler.getInstance()); - frame2handlerMap.put(BasicRejectBody.class, BasicRejectMethodHandler.getInstance()); + ChannelOpenHandler.getInstance(); + ChannelCloseHandler.getInstance(); + ChannelCloseOkHandler.getInstance(); + ConnectionCloseMethodHandler.getInstance(); + ConnectionCloseOkMethodHandler.getInstance(); + ConnectionTuneOkMethodHandler.getInstance(); + ConnectionSecureOkMethodHandler.getInstance(); + ConnectionStartOkMethodHandler.getInstance(); + ExchangeDeclareHandler.getInstance(); + ExchangeDeleteHandler.getInstance(); + ExchangeBoundHandler.getInstance(); + BasicAckMethodHandler.getInstance(); + BasicRecoverMethodHandler.getInstance(); + BasicConsumeMethodHandler.getInstance(); + BasicGetMethodHandler.getInstance(); + BasicCancelMethodHandler.getInstance(); + BasicPublishMethodHandler.getInstance(); + BasicQosHandler.getInstance(); + QueueBindHandler.getInstance(); + QueueDeclareHandler.getInstance(); + QueueDeleteHandler.getInstance(); + QueuePurgeHandler.getInstance(); + ChannelFlowHandler.getInstance(); + TxSelectHandler.getInstance(); + TxCommitHandler.getInstance(); + TxRollbackHandler.getInstance(); + BasicRejectMethodHandler.getInstance(); _state2HandlersMap.put(AMQState.CONNECTION_OPEN, frame2handlerMap); frame2handlerMap = new HashMap<Class<? extends AMQMethodBody>, StateAwareMethodListener<? extends AMQMethodBody>>(); - frame2handlerMap.put(ConnectionCloseOkBody.class, ConnectionCloseOkMethodHandler.getInstance()); + _state2HandlersMap.put(AMQState.CONNECTION_CLOSING, frame2handlerMap); - } + } */ public AMQState getCurrentState() { @@ -214,18 +184,25 @@ public class AMQStateManager implements AMQMethodListener public <B extends AMQMethodBody> boolean methodReceived(AMQMethodEvent<B> evt) throws AMQException { - StateAwareMethodListener<B> handler = findStateTransitionHandler(_currentState, evt.getMethod()); - if (handler != null) - { + MethodDispatcher dispatcher = _protocolSession.getMethodDispatcher(); + + final int channelId = evt.getChannelId(); + B body = evt.getMethod(); - checkChannel(evt, _protocolSession); + if(channelId != 0 && _protocolSession.getChannel(channelId)== null) + { - handler.methodReceived(this, evt); + if(! ((body instanceof ChannelOpenBody) + || (body instanceof ChannelCloseOkBody) + || (body instanceof ChannelCloseBody))) + { + throw body.getConnectionException(AMQConstant.CHANNEL_ERROR, "channel is closed"); + } - return true; } - return false; + return body.execute(dispatcher, channelId); + } private <B extends AMQMethodBody> void checkChannel(AMQMethodEvent<B> evt, AMQProtocolSession protocolSession) @@ -239,6 +216,7 @@ public class AMQStateManager implements AMQMethodListener } } +/* protected <B extends AMQMethodBody> StateAwareMethodListener<B> findStateTransitionHandler(AMQState currentState, B frame) // throws IllegalStateTransitionException @@ -260,6 +238,7 @@ public class AMQStateManager implements AMQMethodListener return handler; } } +*/ public void addStateListener(StateListener listener) { diff --git a/java/broker/src/main/java/org/apache/qpid/server/state/StateAwareMethodListener.java b/java/broker/src/main/java/org/apache/qpid/server/state/StateAwareMethodListener.java index e3af0bc486..3c11bb8a9c 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/state/StateAwareMethodListener.java +++ b/java/broker/src/main/java/org/apache/qpid/server/state/StateAwareMethodListener.java @@ -29,7 +29,7 @@ import org.apache.qpid.protocol.AMQMethodEvent; * the opportunity to update state. * */ -public interface StateAwareMethodListener <B extends AMQMethodBody> +public interface StateAwareMethodListener<B extends AMQMethodBody> { - void methodReceived(AMQStateManager stateManager, AMQMethodEvent<B> evt) throws AMQException; + void methodReceived(AMQStateManager stateManager, B evt, int channelId) throws AMQException; } diff --git a/java/broker/src/test/java/org/apache/qpid/server/RunBrokerWithCommand.java b/java/broker/src/test/java/org/apache/qpid/server/RunBrokerWithCommand.java index b8803206e0..d8b5f5f7e6 100644 --- a/java/broker/src/test/java/org/apache/qpid/server/RunBrokerWithCommand.java +++ b/java/broker/src/test/java/org/apache/qpid/server/RunBrokerWithCommand.java @@ -43,7 +43,7 @@ public class RunBrokerWithCommand } catch (Exception e) { - System.out.println("Unable to start broker due to: " + e.getMessage()); + System.err.println("Unable to start broker due to: " + e.getMessage()); e.printStackTrace(); exit(1); @@ -55,7 +55,7 @@ public class RunBrokerWithCommand try { Process task = Runtime.getRuntime().exec(args[0]); - System.out.println("Started Proccess: " + args[0]); + System.err.println("Started Proccess: " + args[0]); InputStream inputStream = task.getInputStream(); @@ -70,19 +70,21 @@ public class RunBrokerWithCommand out.join(); err.join(); - System.out.println("Waiting for process to exit: " + args[0]); + System.err.println("Waiting for process to exit: " + args[0]); task.waitFor(); - System.out.println("Done Proccess: " + args[0]); + System.err.println("Done Proccess: " + args[0]); } catch (IOException e) { - System.out.println("Proccess had problems: " + e.getMessage()); + System.err.println("Proccess had problems: " + e.getMessage()); + e.printStackTrace(System.err); exit(1); } catch (InterruptedException e) { - System.out.println("Proccess had problems: " + e.getMessage()); + System.err.println("Proccess had problems: " + e.getMessage()); + e.printStackTrace(System.err); exit(1); } diff --git a/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueMBeanTest.java b/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueMBeanTest.java index 3caf6ad73d..4770779363 100644 --- a/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueMBeanTest.java +++ b/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueMBeanTest.java @@ -288,7 +288,7 @@ public class AMQQueueMBeanTest extends TestCase // Add the body so we have somthing to test later currentMessage.addContentBodyFrame(_storeContext, - _protocolSession.getRegistry() + _protocolSession.getMethodRegistry() .getProtocolVersionMethodConverter() .convertToContentChunk( new ContentBody(ByteBuffer.allocate((int) MESSAGE_SIZE), |