small op

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@509728 13f79535-47bb-0310-9956-ffa450edef68

activemq-core/src/main/java/org/apache/activemq/transport/ResponseCorrelator.java

@@ -1,49 +1,45 @@
 /**
  * 
- * Licensed to the Apache Software Foundation (ASF) under one or more
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements. See the NOTICE
- * contributor license agreements.  See the NOTICE file distributed with
+ * file distributed with this work for additional information regarding copyright ownership. The ASF licenses this file
- * this work for additional information regarding copyright ownership.
+ * to You under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the
- * The ASF licenses this file to You under the Apache License, Version 2.0
+ * License. You may obtain a copy of the License at
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
  * 
  * http://www.apache.org/licenses/LICENSE-2.0
  * 
- * Unless required by applicable law or agreed to in writing, software
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
- * distributed under the License is distributed on an "AS IS" BASIS,
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * specific language governing permissions and limitations under the License.
- * See the License for the specific language governing permissions and
- * limitations under the License.
  */
+
 package org.apache.activemq.transport;
 
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.Iterator;
-
+import java.util.Map;
 import org.apache.activemq.command.Command;
 import org.apache.activemq.command.ExceptionResponse;
 import org.apache.activemq.command.Response;
 import org.apache.activemq.util.IntSequenceGenerator;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
-
 import java.util.concurrent.ConcurrentHashMap;
 
-
 /**
- * Adds the incrementing sequence number to commands along with performing the corelation of
+ * Adds the incrementing sequence number to commands along with performing the corelation of responses to requests to
- * responses to requests to create a blocking request-response semantics.
+ * create a blocking request-response semantics.
  * 
  * @version $Revision: 1.4 $
  */
 public class ResponseCorrelator extends TransportFilter{
 
     private static final Log log=LogFactory.getLog(ResponseCorrelator.class);
-    
+    private final Map requestMap=new HashMap();
-    private final ConcurrentHashMap requestMap = new ConcurrentHashMap();
     private IntSequenceGenerator sequenceGenerator;
     private final boolean debug=log.isDebugEnabled();
+
     public ResponseCorrelator(Transport next){
         this(next,new IntSequenceGenerator());
     }
@@ -65,7 +61,9 @@ public class ResponseCorrelator extends TransportFilter {
         command.setCommandId(sequenceGenerator.getNextSequenceId());
         command.setResponseRequired(true);
         FutureResponse future=new FutureResponse(responseCallback);
+        synchronized(requestMap){
             requestMap.put(new Integer(command.getCommandId()),future);
+        }
         next.oneway(command);
         return future;
     }
@@ -82,14 +80,17 @@ public class ResponseCorrelator extends TransportFilter {
 
     public void onCommand(Object o){
         Command command=(Command)o;
-        
         if(command.isResponse()){
             Response response=(Response)command;
-            FutureResponse future = (FutureResponse) requestMap.remove(new Integer(response.getCorrelationId()));
+            FutureResponse future=null;
+            synchronized(requestMap){
+                future=(FutureResponse)requestMap.remove(new Integer(response.getCorrelationId()));
+            }
             if(future!=null){
                 future.set(response);
             }else{
-                if( debug ) log.debug("Received unexpected response for command id: "+response.getCorrelationId());
+                if(debug)
+                    log.debug("Received unexpected response for command id: "+response.getCorrelationId());
             }
         }else{
             getTransportListener().onCommand(command);
@@ -97,20 +98,17 @@ public class ResponseCorrelator extends TransportFilter {
     }
 
     /**
-     * If an async exception occurs, then assume no responses will arrive for any of
+     * If an async exception occurs, then assume no responses will arrive for any of current requests. Lets let them
-     * current requests.  Lets let them know of the problem.
+     * know of the problem.
      */
     public void onException(IOException error){
-        
         // Copy and Clear the request Map
         ArrayList requests=new ArrayList(requestMap.values());
         requestMap.clear();
-        
         for(Iterator iter=requests.iterator();iter.hasNext();){
             FutureResponse fr=(FutureResponse)iter.next();
             fr.set(new ExceptionResponse(error));
         }
-        
         super.onException(error);
     }
 
@@ -121,5 +119,4 @@ public class ResponseCorrelator extends TransportFilter {
     public String toString(){
         return next.toString();
     }
-
 }