resolve: https://issues.apache.org/activemq/browse/AMQ-2867, addition to https://issues.apache.org/activemq/browse/AMQ-2507 - if no response received and timeout > 0 a RequestTimedOutIOException is thrown on the client side

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

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

@@ -49,7 +49,11 @@ public class FutureResponse {
 
     public Response getResult(int timeout) throws IOException {
         try {
-            return responseSlot.poll(timeout, TimeUnit.MILLISECONDS);
+            Response result = responseSlot.poll(timeout, TimeUnit.MILLISECONDS);
+            if (result == null && timeout > 0) {
+                throw new RequestTimedOutIOException();
+            }
+            return result;
         } catch (InterruptedException e) {
             throw new InterruptedIOException("Interrupted.");
         }

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

@@ -0,0 +1,26 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (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
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.transport;
+
+import java.io.IOException;
+
+/**
+ * thrown when the timeout specified on a request expires before
+ * a reply or response is received
+ */
+public class RequestTimedOutIOException extends IOException {
+}

activemq-core/src/test/java/org/apache/activemq/bugs/JmsTimeoutTest.java

@@ -17,6 +17,7 @@
 package org.apache.activemq.bugs;
 
 
+import java.io.IOException;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import javax.jms.DeliveryMode;
@@ -29,6 +30,7 @@ import javax.jms.TextMessage;
 import org.apache.activemq.ActiveMQConnection;
 import org.apache.activemq.EmbeddedBrokerTestSupport;
 import org.apache.activemq.command.ActiveMQDestination;
+import org.apache.activemq.transport.RequestTimedOutIOException;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
@@ -50,7 +52,7 @@ public class JmsTimeoutTest extends EmbeddedBrokerTestSupport {
 	        final ActiveMQConnection cx = (ActiveMQConnection)createConnection();
 	        final ActiveMQDestination queue = createDestination("testqueue");
 	        
-	        // we should not take longer than 5 seconds to return from send
+	        // we should not take longer than 10 seconds to return from send
 	        cx.setSendTimeout(10000);
 	        	
 	        Runnable r = new Runnable() {
@@ -64,18 +66,61 @@ public class JmsTimeoutTest extends EmbeddedBrokerTestSupport {
 	                    TextMessage message = session.createTextMessage(createMessageText());
 	                    for(int count=0; count<messageCount; count++){
 	                    	producer.send(message);
-	                    	// Currently after the timeout producer just
-	                    	// returns but there is no way to know that
-	                    	// the send timed out
 	                    }	  
 	                    LOG.info("Done sending..");
-	                } catch (JMSException e) {
+                    } catch (JMSException e) {
-	                    e.printStackTrace();
+                        if (e.getCause() instanceof RequestTimedOutIOException) {
-	                    if (e instanceof ResourceAllocationException) {
 	                        exceptionCount.incrementAndGet();
-	                    }
+                        } else {
+                            e.printStackTrace();
+                        }
 	                    return;
 	                }
+
+	            }
+	        };
+	        cx.start();
+	        Thread producerThread = new Thread(r);
+	        producerThread.start();
+	        producerThread.join(30000);
+	        cx.close();
+	        // We should have a few timeout exceptions as memory store will fill up
+	        assertTrue("No exception from the broker", exceptionCount.get() > 0);
+	    }
+
+
+        /**
+	     * Test the case where the broker is blocked due to a memory limit
+	     * with a fail timeout
+	     * @throws Exception
+	     */
+	    public void testBlockedProducerUsageSendFailTimeout() throws Exception {
+	        final ActiveMQConnection cx = (ActiveMQConnection)createConnection();
+	        final ActiveMQDestination queue = createDestination("testqueue");
+
+            broker.getSystemUsage().setSendFailIfNoSpaceAfterTimeout(5000);
+	        Runnable r = new Runnable() {
+	            public void run() {
+	                try {
+	                	LOG.info("Sender thread starting");
+	                    Session session = cx.createSession(false, 1);
+	                    MessageProducer producer = session.createProducer(queue);
+	                    producer.setDeliveryMode(DeliveryMode.PERSISTENT);
+
+	                    TextMessage message = session.createTextMessage(createMessageText());
+	                    for(int count=0; count<messageCount; count++){
+	                    	producer.send(message);
+	                    }
+	                    LOG.info("Done sending..");
+                    } catch (JMSException e) {
+                        if (e instanceof ResourceAllocationException || e.getCause() instanceof RequestTimedOutIOException) {
+	                        exceptionCount.incrementAndGet();
+                        } else {
+                            e.printStackTrace();
+                        }
+	                    return;
+	                }
+
 	            }
 	        };
 	        cx.start();
@@ -88,11 +133,12 @@ public class JmsTimeoutTest extends EmbeddedBrokerTestSupport {
 	    }
 
 	    protected void setUp() throws Exception {
+            exceptionCount.set(0);
 	        bindAddress = "tcp://localhost:61616";
 	        broker = createBroker();
 	        broker.setDeleteAllMessagesOnStartup(true);
 	        broker.getSystemUsage().getMemoryUsage().setLimit(5*1024*1024);
-	        broker.getSystemUsage().setSendFailIfNoSpaceAfterTimeout(5000);
+
 	        super.setUp();
 	    }