svn commit: r395641 - /incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/ResponseCorrelator.java

[ch...@apache.org]

Author: chirino
Date: Thu Apr 20 09:58:58 2006
New Revision: 395641

URL: http://svn.apache.org/viewcvs?rev=395641&view=rev
Log:
If we get an async exception, report it to all blocked sync requests.
https://issues.apache.org/activemq/browse/AMQ-691

Modified:
    incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/ResponseCorrelator.java

Modified: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/ResponseCorrelator.java
URL: http://svn.apache.org/viewcvs/incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/ResponseCorrelator.java?rev=395641&r1=395640&r2=395641&view=diff
==============================================================================
--- incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/ResponseCorrelator.java (original)
+++ incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/ResponseCorrelator.java Thu Apr 20 09:58:58 2006
@@ -18,8 +18,12 @@
 
 import java.io.IOException;
 import java.io.InterruptedIOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Iterator;
 
 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;
@@ -92,6 +96,28 @@
         } else {
             getTransportListener().onCommand(command);
         }
+    }
+    
+    /**
+     * If an async exception occurs, then assume no responses will arrive for any of
+     * current requests.  Lets let them 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();) {
+            try {
+                FutureResponse fr = (FutureResponse) iter.next();
+                fr.set(new ExceptionResponse(error));
+            } catch (InterruptedIOException e) {
+                Thread.currentThread().interrupt();
+            }
+        }
+        
+        super.onException(error);
     }
     
     public IntSequenceGenerator getSequenceGenerator() {