From commits-return-8401-apmail-activemq-commits-archive=activemq.apache.org@activemq.apache.org Wed Apr 09 15:28:05 2008 Return-Path: Delivered-To: apmail-activemq-commits-archive@www.apache.org Received: (qmail 68113 invoked from network); 9 Apr 2008 15:28:04 -0000 Received: from hermes.apache.org (HELO mail.apache.org) (140.211.11.2) by minotaur.apache.org with SMTP; 9 Apr 2008 15:28:04 -0000 Received: (qmail 43168 invoked by uid 500); 9 Apr 2008 15:28:05 -0000 Delivered-To: apmail-activemq-commits-archive@activemq.apache.org Received: (qmail 43145 invoked by uid 500); 9 Apr 2008 15:28:05 -0000 Mailing-List: contact commits-help@activemq.apache.org; run by ezmlm Precedence: bulk List-Help: List-Unsubscribe: List-Post: List-Id: Reply-To: dev@activemq.apache.org Delivered-To: mailing list commits@activemq.apache.org Received: (qmail 43136 invoked by uid 99); 9 Apr 2008 15:28:05 -0000 Received: from athena.apache.org (HELO athena.apache.org) (140.211.11.136) by apache.org (qpsmtpd/0.29) with ESMTP; Wed, 09 Apr 2008 08:28:05 -0700 X-ASF-Spam-Status: No, hits=-2000.0 required=10.0 tests=ALL_TRUSTED X-Spam-Check-By: apache.org Received: from [140.211.11.3] (HELO eris.apache.org) (140.211.11.3) by apache.org (qpsmtpd/0.29) with ESMTP; Wed, 09 Apr 2008 15:27:31 +0000 Received: by eris.apache.org (Postfix, from userid 65534) id BAE2F1A9832; Wed, 9 Apr 2008 08:27:43 -0700 (PDT) Content-Type: text/plain; charset="utf-8" MIME-Version: 1.0 Content-Transfer-Encoding: 7bit Subject: svn commit: r646401 - /activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/failover/FailoverTransport.java Date: Wed, 09 Apr 2008 15:27:43 -0000 To: commits@activemq.apache.org From: rajdavies@apache.org X-Mailer: svnmailer-1.0.8 Message-Id: <20080409152743.BAE2F1A9832@eris.apache.org> X-Virus-Checked: Checked by ClamAV on apache.org Author: rajdavies Date: Wed Apr 9 08:27:42 2008 New Revision: 646401 URL: http://svn.apache.org/viewvc?rev=646401&view=rev Log: Fix for https://issues.apache.org/activemq/browse/AMQ-1488 Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/failover/FailoverTransport.java Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/failover/FailoverTransport.java URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/failover/FailoverTransport.java?rev=646401&r1=646400&r2=646401&view=diff ============================================================================== --- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/failover/FailoverTransport.java (original) +++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/failover/FailoverTransport.java Wed Apr 9 08:27:42 2008 @@ -22,9 +22,10 @@ import java.net.URI; import java.util.ArrayList; import java.util.Iterator; +import java.util.LinkedHashMap; import java.util.List; +import java.util.Map; import java.util.Random; -import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.atomic.AtomicReference; @@ -69,7 +70,7 @@ private final Object backupMutex = new Object(); private final Object sleepMutex = new Object(); private final ConnectionStateTracker stateTracker = new ConnectionStateTracker(); - private final ConcurrentHashMap requestMap = new ConcurrentHashMap(); + private final Map requestMap = new LinkedHashMap(); private URI connectedTransportURI; private URI failedConnectTransportURI; @@ -139,7 +140,10 @@ return; } if (command.isResponse()) { - Object object = requestMap.remove(Integer.valueOf(((Response)command).getCorrelationId())); + Object object = null; + synchronized(requestMap) { + object = requestMap.remove(Integer.valueOf(((Response)command).getCorrelationId())); + } if (object != null && object.getClass() == Tracked.class) { ((Tracked)object).onResponses(); } @@ -426,10 +430,12 @@ // then hold it in the requestMap so that we can replay // it later. Tracked tracked = stateTracker.track(command); - if (tracked != null && tracked.isWaitingForResponse()) { - requestMap.put(Integer.valueOf(command.getCommandId()), tracked); - } else if (tracked == null && command.isResponseRequired()) { - requestMap.put(Integer.valueOf(command.getCommandId()), command); + synchronized(requestMap) { + if (tracked != null && tracked.isWaitingForResponse()) { + requestMap.put(Integer.valueOf(command.getCommandId()), tracked); + } else if (tracked == null && command.isResponseRequired()) { + requestMap.put(Integer.valueOf(command.getCommandId()), command); + } } // Send the message. @@ -581,7 +587,11 @@ cc.setFaultTolerant(true); t.oneway(cc); stateTracker.restore(t); - for (Iterator iter2 = requestMap.values().iterator(); iter2.hasNext();) { + Map tmpMap = null; + synchronized(requestMap) { + tmpMap = new LinkedHashMap(requestMap); + } + for (Iterator iter2 = tmpMap.values().iterator(); iter2.hasNext();) { Command command = iter2.next(); t.oneway(command); }