[jboss-cvs] JBossAS SVN: r58406 - in branches/Branch_4_2: messaging/src/main/org/jboss/mq testsuite/src/main/org/jboss/test/jbossmq testsuite/src/main/org/jboss/test/jbossmq/support testsuite/src/main/org/jboss/test/jbossmq/test testsuite/src/resources/org/jboss/test/jbossmq/test
jboss-cvs-commits at lists.jboss.org
jboss-cvs-commits at lists.jboss.org
Wed Nov 15 11:53:11 EST 2006
Author: adrian at jboss.org
Date: 2006-11-15 11:53:02 -0500 (Wed, 15 Nov 2006)
New Revision: 58406
Added:
branches/Branch_4_2/testsuite/src/main/org/jboss/test/jbossmq/support/MockServerFailureInterceptor.java
branches/Branch_4_2/testsuite/src/main/org/jboss/test/jbossmq/test/ReceiveAfterErrorUnitTestCase.java
branches/Branch_4_2/testsuite/src/resources/org/jboss/test/jbossmq/test/ReceiveAfterErrorUnitTestCase.xml
Modified:
branches/Branch_4_2/messaging/src/main/org/jboss/mq/SpyMessageConsumer.java
branches/Branch_4_2/testsuite/src/main/org/jboss/test/jbossmq/JBossMQMicrocontainerTest.java
Log:
[JBAS-3821] - Error from server during receive corrupts internal receiving state.
Modified: branches/Branch_4_2/messaging/src/main/org/jboss/mq/SpyMessageConsumer.java
===================================================================
--- branches/Branch_4_2/messaging/src/main/org/jboss/mq/SpyMessageConsumer.java 2006-11-15 16:52:59 UTC (rev 58405)
+++ branches/Branch_4_2/messaging/src/main/org/jboss/mq/SpyMessageConsumer.java 2006-11-15 16:53:02 UTC (rev 58406)
@@ -268,84 +268,80 @@
log.trace("receive() " + this);
}
- synchronized (messages)
+ try
{
- //see if we have any undelivered messages before we go to the JMS
- //server to look.
- Message message = getMessage();
- if (message != null)
+ synchronized (messages)
{
- synchronized (stateLock)
+ //see if we have any undelivered messages before we go to the JMS
+ //server to look.
+ Message message = getMessage();
+ if (message != null)
{
- receiving = false;
-
if (trace)
log.trace("receive() message in list " + message.getJMSMessageID() + " " + this);
+ return message;
}
- return message;
- }
-
- // Loop through expired messages
- while (true)
- {
- SpyMessage msg = session.connection.receive(subscription, 0);
- if (msg != null)
+
+ // Loop through expired messages
+ while (true)
{
- Message mes = preProcessMessage(msg);
- if (mes != null)
+ SpyMessage msg = session.connection.receive(subscription, 0);
+ if (msg != null)
{
- synchronized (stateLock)
+ Message mes = preProcessMessage(msg);
+ if (mes != null)
{
- receiving = false;
-
if (trace)
log.trace("receive() message from server " + mes.getJMSMessageID() + " " + this);
+ return mes;
}
- return mes;
}
+ else
+ break;
}
- else
- break;
- }
- if (trace)
- log.trace("No message in receive(), waiting " + this);
-
- try
- {
- waitingForMessage = true;
- while (true)
+ if (trace)
+ log.trace("No message in receive(), waiting " + this);
+
+ try
{
- if (isClosed())
+ waitingForMessage = true;
+ while (true)
{
- if (trace)
- log.trace("Consumer closed in receive() " + this);
- return null;
+ if (isClosed())
+ {
+ if (trace)
+ log.trace("Consumer closed in receive() " + this);
+ return null;
+ }
+ Message mes = getMessage();
+ if (mes != null)
+ {
+ if (trace)
+ log.trace("receive() message from list after wait " + this);
+ return mes;
+ }
+ messages.wait();
}
- Message mes = getMessage();
- if (mes != null)
- {
- if (trace)
- log.trace("receive() message from list after wait " + this);
- return mes;
- }
- messages.wait();
}
- }
- catch (Throwable t)
- {
- SpyJMSException.rethrowAsJMSException("Receive interupted", t);
- throw new UnreachableStatementException();
- }
- finally
- {
- waitingForMessage = false;
- synchronized (stateLock)
+ catch (Throwable t)
{
- receiving = false;
+ SpyJMSException.rethrowAsJMSException("Receive interupted", t);
+ throw new UnreachableStatementException();
}
+ finally
+ {
+ waitingForMessage = false;
+ }
}
}
+ finally
+ {
+ synchronized (stateLock)
+ {
+ receiving = false;
+ }
+ }
}
public Message receive(long timeOut) throws JMSException
@@ -375,94 +371,89 @@
if (trace)
log.trace("receive(long) endTime=" + endTime + " " + this);
- synchronized (messages)
+ try
{
- //see if we have any undelivered messages before we go to the JMS
- //server to look.
- Message message = getMessage();
- if (message != null)
+ synchronized (messages)
{
- synchronized (stateLock)
+ //see if we have any undelivered messages before we go to the JMS
+ //server to look.
+ Message message = getMessage();
+ if (message != null)
{
- receiving = false;
-
if (trace)
log.trace("receive(long) message in list " + message.getJMSMessageID() + " " + this);
+ return message;
}
- return message;
- }
- // Loop through expired messages
- while (true)
- {
- SpyMessage msg = session.connection.receive(subscription, timeOut);
- if (msg != null)
+ // Loop through expired messages
+ while (true)
{
- Message mes = preProcessMessage(msg);
- if (mes != null)
+ SpyMessage msg = session.connection.receive(subscription, timeOut);
+ if (msg != null)
{
- synchronized (stateLock)
+ Message mes = preProcessMessage(msg);
+ if (mes != null)
{
- receiving = false;
-
if (trace)
log.trace("receive(long) message from server " + mes.getJMSMessageID() + " " + this);
+ return mes;
}
- return mes;
}
+ else
+ break;
}
- else
- break;
- }
- if (trace)
- log.trace("No message in receive(), waiting " + this);
-
- try
- {
- waitingForMessage = true;
- while (true)
+ if (trace)
+ log.trace("No message in receive(), waiting " + this);
+
+ try
{
- if (isClosed())
+ waitingForMessage = true;
+ while (true)
{
- if (trace)
- log.trace("Consumer closed in receive(long) " + this);
- return null;
- }
+ if (isClosed())
+ {
+ if (trace)
+ log.trace("Consumer closed in receive(long) " + this);
+ return null;
+ }
- Message mes = getMessage();
- if (mes != null)
- {
- if (trace)
- log.trace("receive(long) message from list after wait " + this);
- return mes;
- }
+ Message mes = getMessage();
+ if (mes != null)
+ {
+ if (trace)
+ log.trace("receive(long) message from list after wait " + this);
+ return mes;
+ }
- long att = endTime - System.currentTimeMillis();
- if (att <= 0)
- {
- if (trace)
- log.trace("receive(long) timed out endTime=" + endTime + " " + this);
- return null;
- }
+ long att = endTime - System.currentTimeMillis();
+ if (att <= 0)
+ {
+ if (trace)
+ log.trace("receive(long) timed out endTime=" + endTime + " " + this);
+ return null;
+ }
- messages.wait(att);
+ messages.wait(att);
+ }
}
- }
- catch (Throwable t)
- {
- SpyJMSException.rethrowAsJMSException("Receive interupted", t);
- throw new UnreachableStatementException();
- }
- finally
- {
- waitingForMessage = false;
- synchronized (stateLock)
+ catch (Throwable t)
{
- receiving = false;
+ SpyJMSException.rethrowAsJMSException("Receive interupted", t);
+ throw new UnreachableStatementException();
}
+ finally
+ {
+ waitingForMessage = false;
+ }
}
}
-
+ finally
+ {
+ synchronized (stateLock)
+ {
+ receiving = false;
+ }
+ }
}
public Message receiveNoWait() throws JMSException
@@ -480,54 +471,49 @@
log.trace("receiveNoWait() " + this);
}
- //see if we have any undelivered messages before we go to the JMS
- //server to look.
- synchronized (messages)
+ try
{
- Message mes = getMessage();
- if (mes != null)
+ //see if we have any undelivered messages before we go to the JMS
+ //server to look.
+ synchronized (messages)
{
- synchronized (stateLock)
+ Message mes = getMessage();
+ if (mes != null)
{
- receiving = false;
-
if (trace)
log.trace("receiveNoWait() message in list " + mes.getJMSMessageID() + " " + this);
+ return mes;
}
- return mes;
}
- }
- // Loop through expired messages
- while (true)
- {
- SpyMessage msg = session.connection.receive(subscription, -1);
- if (msg != null)
+ // Loop through expired messages
+ while (true)
{
- Message mes = preProcessMessage(msg);
- if (mes != null)
+ SpyMessage msg = session.connection.receive(subscription, -1);
+ if (msg != null)
{
- synchronized (stateLock)
+ Message mes = preProcessMessage(msg);
+ if (mes != null)
{
- receiving = false;
-
if (trace)
log.trace("receiveNoWait() message from server " + mes.getJMSMessageID() + " " + this);
+ return mes;
}
- return mes;
}
- }
- else
- {
- synchronized (stateLock)
+ else
{
- receiving = false;
+ if (trace)
+ log.trace("receiveNoWait() no message " + this);
+ return null;
}
-
- if (trace)
- log.trace("receiveNoWait() no message " + this);
- return null;
}
}
+ finally
+ {
+ synchronized (stateLock)
+ {
+ receiving = false;
+ }
+ }
}
public void close() throws JMSException
Modified: branches/Branch_4_2/testsuite/src/main/org/jboss/test/jbossmq/JBossMQMicrocontainerTest.java
===================================================================
--- branches/Branch_4_2/testsuite/src/main/org/jboss/test/jbossmq/JBossMQMicrocontainerTest.java 2006-11-15 16:52:59 UTC (rev 58405)
+++ branches/Branch_4_2/testsuite/src/main/org/jboss/test/jbossmq/JBossMQMicrocontainerTest.java 2006-11-15 16:53:02 UTC (rev 58406)
@@ -21,15 +21,17 @@
*/
package org.jboss.test.jbossmq;
-import javax.jms.Connection;
-import javax.jms.ConnectionFactory;
+import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
+
+import org.jboss.mq.SpyDestination;
+import org.jboss.mq.SpyQueue;
+import org.jboss.mq.server.JMSDestinationManager;
+import org.jboss.mq.server.JMSQueue;
+import org.jboss.test.AbstractTestDelegate;
+import org.jboss.test.jbossmq.support.MockServerFailureInterceptor;
+import org.jboss.test.kernel.junit.MicrocontainerTest;
-import org.jboss.mq.SpyQueue;
-import org.jboss.mq.server.JMSDestinationManager;
-import org.jboss.mq.server.JMSQueue;
-import org.jboss.test.AbstractTestDelegate;
-import org.jboss.test.kernel.junit.MicrocontainerTest;
-
/**
* JBossMQMicrocontainerTest.
*
@@ -83,4 +85,20 @@
server.addDestination(realQueue);
return queue;
}
+
+ protected void removeDestination(SpyDestination destination) throws Exception
+ {
+ JMSDestinationManager server = getJMSServer();
+ server.closeDestination(destination);
+ }
+
+ protected MockServerFailureInterceptor getMockServerFailure() throws Exception
+ {
+ return (MockServerFailureInterceptor) getBean("MockServerFailure");
+ }
+
+ protected void raiseReceiveError(boolean value) throws Exception
+ {
+ getMockServerFailure().raiseReceiveError = value;
+ }
}
Added: branches/Branch_4_2/testsuite/src/main/org/jboss/test/jbossmq/support/MockServerFailureInterceptor.java
===================================================================
--- branches/Branch_4_2/testsuite/src/main/org/jboss/test/jbossmq/support/MockServerFailureInterceptor.java 2006-11-15 16:52:59 UTC (rev 58405)
+++ branches/Branch_4_2/testsuite/src/main/org/jboss/test/jbossmq/support/MockServerFailureInterceptor.java 2006-11-15 16:53:02 UTC (rev 58406)
@@ -0,0 +1,46 @@
+/*
+* JBoss, Home of Professional Open Source
+* Copyright 2006, JBoss Inc., and individual contributors as indicated
+* by the @authors tag. See the copyright.txt in the distribution for a
+* full listing of individual contributors.
+*
+* This is free software; you can redistribute it and/or modify it
+* under the terms of the GNU Lesser General Public License as
+* published by the Free Software Foundation; either version 2.1 of
+* the License, or (at your option) any later version.
+*
+* This software is distributed in the hope that it will be useful,
+* but WITHOUT ANY WARRANTY; without even the implied warranty of
+* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+* Lesser General Public License for more details.
+*
+* You should have received a copy of the GNU Lesser General Public
+* License along with this software; if not, write to the Free
+* Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
+* 02110-1301 USA, or see the FSF site: http://www.fsf.org.
+*/
+package org.jboss.test.jbossmq.support;
+
+import javax.jms.JMSException;
+
+import org.jboss.mq.ConnectionToken;
+import org.jboss.mq.SpyMessage;
+import org.jboss.mq.server.JMSServerInterceptorSupport;
+
+/**
+ * MockServerFailureInterceptor.
+ *
+ * @author <a href="adrian at jboss.com">Adrian Brock</a>
+ * @version $Revision: 1.1 $
+ */
+public class MockServerFailureInterceptor extends JMSServerInterceptorSupport
+{
+ public boolean raiseReceiveError = false;
+
+ public SpyMessage receive(ConnectionToken dc, int subscriberId, long wait) throws JMSException
+ {
+ if (raiseReceiveError)
+ throw new JMSException("Error in receive as required.");
+ return super.receive(dc, subscriberId, wait);
+ }
+}
Added: branches/Branch_4_2/testsuite/src/main/org/jboss/test/jbossmq/test/ReceiveAfterErrorUnitTestCase.java
===================================================================
--- branches/Branch_4_2/testsuite/src/main/org/jboss/test/jbossmq/test/ReceiveAfterErrorUnitTestCase.java 2006-11-15 16:52:59 UTC (rev 58405)
+++ branches/Branch_4_2/testsuite/src/main/org/jboss/test/jbossmq/test/ReceiveAfterErrorUnitTestCase.java 2006-11-15 16:53:02 UTC (rev 58406)
@@ -0,0 +1,136 @@
+/*
+ * JBoss, Home of Professional Open Source.
+ * Copyright 2006, Red Hat Middleware LLC, and individual contributors
+ * as indicated by the @author tags. See the copyright.txt file in the
+ * distribution for a full listing of individual contributors.
+ *
+ * This is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU Lesser General Public License as
+ * published by the Free Software Foundation; either version 2.1 of
+ * the License, or (at your option) any later version.
+ *
+ * This software is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this software; if not, write to the Free
+ * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
+ * 02110-1301 USA, or see the FSF site: http://www.fsf.org.
+ */
+package org.jboss.test.jbossmq.test;
+
+import javax.jms.Connection;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+
+import junit.framework.Test;
+
+import org.jboss.mq.SpyDestination;
+import org.jboss.test.jbossmq.JBossMQMicrocontainerTest;
+
+/**
+ * A test to make sure an error during receive doesn't lead to corrupt "receiving" state
+ *
+ * @author <a href="mailto:adrian at jboss.org>Adrian Brock</a>
+ * @version <tt>$Revision: 57211 $</tt>
+ */
+public class ReceiveAfterErrorUnitTestCase extends JBossMQMicrocontainerTest
+{
+ public ReceiveAfterErrorUnitTestCase(String name) throws Exception
+ {
+ super(name);
+ }
+
+ public static Test suite()
+ {
+ return suite(ReceiveAfterErrorUnitTestCase.class);
+ }
+
+ public interface ReceiveOperation
+ {
+ Message receive(MessageConsumer consumer) throws JMSException;
+ }
+
+ public void testReceiveAfterError() throws Exception
+ {
+ receiveAfterError(new ReceiveOperation()
+ {
+ public Message receive(MessageConsumer consumer) throws JMSException
+ {
+ return consumer.receive();
+ }
+ });
+ }
+
+ public void testReceiveWithWaitAfterError() throws Exception
+ {
+ receiveAfterError(new ReceiveOperation()
+ {
+ public Message receive(MessageConsumer consumer) throws JMSException
+ {
+ return consumer.receive(1000);
+ }
+ });
+ }
+
+ public void testReceiveNoWaitAfterError() throws Exception
+ {
+ receiveAfterError(new ReceiveOperation()
+ {
+ public Message receive(MessageConsumer consumer) throws JMSException
+ {
+ return consumer.receiveNoWait();
+ }
+ });
+ }
+
+ protected void receiveAfterError(ReceiveOperation operation) throws Exception
+ {
+ SpyDestination destination = createQueue("testQueue");
+ try
+ {
+ Connection connection = createConnection();
+ try
+ {
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ connection.start();
+
+ MessageProducer producer = session.createProducer(destination);
+ Message message = session.createMessage();
+ producer.send(message);
+
+ MessageConsumer consumer = session.createConsumer(destination);
+
+ // Receive should now throw an error
+ raiseReceiveError(true);
+ try
+ {
+ operation.receive(consumer);
+ fail("Should not be here!");
+ }
+ catch (Throwable t)
+ {
+ checkThrowable(JMSException.class, t);
+ }
+ raiseReceiveError(false);
+
+ message = operation.receive(consumer);
+ assertNotNull(message);
+ }
+ finally
+ {
+ connection.close();
+ }
+ }
+ finally
+ {
+ removeDestination(destination);
+ }
+ }
+}
+
Added: branches/Branch_4_2/testsuite/src/resources/org/jboss/test/jbossmq/test/ReceiveAfterErrorUnitTestCase.xml
===================================================================
--- branches/Branch_4_2/testsuite/src/resources/org/jboss/test/jbossmq/test/ReceiveAfterErrorUnitTestCase.xml 2006-11-15 16:52:59 UTC (rev 58405)
+++ branches/Branch_4_2/testsuite/src/resources/org/jboss/test/jbossmq/test/ReceiveAfterErrorUnitTestCase.xml 2006-11-15 16:53:02 UTC (rev 58406)
@@ -0,0 +1,60 @@
+<?xml version="1.0" encoding="UTF-8"?>
+
+<deployment xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="urn:jboss:bean-deployer bean-deployer_1_0.xsd"
+ xmlns="urn:jboss:bean-deployer">
+
+ <bean name="PersistenceManager" class="org.jboss.mq.pm.none.PersistenceManager"/>
+
+ <bean name="StateManager" class="org.jboss.mq.sm.none.NullStateManager"/>
+
+ <bean name="MessageCache" class="org.jboss.mq.kernel.MessageCache">
+ <property name="persistenceManager"><inject bean="PersistenceManager"/></property>
+ </bean>
+
+ <bean name="DestinationManager" class="org.jboss.mq.kernel.DestinationManager">
+ <property name="persistenceManagerInstance"><inject bean="PersistenceManager"/></property>
+ <property name="stateManagerInstance"><inject bean="StateManager"/></property>
+ <property name="messageCacheInstance"><inject bean="MessageCache"/></property>
+ </bean>
+
+ <bean name="JMSServer" class="org.jboss.mq.server.JMSDestinationManager">
+ <constructor factoryMethod="getInterceptor">
+ <factory bean="DestinationManager"/>
+ </constructor>
+ </bean>
+
+ <bean name="MockServerFailure" class="org.jboss.test.jbossmq.support.MockServerFailureInterceptor">
+ <property name="next"><inject bean="JMSServer"/></property>
+ </bean>
+
+ <bean name="Invoker" class="org.jboss.mq.server.JMSServerInvoker">
+ <property name="next"><inject bean="MockServerFailure"/></property>
+ </bean>
+
+ <bean name="ServerIL" class="org.jboss.mq.il.jvm.JVMServerIL">
+ <constructor>
+ <parameter><inject bean="Invoker"/></parameter>
+ </constructor>
+ </bean>
+
+ <bean name="GCF" class="org.jboss.mq.GenericConnectionFactory">
+ <constructor>
+ <parameter><inject bean="ServerIL"/></parameter>
+ <parameter>
+ <map class="java.util.Properties" keyClass="java.lang.String" valueClass="java.lang.String">
+ <entry>
+ <key>ClientILService</key><value>org.jboss.mq.il.jvm.JVMClientILService</value>
+ </entry>
+ </map>
+ </parameter>
+ </constructor>
+ </bean>
+
+ <bean name="ConnectionFactory" class="org.jboss.mq.SpyConnectionFactory">
+ <constructor>
+ <parameter class="org.jboss.mq.GenericConnectionFactory"><inject bean="GCF"/></parameter>
+ </constructor>
+ </bean>
+
+</deployment>
More information about the jboss-cvs-commits
mailing list