From 66f97f32c78e0cf5914a441ae8277ee3aa659ce9 Mon Sep 17 00:00:00 2001 From: Robert Godfrey Date: Sat, 24 Nov 2007 21:14:14 +0000 Subject: 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 --- .../java/org/apache/qpid/server/AMQChannel.java | 4 +- .../qpid/server/handler/AccessRequestHandler.java | 44 ++ .../qpid/server/handler/BasicAckMethodHandler.java | 12 +- .../server/handler/BasicCancelMethodHandler.java | 29 +- .../server/handler/BasicConsumeMethodHandler.java | 97 ++-- .../qpid/server/handler/BasicGetMethodHandler.java | 24 +- .../server/handler/BasicPublishMethodHandler.java | 32 +- .../qpid/server/handler/BasicQosHandler.java | 20 +- .../server/handler/BasicRecoverMethodHandler.java | 29 +- .../handler/BasicRecoverSyncMethodHandler.java | 54 ++ .../server/handler/BasicRejectMethodHandler.java | 16 +- .../qpid/server/handler/ChannelCloseHandler.java | 19 +- .../qpid/server/handler/ChannelCloseOkHandler.java | 4 +- .../qpid/server/handler/ChannelFlowHandler.java | 26 +- .../qpid/server/handler/ChannelOpenHandler.java | 63 ++- .../handler/ConnectionCloseMethodHandler.java | 19 +- .../handler/ConnectionCloseOkMethodHandler.java | 2 +- .../handler/ConnectionOpenMethodHandler.java | 29 +- .../handler/ConnectionSecureOkMethodHandler.java | 61 +-- .../handler/ConnectionStartOkMethodHandler.java | 68 +-- .../handler/ConnectionTuneOkMethodHandler.java | 6 +- .../qpid/server/handler/ExchangeBoundHandler.java | 105 ++-- .../server/handler/ExchangeDeclareHandler.java | 41 +- .../qpid/server/handler/ExchangeDeleteHandler.java | 15 +- .../qpid/server/handler/QueueBindHandler.java | 56 +- .../qpid/server/handler/QueueDeclareHandler.java | 67 ++- .../qpid/server/handler/QueueDeleteHandler.java | 35 +- .../qpid/server/handler/QueuePurgeHandler.java | 29 +- .../server/handler/ServerMethodDispatcherImpl.java | 566 +++++++++++++++++++++ .../handler/ServerMethodDispatcherImpl_0_9.java | 160 ++++++ .../handler/ServerMethodDispatcherImpl_8_0.java | 86 ++++ .../qpid/server/handler/TxCommitHandler.java | 21 +- .../qpid/server/handler/TxRollbackHandler.java | 20 +- .../qpid/server/handler/TxSelectHandler.java | 14 +- .../server/handler/UnexpectedMethodException.java | 33 ++ .../output/ProtocolOutputConverterRegistry.java | 21 +- .../amqp0_8/ProtocolOutputConverterImpl.java | 58 ++- .../amqp0_9/ProtocolOutputConverterImpl.java | 260 ++++++++++ .../server/protocol/AMQMinaProtocolSession.java | 53 +- .../server/protocol/AMQPFastProtocolHandler.java | 23 +- .../qpid/server/protocol/AMQProtocolSession.java | 10 +- .../server/protocol/AMQProtocolSessionMBean.java | 23 +- .../org/apache/qpid/server/queue/AMQMessage.java | 2 +- .../apache/qpid/server/state/AMQStateManager.java | 129 ++--- .../server/state/StateAwareMethodListener.java | 4 +- .../apache/qpid/server/RunBrokerWithCommand.java | 14 +- .../qpid/server/queue/AMQQueueMBeanTest.java | 2 +- 47 files changed, 1843 insertions(+), 662 deletions(-) create mode 100644 java/broker/src/main/java/org/apache/qpid/server/handler/AccessRequestHandler.java create mode 100644 java/broker/src/main/java/org/apache/qpid/server/handler/BasicRecoverSyncMethodHandler.java create mode 100644 java/broker/src/main/java/org/apache/qpid/server/handler/ServerMethodDispatcherImpl.java create mode 100644 java/broker/src/main/java/org/apache/qpid/server/handler/ServerMethodDispatcherImpl_0_9.java create mode 100644 java/broker/src/main/java/org/apache/qpid/server/handler/ServerMethodDispatcherImpl_8_0.java create mode 100644 java/broker/src/main/java/org/apache/qpid/server/handler/UnexpectedMethodException.java create mode 100644 java/broker/src/main/java/org/apache/qpid/server/output/amqp0_9/ProtocolOutputConverterImpl.java (limited to 'java/broker') 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 +{ + 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 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 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 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 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 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 return _instance; } - public void methodReceived(AMQStateManager stateManager, AMQMethodEvent 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 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 +{ + 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 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 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 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 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 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 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 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 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 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 getInstance() + public static ConnectionStartOkMethodHandler getInstance() { return _instance; } @@ -62,51 +58,51 @@ public class ConnectionStartOkMethodHandler implements StateAwareMethodListener< { } - public void methodReceived(AMQStateManager stateManager, AMQMethodEvent 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 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 { @@ -64,35 +64,32 @@ public class ExchangeBoundHandler implements StateAwareMethodListener 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 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 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 { } - public void methodReceived(AMQStateManager stateManager, AMQMethodEvent 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 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 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 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 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 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 _dispatcherFactories = + new HashMap(); + + + 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 { } - public void methodReceived(AMQStateManager stateManager, AMQMethodEvent 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 { 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 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 { } - public void methodReceived(AMQStateManager stateManager, AMQMethodEvent 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> _registry = - new HashMap>(); + private static final Map _registry = + new HashMap(); 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()); - } - _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 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 _taskList = new CopyOnWriteArrayList(); - private VersionSpecificRegistry _registry = MainRegistry.getVersionSpecificRegistry(_protocolVersion); + private List _closingChannelsList = new ArrayList(); 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, StateAwareMethodListener>> _state2HandlersMap = +/* private final EnumMap, StateAwareMethodListener>> _state2HandlersMap = new EnumMap, StateAwareMethodListener>>( AMQState.class); + */ + private CopyOnWriteArraySet _stateListeners = new CopyOnWriteArraySet(); 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, StateAwareMethodListener> frame2handlerMap; frame2handlerMap = new HashMap, StateAwareMethodListener>(); - frame2handlerMap.put(ConnectionStartOkBody.class, ConnectionStartOkMethodHandler.getInstance()); _state2HandlersMap.put(AMQState.CONNECTION_NOT_STARTED, frame2handlerMap); frame2handlerMap = new HashMap, StateAwareMethodListener>(); - frame2handlerMap.put(ConnectionSecureOkBody.class, ConnectionSecureOkMethodHandler.getInstance()); _state2HandlersMap.put(AMQState.CONNECTION_NOT_AUTH, frame2handlerMap); frame2handlerMap = new HashMap, StateAwareMethodListener>(); - frame2handlerMap.put(ConnectionTuneOkBody.class, ConnectionTuneOkMethodHandler.getInstance()); _state2HandlersMap.put(AMQState.CONNECTION_NOT_TUNED, frame2handlerMap); frame2handlerMap = new HashMap, StateAwareMethodListener>(); @@ -154,37 +120,41 @@ public class AMQStateManager implements AMQMethodListener // ConnectionOpen handlers // frame2handlerMap = new HashMap, StateAwareMethodListener>(); - 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, StateAwareMethodListener>(); - 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 boolean methodReceived(AMQMethodEvent evt) throws AMQException { - StateAwareMethodListener 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 void checkChannel(AMQMethodEvent evt, AMQProtocolSession protocolSession) @@ -239,6 +216,7 @@ public class AMQStateManager implements AMQMethodListener } } +/* protected StateAwareMethodListener 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 +public interface StateAwareMethodListener { - void methodReceived(AMQStateManager stateManager, AMQMethodEvent 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), -- cgit v1.2.1