From 72de13352dc9c42acfe95a1d76f049c507eb5cfd Mon Sep 17 00:00:00 2001 From: Kim van der Riet Date: Fri, 16 Feb 2007 16:02:26 +0000 Subject: Additions to allow refs to be sent from broker to client. Also some tidy-up. git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/branches/qpid.0-9@508460 13f79535-47bb-0310-9956-ffa450edef68 --- .../java/org/apache/qpid/server/AMQChannel.java | 131 ++++----------------- .../handler/ConnectionCloseOkMethodHandler.java | 6 + .../qpid/server/handler/QueueDeclareHandler.java | 7 +- .../server/protocol/AMQMinaProtocolSession.java | 18 ++- .../server/protocol/AMQPFastProtocolHandler.java | 2 + 5 files changed, 52 insertions(+), 112 deletions(-) (limited to 'java/broker/src/main') 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 c5d0220989..b2d4215bd0 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 @@ -309,33 +309,6 @@ public class AMQChannel _returnMessages.add(e); } } -// -// public void deliver(AMQMessage msg, AMQShortString destination, final long deliveryTag) -// { -// deliver(msg, destination, new AMQMethodListener() -// { -// public boolean methodReceived(AMQMethodEvent evt) throws AMQException -// { -// AMQMethodBody method = evt.getMethod(); -// if (_log.isDebugEnabled()) -// { -// _log.debug(method + " received on channel " + _channelId); -// } -// // XXX: multiple? -// if (method instanceof MessageOkBody) -// { -// acknowledgeMessage(deliveryTag, false); -// return true; -// } -// else -// { -// // TODO: implement reject -// return false; -// } -// } -// public void error(Exception e) {} -// }); -// } public void deliver(AMQMessage msg, AMQShortString destination, final long deliveryTag) { @@ -343,7 +316,6 @@ public class AMQChannel long maxFrameSize = _session.getFrameMax(); Iterable contentItr = msg.getContents(); if (msg.getSize() > maxFrameSize) - //if(true) { Iterator cItr = contentItr.iterator(); if (cItr.next().limit() > maxFrameSize) // First chunk should equal incoming frame size @@ -410,34 +382,29 @@ public class AMQChannel public void deliverRef(final AMQMessage msg, final AMQShortString destination, final long deliveryTag) { final byte[] refId = String.valueOf(System.currentTimeMillis()).getBytes(); - deliverRef(refId, msg, destination, _session.getStateManager()); -// AMQMethodBody openBody = MessageOpenBody.createMethodBody( -// _session.getProtocolMajorVersion(), // AMQP major version -// _session.getProtocolMinorVersion(), // AMQP minor version -// refId); -// _session.writeRequest(_channelId, openBody, new AMQMethodListener() -// { -// public boolean methodReceived(AMQMethodEvent evt) throws AMQException -// { -// AMQMethodBody method = evt.getMethod(); -// if (_log.isDebugEnabled()) -// { -// _log.debug(method + " received on channel " + _channelId); -// } -// if (method instanceof MessageOkBody) -// { -// acknowledgeMessage(deliveryTag, false); -// deliverRef(refId, msg, destination, _session.getStateManager()); -// return true; -// } -// else -// { -// // TODO: implement reject -// return false; -// } -// } -// public void error(Exception e) {} -// }); + deliverRef(refId, msg, destination, new AMQMethodListener() + { + public boolean methodReceived(AMQMethodEvent evt) throws AMQException + { + AMQMethodBody method = evt.getMethod(); + if (_log.isDebugEnabled()) + { + _log.debug(method + " received on channel " + _channelId); + } + // XXX: multiple? + if (method instanceof MessageOkBody) + { + acknowledgeMessage(deliveryTag, false); + return true; + } + else + { + // TODO: implement reject + return false; + } + } + public void error(Exception e) {} + }); } public void deliverRef(byte[] refId, AMQMessage msg, AMQShortString destination, AMQMethodListener listener) @@ -471,58 +438,6 @@ public class AMQChannel _session.writeRequest(_channelId, closeBody, listener); } -// protected void route(AMQMessage msg) throws AMQException -// { -// if (isTransactional()) -// { -// //don't create a transaction unless needed -// if (msg.isPersistent()) -// { -// // _txnBuffer.containsPersistentChanges(); -// } -// -// //A publication will result in the enlisting of several -// //TxnOps. The first is an op that will store the message. -// //Following that (and ordering is important), an op will -// //be added for every queue onto which the message is -// //enqueued. Finally a cleanup op will be added to decrement -// //the reference associated with the routing. -// // Store storeOp = new Store(msg); -// // _txnBuffer.enlist(storeOp); -// // msg.setTxnBuffer(_txnBuffer); -// try -// { -// _exchanges.routeContent(msg); -// // _txnBuffer.enlist(new Cleanup(msg)); -// } -// catch (RequiredDeliveryException e) -// { -// //Can only be due to the mandatory flag, as no attempt -// //has yet been made to deliver the message. The -// //message will thus not have been delivered to any -// //queue so we can return the message (without killing -// //the transaction) and for efficiency remove the store -// //operation from the buffer. -// // _txnBuffer.cancel(storeOp); -// throw e; -// } -// } -// else -// { -// try -// { -// _exchanges.routeContent(msg); -// //following check implements the functionality -// //required by the 'immediate' flag: -// msg.checkDeliveredToConsumer(); -// } -// finally -// { -// msg.decrementReference(_storeContext); -// } -// } -// } - public RequestManager getRequestManager() { return _requestManager; diff --git a/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionCloseOkMethodHandler.java b/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionCloseOkMethodHandler.java index 9743ff713b..e2aff8cd22 100644 --- a/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionCloseOkMethodHandler.java +++ b/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionCloseOkMethodHandler.java @@ -46,6 +46,12 @@ public class ConnectionCloseOkMethodHandler implements StateAwareMethodListener< { AMQProtocolSession session = stateManager.getProtocolSession(); _logger.info("Received Connection-close-ok"); + // We wait for the Mina library to close the connection, which will happen when + // the client closes the Mina connection, causing AMQFastProtocolHand.sessionClosed() + // to be called. + // TODO - Find a better way of doing this without holding up this thread... + try { Thread.currentThread().sleep(2000); } // 2 seconds + catch (InterruptedException e) {} session.closeSession(); } } 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 4ae80480aa..b7ad8790f5 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 @@ -29,6 +29,7 @@ import org.apache.qpid.framing.QueueDeclareBody; import org.apache.qpid.framing.QueueDeclareOkBody; import org.apache.qpid.protocol.AMQConstant; import org.apache.qpid.protocol.AMQMethodEvent; +import org.apache.qpid.server.AMQChannel; import org.apache.qpid.server.configuration.Configurator; import org.apache.qpid.server.exchange.Exchange; import org.apache.qpid.server.exchange.ExchangeRegistry; @@ -120,7 +121,11 @@ public class QueueDeclareHandler implements StateAwareMethodListener