Mailing-List: contact commits-help@activemq.apache.org; run by ezmlm
Precedence: bulk
Reply-To: dev@activemq.apache.org
Content-Type: text/plain; charset="utf-8"
MIME-Version: 1.0
Content-Transfer-Encoding: 7bit
Subject: svn commit: r734114 - in
 /activemq/trunk/activemq-core/src/test/java/org/apache/activemq:
 ProducerFlowControlSendFailTest.java ProducerFlowControlTest.java
Date: Tue, 13 Jan 2009 12:10:36 -0000
To: commits@activemq.apache.org
From: gtully@apache.org
Message-Id: <20090113121044.F273E238895D@eris.apache.org>

Author: gtully
Date: Tue Jan 13 04:10:22 2009
New Revision: 734114

URL: http://svn.apache.org/viewvc?rev=734114&view=rev
Log:
add test case to exercise producerFlowControl sendFailIfNoSpace with queue memory limit

Added:
    activemq/trunk/activemq-core/src/test/java/org/apache/activemq/ProducerFlowControlSendFailTest.java   (with props)
Modified:
    activemq/trunk/activemq-core/src/test/java/org/apache/activemq/ProducerFlowControlTest.java

Added: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/ProducerFlowControlSendFailTest.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/ProducerFlowControlSendFailTest.java?rev=734114&view=auto
==============================================================================
--- activemq/trunk/activemq-core/src/test/java/org/apache/activemq/ProducerFlowControlSendFailTest.java (added)
+++ activemq/trunk/activemq-core/src/test/java/org/apache/activemq/ProducerFlowControlSendFailTest.java Tue Jan 13 04:10:22 2009
@@ -0,0 +1,124 @@
+/**
+ * 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;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+
+import javax.jms.ConnectionFactory;
+import javax.jms.ExceptionListener;
+import javax.jms.JMSException;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.ResourceAllocationException;
+import javax.jms.Session;
+import javax.jms.TextMessage;
+
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.broker.region.policy.PolicyEntry;
+import org.apache.activemq.broker.region.policy.PolicyMap;
+import org.apache.activemq.broker.region.policy.VMPendingQueueMessageStoragePolicy;
+import org.apache.activemq.broker.region.policy.VMPendingSubscriberMessageStoragePolicy;
+
+public class ProducerFlowControlSendFailTest extends ProducerFlowControlTest {
+
+    protected BrokerService createBroker() throws Exception {
+        BrokerService service = new BrokerService();
+        service.setPersistent(false);
+        service.setUseJmx(false);
+
+        // Setup a destination policy where it takes only 1 message at a time.
+        PolicyMap policyMap = new PolicyMap();
+        PolicyEntry policy = new PolicyEntry();
+        policy.setMemoryLimit(1);
+        policy.setPendingSubscriberPolicy(new VMPendingSubscriberMessageStoragePolicy());
+        policy.setPendingQueuePolicy(new VMPendingQueueMessageStoragePolicy());
+        policyMap.setDefaultEntry(policy);
+        service.setDestinationPolicy(policyMap);
+        
+        service.getSystemUsage().setSendFailIfNoSpace(true);
+
+        connector = service.addConnector("tcp://localhost:0");
+        return service;
+    }
+    
+    @Override
+    public void test2ndPubisherWithStandardConnectionThatIsBlocked() throws Exception {
+        // with sendFailIfNoSpace set, there is no blocking of the connection
+    }
+    
+    @Override
+    public void testPubisherRecoverAfterBlock() throws Exception {
+        ActiveMQConnectionFactory factory = (ActiveMQConnectionFactory)createConnectionFactory();
+        // with sendFail, there must be no flowControllwindow
+        // sendFail is an alternative flow control mechanism that does not block
+        factory.setUseAsyncSend(true);
+        connection = (ActiveMQConnection)factory.createConnection();
+        connections.add(connection);
+        connection.start();
+
+        final Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
+        final MessageProducer producer = session.createProducer(queueA);
+        
+        final AtomicBoolean keepGoing = new AtomicBoolean(true);
+   
+        Thread thread = new Thread("Filler") {
+            @Override
+            public void run() {
+                while (keepGoing.get()) {
+                    try {
+                        producer.send(session.createTextMessage("Test message"));
+                        if (gotResourceException.get()) {
+                            // do not flood the broker with requests when full as we are sending async and they 
+                            // will be limited by the network buffers
+                            Thread.sleep(200);
+                        }
+                    } catch (Exception e) {
+                        // with async send, there will be no exceptions
+                        e.printStackTrace();
+                    }
+                }
+            }
+        };
+        thread.start();
+        waitForBlockedOrResourceLimit(new AtomicBoolean(false));
+
+        // resourceException on second message, resumption if we
+        // can receive 10
+        MessageConsumer consumer = session.createConsumer(queueA);
+        TextMessage msg;
+        for (int idx = 0; idx < 10; ++idx) {
+            msg = (TextMessage) consumer.receive(1000);
+            msg.acknowledge();
+        }
+        keepGoing.set(false);
+    }
+
+    
+	@Override
+	protected ConnectionFactory createConnectionFactory() throws Exception {
+		ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(connector.getConnectUri());
+		connectionFactory.setExceptionListener(new ExceptionListener() {
+				public void onException(JMSException arg0) {
+					if (arg0 instanceof ResourceAllocationException) {
+						gotResourceException.set(true);
+					}
+				}
+	        });
+		return connectionFactory;
+	}
+}

Propchange: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/ProducerFlowControlSendFailTest.java
------------------------------------------------------------------------------
    svn:eol-style = native

Propchange: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/ProducerFlowControlSendFailTest.java
------------------------------------------------------------------------------
    svn:keywords = Rev Date

Modified: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/ProducerFlowControlTest.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/ProducerFlowControlTest.java?rev=734114&r1=734113&r2=734114&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/test/java/org/apache/activemq/ProducerFlowControlTest.java (original)
+++ activemq/trunk/activemq-core/src/test/java/org/apache/activemq/ProducerFlowControlTest.java Tue Jan 13 04:10:22 2009
@@ -42,8 +42,10 @@
 
     ActiveMQQueue queueA = new ActiveMQQueue("QUEUE.A");
     ActiveMQQueue queueB = new ActiveMQQueue("QUEUE.B");
-    private TransportConnector connector;
-    private ActiveMQConnection connection;
+    protected TransportConnector connector;
+    protected ActiveMQConnection connection;
+    // used to test sendFailIfNoSpace on SystemUsage 
+    protected final AtomicBoolean gotResourceException = new AtomicBoolean(false);
 
     public void test2ndPubisherWithProducerWindowSendConnectionThatIsBlocked() throws Exception {
         ActiveMQConnectionFactory factory = (ActiveMQConnectionFactory)createConnectionFactory();
@@ -89,6 +91,8 @@
         
         final AtomicBoolean done = new AtomicBoolean(true);
         final AtomicBoolean keepGoing = new AtomicBoolean(true);
+        
+   
 		Thread thread = new Thread("Filler") {
 			@Override
 			public void run() {
@@ -102,14 +106,7 @@
 			}
 		};
 		thread.start();
-        while (true) {
-            Thread.sleep(1000);
-            // the producer is blocked once the done flag stays true.
-            if (done.get()) {
-                break;
-            }
-            done.set(true);
-        }
+        waitForBlockedOrResourceLimit(done);
 
         // after receiveing messges, producer should continue sending messages 
         // (done == false)
@@ -124,6 +121,7 @@
     	
 		assertFalse(done.get());
     }
+    
     public void test2ndPubisherWithSyncSendConnectionThatIsBlocked() throws Exception {
         ActiveMQConnectionFactory factory = (ActiveMQConnectionFactory)createConnectionFactory();
         factory.setAlwaysSyncSend(true);
@@ -224,15 +222,20 @@
             }
         }.start();
 
+        waitForBlockedOrResourceLimit(done);
+        keepGoing.set(false);
+    }
+
+    protected void waitForBlockedOrResourceLimit(final AtomicBoolean done)
+            throws InterruptedException {
         while (true) {
             Thread.sleep(1000);
-            // the producer is blocked once the done flag stays true.
-            if (done.get()) {
+            // the producer is blocked once the done flag stays true or there is a resource exception
+            if (done.get() || gotResourceException.get()) {
                 break;
             }
             done.set(true);
         }
-        keepGoing.set(false);
     }
 
     private CountDownLatch asyncSendTo(final ActiveMQQueue queue, final String message) throws JMSException {
@@ -274,10 +277,12 @@
     }
 
     protected void tearDown() throws Exception {
-        TcpTransport t = (TcpTransport)connection.getTransport().narrow(TcpTransport.class);
-        t.getTransportListener().onException(new IOException("Disposed."));
-        connection.getTransport().stop();
-        super.tearDown();
+        if (connection != null) {
+            TcpTransport t = (TcpTransport)connection.getTransport().narrow(TcpTransport.class);
+            t.getTransportListener().onException(new IOException("Disposed."));
+            connection.getTransport().stop();
+            super.tearDown();
+        }
     }
 
     protected ConnectionFactory createConnectionFactory() throws Exception {