11 a.m.
Author: ritchiem
Date: 2008-06-13 12:00:39 -0400 (Fri, 13 Jun 2008)
New Revision: 2152
Added:
store/branches/java/M2.1.x/java/bdbstore/src/test/java/org/etp/qpid/server/MessageReSendTest.java
Modified:
store/branches/java/M2.1.x/
store/branches/java/M2.1.x/java/bdbstore/src/test/java/org/etp/qpid/server/StoreContextRaceConditionTest.java
store/branches/java/M2.1.x/java/bdbstore/src/test/java/org/etp/qpid/testutil/BDBVMTestCase.java
Log:
Merged revisions 2151 via svnmerge from
https://svn.jboss.org/repos/rhmessaging/store/branches/java/M2.x
........
r2151 | ritchiem | 2008-06-13 16:54:41 +0100 (Fri, 13 Jun 2008) | 3 lines
QPID-1136 : New BDBTest to ensure that a the store doesn't accidentally restore
messages that have been received and acked successfully.
Changes required to BDBVMTestCase that are not complete. Changing to the persistent
environment-path will only work if the configuration is changed not the environment
variable. i.e When QPID_WORK is set and the environment loaded
........
Property changes on: store/branches/java/M2.1.x
___________________________________________________________________
Name: svnmerge-integrated
- /store/branches/java/M2.x:1-2103
+ /store/branches/java/M2.x:1-2103,2151
Copied:
store/branches/java/M2.1.x/java/bdbstore/src/test/java/org/etp/qpid/server/MessageReSendTest.java
(from rev 2151,
store/branches/java/M2.x/java/bdbstore/src/test/java/org/etp/qpid/server/MessageReSendTest.java)
===================================================================
---
store/branches/java/M2.1.x/java/bdbstore/src/test/java/org/etp/qpid/server/MessageReSendTest.java
(rev 0)
+++
store/branches/java/M2.1.x/java/bdbstore/src/test/java/org/etp/qpid/server/MessageReSendTest.java 2008-06-13
16:00:39 UTC (rev 2152)
@@ -0,0 +1,234 @@
+/*
+ *
+ * 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.etp.qpid.server;
+
+import org.apache.commons.configuration.Configuration;
+import org.apache.log4j.Level;
+import org.apache.log4j.Logger;
+import org.apache.qpid.server.registry.ApplicationRegistry;
+import org.apache.qpid.server.registry.ConfigurationFileApplicationRegistry;
+import org.apache.qpid.server.virtualhost.VirtualHostRegistry;
+import org.etp.qpid.testutil.BDBVMTestCase;
+
+import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Queue;
+import javax.jms.Session;
+import javax.jms.TextMessage;
+import javax.naming.NamingException;
+import java.util.Map;
+
+public class MessageReSendTest extends BDBVMTestCase
+{
+ private static final Logger _logger = Logger.getLogger(MessageReSendTest.class);
+ private long _TimeToLive = 0L;
+ private static long SECOND = 1000L;
+
+ private static final String LOGGING_KEY = "amqj.logging.level";
+
+ private String _loggingOriginal;
+ ConfigurationFileApplicationRegistry _config;
+
+ public void setUp() throws Exception
+ {
+ //Disable the logging
+ _loggingOriginal = System.getProperty(LOGGING_KEY);
+ System.setProperty(LOGGING_KEY, Level.WARN.toString());
+
+ //Set the Work Directory
+ setupWorkDirectory();
+
+ Configuration configuration =
ConfigurationFileApplicationRegistry.config(_persistentConfigFile);
+
+ //Disable management
+ configuration.setProperty("management.enabled", "false");
+
+ _config = new ConfigurationFileApplicationRegistry(configuration);
+
+ ApplicationRegistry.initialise(_config, 1);
+
+ //Remove the Vhosts we are not using to free up CPU from extra housekeeping
threads.
+ VirtualHostRegistry vHostRegistry =
ApplicationRegistry.getInstance().getVirtualHostRegistry();
+
+ _connections.put("connection2",
"amqp://guest:guest@client2/test?brokerlist='vm://:2'");
+
+ //Create the Broker
+ super.setUp();
+
+ _queue = (Queue) _context.lookup("queue");
+ }
+
+ public void tearDown() throws Exception
+ {
+
+ if (_loggingOriginal != null)
+ {
+ System.setProperty(LOGGING_KEY, _loggingOriginal);
+ }
+ //set back to null
+
+ super.tearDown();
+ }
+
+ protected static final String MESSAGE_ID_PROPERTY = "MessageIDProperty";
+
+ protected Queue _queue;
+
+ protected void sendMessages(int num) throws JMSException
+ {
+ Connection producerConnection = null;
+ try
+ {
+ producerConnection = ((ConnectionFactory)
_context.lookup("connection")).createConnection();
+ }
+ catch (NamingException e)
+ {
+ fail("Unable to lookup connection in JNDI.");
+ }
+
+ sendMessages(producerConnection, num);
+ }
+
+ protected void sendMessages(Connection producerConnection, int num) throws
JMSException
+ {
+ Session producerSession = producerConnection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
+
+ //Ensure _queue is created
+ producerSession.createConsumer(_queue).close();
+
+ MessageProducer producer = producerSession.createProducer(_queue);
+
+ producer.setTimeToLive(_TimeToLive);
+ producer.setDisableMessageTimestamp(false);
+
+ for (int messsageID = 0; messsageID < num; messsageID++)
+ {
+ TextMessage textMsg = producerSession.createTextMessage("Message "
+ messsageID);
+ textMsg.setIntProperty(MESSAGE_ID_PROPERTY, messsageID);
+ producer.send(textMsg);
+ }
+
+ producerConnection.close();
+ }
+
+ public void test() throws InterruptedException, NamingException, JMSException
+ {
+
+ //Send Message
+ sendMessages(createConnection(), 1);
+ System.err.println("SEND");
+
+ //Create Connection
+ Connection connection = createConnection();
+ System.err.println("RECEIVE");
+
+ //Receive Message
+ checkMessagesOnQueue(connection, _queue, 1);
+ //Close connections
+ connection.close();
+ System.err.println("VALIDATE");
+
+ //Reconnect and ensure message is gone
+ connection = createConnection();
+ checkMessagesOnQueue(connection, _queue, 0);
+ connection.close();
+
+ try
+ {
+ //restart broker
+ stopVMBroker(1);
+ System.err.println("START");
+ startVMBroker(1, _config);
+ }
+ catch (Exception e)
+ {
+ fail(e.getMessage());
+ }
+
+ //reconnect and ensure message is gone
+ connection = createConnection();
+ checkMessagesOnQueue(connection, _queue, 0);
+ connection.close();
+ }
+
+ private void checkMessagesOnQueue(Connection connection, Queue queue, int count)
+ {
+ try
+ {
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ MessageConsumer consumer = session.createConsumer(queue);
+
+ connection.start();
+
+ Message msg = consumer.receive(1000);
+
+ if (count > 0)
+ {
+ int received = 1;
+ while (received < count)
+ {
+ assertNotNull(msg);
+ assertEquals(received, msg.getIntProperty(MESSAGE_ID_PROPERTY));
+
+ //get next message
+ msg = consumer.receive(1000);
+ }
+
+ }
+ else
+ {
+ assertNull("Received Message when none expected", msg);
+ }
+ }
+ catch (JMSException e)
+ {
+ fail(e.getMessage());
+ }
+ }
+
+ private Connection createConnection()
+ {
+
+ try
+ {
+ try
+ {
+ return ((ConnectionFactory)
_context.lookup("connection")).createConnection();
+ }
+ catch (NamingException e)
+ {
+ fail("Unable to lookup connection in JNDI.");
+ }
+ }
+ catch (JMSException e)
+ {
+ fail(e.getMessage());
+ }
+ return null;
+
+ }
+
+}
Modified:
store/branches/java/M2.1.x/java/bdbstore/src/test/java/org/etp/qpid/server/StoreContextRaceConditionTest.java
===================================================================
---
store/branches/java/M2.1.x/java/bdbstore/src/test/java/org/etp/qpid/server/StoreContextRaceConditionTest.java 2008-06-13
15:54:41 UTC (rev 2151)
+++
store/branches/java/M2.1.x/java/bdbstore/src/test/java/org/etp/qpid/server/StoreContextRaceConditionTest.java 2008-06-13
16:00:39 UTC (rev 2152)
@@ -38,7 +38,6 @@
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.naming.NamingException;
-import java.io.File;
public class StoreContextRaceConditionTest extends BDBVMTestCase
{
Modified:
store/branches/java/M2.1.x/java/bdbstore/src/test/java/org/etp/qpid/testutil/BDBVMTestCase.java
===================================================================
---
store/branches/java/M2.1.x/java/bdbstore/src/test/java/org/etp/qpid/testutil/BDBVMTestCase.java 2008-06-13
15:54:41 UTC (rev 2151)
+++
store/branches/java/M2.1.x/java/bdbstore/src/test/java/org/etp/qpid/testutil/BDBVMTestCase.java 2008-06-13
16:00:39 UTC (rev 2152)
@@ -80,7 +80,14 @@
public void startVMBroker(int vmID, File configFile)
{
- testWork = BDB_WORK_PRE_TEST + "-" + vmID;
+ if (vmID != 1)
+ {
+ testWork = BDB_WORK_PRE_TEST + "-" + vmID;
+ }
+ else
+ {
+ testWork = BDB_WORK_PRE_TEST;
+ }
System.setProperty(BDB_WORK, testWork);
System.setProperty(QPID_WORK, testWork);