[jboss-cvs] JBoss Messaging SVN: r8329 - branches/JBMESSAGING_1878/src/main/org/jboss/jms/client/container.

[jboss-cvs-commits at lists.jboss.org]

Author: gaohoward
Date: 2011-06-13 04:25:00 -0400 (Mon, 13 Jun 2011)
New Revision: 8329

Modified:
   branches/JBMESSAGING_1878/src/main/org/jboss/jms/client/container/ClientConsumer.java
Log:
fix issues


Modified: branches/JBMESSAGING_1878/src/main/org/jboss/jms/client/container/ClientConsumer.java
===================================================================
--- branches/JBMESSAGING_1878/src/main/org/jboss/jms/client/container/ClientConsumer.java	2011-06-13 05:50:11 UTC (rev 8328)
+++ branches/JBMESSAGING_1878/src/main/org/jboss/jms/client/container/ClientConsumer.java	2011-06-13 08:25:00 UTC (rev 8329)
@@ -389,13 +389,15 @@
       {
          proxy.setJMSDestination(msg.getOriginalSuckerDestination());
       }
+      
+      proxy.setSource(cbManager);
 
       //TODO - we temporarily need to execute on a different thread to avoid a deadlock situation in
       //       failover where a message is sent then the valve is locked, and the message send cause
       //       a message delivery back to the same client which tries to ack but can't get through
       //       the valve. This won't be necessary when we move to a non blocking transport
       
-      sessionExecutor.execute(new HandleMessageRunnable(currentToken, proxy, cbManager));
+      sessionExecutor.execute(new HandleMessageRunnable(currentToken, proxy));
    }
    
    public void setMessageListener(MessageListener listener) throws JMSException
@@ -1098,15 +1100,11 @@
    	
    	private MessageProxy proxy;
    	
-   	private Object source;
-   	
-   	HandleMessageRunnable(int token, MessageProxy proxy, Object source)
+   	HandleMessageRunnable(int token, MessageProxy proxy)
    	{
    		this.token = token;
    		
    		this.proxy = proxy;
-   		
-   		this.source = source;
    	}
    	
    	public void run()
@@ -1140,9 +1138,8 @@
                    synchronized (consumerLock)
                    {
                       //if source changed, discard it.
-                      if (source == messageSource)
+                      if (proxy.getSource() == messageSource)
                       {
-                         proxy.setSource(source);
                          //Add it to the buffer
                          buffer.addLast(proxy, proxy.getJMSPriority());
 
@@ -1154,7 +1151,7 @@
                       }
                       else
                       {
-                         log.debug("Discarding message from old source " + source + " on to new source " + messageSource);
+                         log.debug("Discarding message from old source " + proxy.getSource() + " on to new source " + messageSource);
                       }
                    }
                 }