svn commit: r669841 - in /incubator/qpid/trunk/qpid/java: broker/src/main/java/org/apache/qpid/server/txn/LocalTransactionalContext.java common/src/main/java/org/apache/qpid/AMQChannelException.java

[rgodfrey]

Author: rgodfrey
Date: Fri Jun 20 02:37:20 2008
New Revision: 669841

URL: http://svn.apache.org/viewvc?rev=669841&view=rev
Log:
QPID-1144 : Reference count drops to zero too early for immediate messages in a 
txn

Modified:
    
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/txn/LocalTransactionalContext.java
    
incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/AMQChannelException.java

Modified: 
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/txn/LocalTransactionalContext.java
URL: 
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/txn/LocalTransactionalContext.java?rev=669841&r1=669840&r2=669841&view=diff
==============================================================================
--- 
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/txn/LocalTransactionalContext.java
 (original)
+++ 
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/txn/LocalTransactionalContext.java
 Fri Jun 20 02:37:20 2008
@@ -93,11 +93,19 @@
         public void process() throws AMQException
         {
 
-            QueueEntry entry = _queue.enqueue(getStoreContext(),_message);
+            _message.incrementReference();
+            try
+            {
+                QueueEntry entry = _queue.enqueue(getStoreContext(),_message);
 
-            if(entry.immediateAndNotDelivered())
+                if(entry.immediateAndNotDelivered())
+                {
+                    getReturnMessages().add(new 
NoConsumersException(_message));
+                }
+            }
+            finally
             {
-                getReturnMessages().add(new NoConsumersException(_message));
+                _message.decrementReference(getStoreContext());
             }
         }
     }

Modified: 
incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/AMQChannelException.java
URL: 
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/AMQChannelException.java?rev=669841&r1=669840&r2=669841&view=diff
==============================================================================
--- 
incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/AMQChannelException.java
 (original)
+++ 
incubator/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/AMQChannelException.java
 Fri Jun 20 02:37:20 2008
@@ -54,6 +54,6 @@
     public AMQFrame getCloseFrame(int channel)
     {
         MethodRegistry reg = MethodRegistry.getMethodRegistry(new 
ProtocolVersion(major,minor));
-        return new AMQFrame(channel, 
reg.createChannelCloseBody(getErrorCode().getCode(), new 
AMQShortString(getMessage()),_classId,_methodId));
+        return new AMQFrame(channel, reg.createChannelCloseBody(getErrorCode() 
== null ? AMQConstant.INTERNAL_ERROR.getCode() : getErrorCode().getCode(), new 
AMQShortString(getMessage()),_classId,_methodId));
     }
 }