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 + .../client/handler/MessageAppendMethodHandler.java | 9 +- .../client/handler/MessageCloseMethodHandler.java | 9 +- .../client/handler/MessageOpenMethodHandler.java | 16 ++- .../qpid/client/message/UnprocessedMessage.java | 17 ++- .../test/unit/basic/PubSubTwoConnectionTest.java | 36 +++--- 10 files changed, 109 insertions(+), 142 deletions(-) (limited to 'java') 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 StateAwareMethodListenerAppending message content to body"); + + // Be aware of possible changes to parameter order as versions change. + final AMQMethodBody methodBody = MessageOkBody.createMethodBody( + protocolSession.getProtocolMajorVersion(), // AMQP major version + protocolSession.getProtocolMinorVersion()); // AMQP minor version + protocolSession.writeResponse(evt.getChannelId(), evt.getRequestId(), methodBody); } catch (Exception e) { diff --git a/java/client/src/main/java/org/apache/qpid/client/handler/MessageCloseMethodHandler.java b/java/client/src/main/java/org/apache/qpid/client/handler/MessageCloseMethodHandler.java index 284b7444d2..62fde4d806 100644 --- a/java/client/src/main/java/org/apache/qpid/client/handler/MessageCloseMethodHandler.java +++ b/java/client/src/main/java/org/apache/qpid/client/handler/MessageCloseMethodHandler.java @@ -24,7 +24,9 @@ import org.apache.qpid.AMQException; import org.apache.qpid.client.protocol.AMQProtocolSession; import org.apache.qpid.client.state.AMQStateManager; import org.apache.qpid.client.state.StateAwareMethodListener; +import org.apache.qpid.framing.AMQMethodBody; import org.apache.qpid.framing.MessageCloseBody; +import org.apache.qpid.framing.MessageOkBody; import org.apache.qpid.protocol.AMQMethodEvent; import org.apache.log4j.Logger; @@ -46,10 +48,15 @@ public class MessageCloseMethodHandler implements StateAwareMethodListener { MessageCloseBody body = (MessageCloseBody)evt.getMethod(); String referenceId = new String(body.getReference()); - System.out.println("Message.closing()-->Handing message to session"); protocolSession.deliverMessageToAMQSession(evt.getChannelId(), referenceId); _logger.debug("Method Close Body received, notify session to accept unprocessed message"); + + // Be aware of possible changes to parameter order as versions change. + final AMQMethodBody methodBody = MessageOkBody.createMethodBody( + protocolSession.getProtocolMajorVersion(), // AMQP major version + protocolSession.getProtocolMinorVersion()); // AMQP minor version + protocolSession.writeResponse(evt.getChannelId(), evt.getRequestId(), methodBody); } } diff --git a/java/client/src/main/java/org/apache/qpid/client/handler/MessageOpenMethodHandler.java b/java/client/src/main/java/org/apache/qpid/client/handler/MessageOpenMethodHandler.java index 06d9c9ff99..8de85accdf 100644 --- a/java/client/src/main/java/org/apache/qpid/client/handler/MessageOpenMethodHandler.java +++ b/java/client/src/main/java/org/apache/qpid/client/handler/MessageOpenMethodHandler.java @@ -25,7 +25,9 @@ import org.apache.qpid.client.message.UnprocessedMessage; import org.apache.qpid.client.protocol.AMQProtocolSession; import org.apache.qpid.client.state.AMQStateManager; import org.apache.qpid.client.state.StateAwareMethodListener; +import org.apache.qpid.framing.AMQMethodBody; import org.apache.qpid.framing.MessageOpenBody; +import org.apache.qpid.framing.MessageOkBody; import org.apache.qpid.protocol.AMQMethodEvent; //import org.apache.log4j.Logger; @@ -45,11 +47,15 @@ public class MessageOpenMethodHandler implements StateAwareMethodListener public void methodReceived (AMQStateManager stateManager, AMQProtocolSession protocolSession, AMQMethodEvent evt) throws AMQException { - final UnprocessedMessage msg = new UnprocessedMessage(evt.getChannelId(), evt.getRequestId(), null, false); - String referenceId = new String(((MessageOpenBody)evt.getMethod()).getReference()); - protocolSession.unprocessedMessageReceived(referenceId, msg); - - System.out.println("Message.open()-->Adding message to map with ref"); + byte[] referenceId = ((MessageOpenBody)evt.getMethod()).getReference(); + final UnprocessedMessage msg = new UnprocessedMessage(evt.getChannelId(), evt.getRequestId(), referenceId); + protocolSession.unprocessedMessageReceived(new String(referenceId), msg); + + // Be aware of possible changes to parameter order as versions change. + final AMQMethodBody methodBody = MessageOkBody.createMethodBody( + protocolSession.getProtocolMajorVersion(), // AMQP major version + protocolSession.getProtocolMinorVersion()); // AMQP minor version + protocolSession.writeResponse(evt.getChannelId(), evt.getRequestId(), methodBody); } } diff --git a/java/client/src/main/java/org/apache/qpid/client/message/UnprocessedMessage.java b/java/client/src/main/java/org/apache/qpid/client/message/UnprocessedMessage.java index f242812d3b..3e5efd9068 100644 --- a/java/client/src/main/java/org/apache/qpid/client/message/UnprocessedMessage.java +++ b/java/client/src/main/java/org/apache/qpid/client/message/UnprocessedMessage.java @@ -35,11 +35,19 @@ public class UnprocessedMessage { private int bytesReceived = 0; private int channelId; + private byte[] referenceId; private List contents = new LinkedList(); private long deliveryTag; private boolean redeliveredFlag; private MessageHeaders messageHeaders; + public UnprocessedMessage(int channelId, long deliveryTag, byte[] referenceId) + { + this.channelId = channelId; + this.deliveryTag = deliveryTag; + this.referenceId = referenceId; + } + public UnprocessedMessage(int channelId, long deliveryTag, MessageHeaders messageHeaders, boolean redeliveredFlag) { this.channelId = channelId; @@ -73,6 +81,11 @@ public class UnprocessedMessage return channelId; } + public byte[] getReferenceId() + { + return referenceId; + } + public List getContents() { return contents; @@ -95,7 +108,9 @@ public class UnprocessedMessage public String toString() { - return "UnprocessedMessage: ch=" + channelId + "; bytesReceived=" + bytesReceived + "; deliveryTag=" + deliveryTag + "; MsgHdrs=" + messageHeaders + "Num contents=" + contents.size() + "; First content=" + new String(contents.get(0)); + return "UnprocessedMessage: ch=" + channelId + "; bytesReceived=" + bytesReceived + "; deliveryTag=" + + deliveryTag + "; MsgHdrs=" + messageHeaders + "Num contents=" + contents.size() + "; First content=" + + new String(contents.get(0)); } public void setMessageHeaders(MessageHeaders messageHeaders) { diff --git a/java/client/src/test/java/org/apache/qpid/test/unit/basic/PubSubTwoConnectionTest.java b/java/client/src/test/java/org/apache/qpid/test/unit/basic/PubSubTwoConnectionTest.java index 5c7f249107..27a312f3df 100644 --- a/java/client/src/test/java/org/apache/qpid/test/unit/basic/PubSubTwoConnectionTest.java +++ b/java/client/src/test/java/org/apache/qpid/test/unit/basic/PubSubTwoConnectionTest.java @@ -65,27 +65,19 @@ public class PubSubTwoConnectionTest extends TestCase assertEquals("Hello", tm1.getText()); } - public static void main(String[] args){ - PubSubTwoConnectionTest test = new PubSubTwoConnectionTest(); - try { - //test.setUp(); - //test.testTwoConnections(); - int a = 5; - - System.out.println(a++); - System.out.println(a); - System.out.println(++a); - - int b = ++a; - int c = a++; - - System.out.println(b); - System.out.println(c); - System.out.println(a); - - } catch (Exception e) { - // TODO Auto-generated catch block - e.printStackTrace(); - } + public static void main(String[] args) + { + PubSubTwoConnectionTest test = new PubSubTwoConnectionTest(); + try + { + test.setUp(); + test.testTwoConnections(); + test.tearDown(); + } + catch (Exception e) + { + // TODO Auto-generated catch block + e.printStackTrace(); + } } } -- cgit v1.2.1