You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by rg...@apache.org on 2006/12/21 22:08:41 UTC
svn commit: r489461 [3/5] - in /incubator/qpid/branches/new_persistence/java:
broker/src/main/java/org/apache/qpid/server/exchange/
broker/src/main/java/org/apache/qpid/server/handler/
broker/src/main/java/org/apache/qpid/server/protocol/ broker/src/ma...
Added: incubator/qpid/branches/new_persistence/java/client/src/old_test/java/org/apache/qpid/weblogic/ServiceRequestingClient.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/new_persistence/java/client/src/old_test/java/org/apache/qpid/weblogic/ServiceRequestingClient.java?view=auto&rev=489461
==============================================================================
--- incubator/qpid/branches/new_persistence/java/client/src/old_test/java/org/apache/qpid/weblogic/ServiceRequestingClient.java (added)
+++ incubator/qpid/branches/new_persistence/java/client/src/old_test/java/org/apache/qpid/weblogic/ServiceRequestingClient.java Thu Dec 21 13:08:38 2006
@@ -0,0 +1,192 @@
+/*
+ *
+ * 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.qpid.weblogic;
+
+import org.apache.qpid.jms.*;
+import org.apache.log4j.Logger;
+
+import javax.naming.NamingException;
+import javax.naming.InitialContext;
+import javax.naming.Context;
+import javax.jms.*;
+import javax.jms.MessageConsumer;
+import javax.jms.Session;
+import java.util.Hashtable;
+import java.io.File;
+import java.io.FilenameFilter;
+import java.io.Reader;
+import java.io.FileReader;
+
+/**
+ * Created by IntelliJ IDEA.
+ * User: U806869
+ * Date: 28-May-2005
+ * Time: 21:54:51
+ * To change this template use File | Settings | File Templates.
+ */
+public class ServiceRequestingClient
+{
+ private static final Logger _log = Logger.getLogger(ServiceRequestingClient.class);
+ private static final String JNDI_FACTORY = "weblogic.jndi.WLInitialContextFactory";
+ private static final String JMS_FACTORY = "transientJMSConnectionFactory";
+
+ private static class CallbackHandler implements MessageListener
+ {
+ private int _expectedMessageCount;
+
+ private int _actualMessageCount;
+
+ private long _startTime;
+
+ private long _averageLatency;
+
+ public CallbackHandler(int expectedMessageCount, long startTime)
+ {
+ _expectedMessageCount = expectedMessageCount;
+ _startTime = startTime;
+ }
+
+ public void onMessage(Message m)
+ {
+ if (_log.isDebugEnabled())
+ {
+ _log.debug("Message received: " + m);
+ }
+ try
+ {
+ if (m.propertyExists("timeSent"))
+ {
+ long timeSent = m.getLongProperty("timeSent");
+ long now = System.currentTimeMillis();
+ if (_averageLatency == 0)
+ {
+ _averageLatency = now - timeSent;
+ _log.info("Latency " + _averageLatency);
+ }
+ else
+ {
+ _log.info("Individual latency: " + (now-timeSent));
+ _averageLatency = (_averageLatency + (now - timeSent))/2;
+ _log.info("Average latency now: " + _averageLatency);
+ }
+ }
+ }
+ catch (JMSException e)
+ {
+ _log.error("Could not calculate latency");
+ }
+
+ _actualMessageCount++;
+ if (_actualMessageCount%1000 == 0)
+ {
+ try
+ {
+ m.acknowledge();
+ }
+ catch (JMSException e)
+ {
+ _log.error("Error acknowledging message");
+ }
+ _log.info("Received message count: " + _actualMessageCount);
+ }
+ /*if (!"henson".equals(m.toString()))
+ {
+ _log.error("Message response not correct: expected 'henson' but got " + m.toString());
+ }
+ else
+ {
+ if (_log.isDebugEnabled())
+ {
+ _log.debug("Message " + m + " received");
+ }
+ else
+ {
+ _log.info("Message received");
+ }
+ } */
+
+ if (_actualMessageCount == _expectedMessageCount)
+ {
+ long timeTaken = System.currentTimeMillis() - _startTime;
+ System.out.println("Total time taken to receive " + _expectedMessageCount+ " messages was " +
+ timeTaken + "ms, equivalent to " +
+ (_expectedMessageCount/(timeTaken/1000.0)) + " messages per second");
+ System.out.println("Average latency is: " + _averageLatency);
+ }
+ }
+ }
+
+ public static void main(String[] args) throws Exception
+ {
+ if (args.length != 3)
+ {
+ System.out.println("Usage: IXPublisher <WLS URL> <sendQueue> <count> will publish count messages to ");
+ System.out.println("queue sendQueue and waits for a response on a temp queue");
+ System.exit(1);
+ }
+
+ String url = args[0];
+ String sendQueue = args[1];
+ int messageCount = Integer.parseInt(args[2]);
+
+ InitialContext ctx = getInitialContext(url);
+
+ QueueConnectionFactory qconFactory = (QueueConnectionFactory) ctx.lookup(JMS_FACTORY);
+ QueueConnection qcon = qconFactory.createQueueConnection();
+ QueueSession qsession = qcon.createQueueSession(false, Session.CLIENT_ACKNOWLEDGE);
+ Queue sendQ = (Queue) ctx.lookup(sendQueue);
+ Queue receiveQ = qsession.createTemporaryQueue();
+ QueueSender qsender = qsession.createSender(sendQ);
+ qsender.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
+ _log.debug("Queue sender created for service queue " + sendQ);
+
+ javax.jms.MessageConsumer messageConsumer = (javax.jms.MessageConsumer) qsession.createConsumer(receiveQ);
+
+ //TextMessage msg = _session.createTextMessage(tempDestination.getQueueName() + "/Presented to in conjunction with Mahnah Mahnah and the Snowths");
+ final long startTime = System.currentTimeMillis();
+
+ messageConsumer.setMessageListener(new CallbackHandler(messageCount, startTime));
+ qcon.start();
+ for (int i = 0; i < messageCount; i++)
+ {
+ TextMessage msg = qsession.createTextMessage("/Presented to in conjunction with Mahnah Mahnah and the Snowths:" + i);
+ msg.setJMSReplyTo(receiveQ);
+ if (i%1000 == 0)
+ {
+ long timeNow = System.currentTimeMillis();
+ msg.setLongProperty("timeSent", timeNow);
+ }
+ qsender.send(msg);
+ }
+
+ new Thread("foo").start();
+ //qsession.close();
+ //qcon.close();
+ }
+
+ private static InitialContext getInitialContext(String url) throws NamingException
+ {
+ Hashtable env = new Hashtable();
+ env.put(Context.INITIAL_CONTEXT_FACTORY, JNDI_FACTORY);
+ env.put(Context.PROVIDER_URL, url);
+ return new InitialContext(env);
+ }
+}
Propchange: incubator/qpid/branches/new_persistence/java/client/src/old_test/java/org/apache/qpid/weblogic/ServiceRequestingClient.java
------------------------------------------------------------------------------
svn:eol-style = native
Modified: incubator/qpid/branches/new_persistence/java/client/src/test/java/org/apache/qpid/test/unit/ack/RecoverTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/new_persistence/java/client/src/test/java/org/apache/qpid/test/unit/ack/RecoverTest.java?view=diff&rev=489461&r1=489460&r2=489461
==============================================================================
--- incubator/qpid/branches/new_persistence/java/client/src/test/java/org/apache/qpid/test/unit/ack/RecoverTest.java (original)
+++ incubator/qpid/branches/new_persistence/java/client/src/test/java/org/apache/qpid/test/unit/ack/RecoverTest.java Thu Dec 21 13:08:38 2006
@@ -36,18 +36,21 @@
{
private static final Logger _logger = Logger.getLogger(RecoverTest.class);
- static
+ protected void setUp() throws Exception
{
- String workdir = System.getProperty("QPID_WORK");
- if (workdir == null || workdir.equals(""))
- {
- String tempdir = System.getProperty("java.io.tmpdir");
- System.out.println("QPID_WORK not set using tmp directory: " + tempdir);
- System.setProperty("QPID_WORK", tempdir);
- }
- DOMConfigurator.configure("../broker/etc/log4j.xml");
+ super.setUp();
+ TransportConnection.createVMBroker(1);
}
+ protected void tearDown() throws Exception
+ {
+ super.tearDown();
+ TransportConnection.killAllVMBrokers();
+ //Thread.sleep(2000);
+ }
+
+
+
public void testRecoverResendsMsgs() throws Exception
{
Connection con = new AMQConnection("vm://:1", "guest", "guest", "consumer1", "/test");
@@ -104,8 +107,74 @@
con.close();
}
+
+ public void testRecoverResendsMsgsAckOnEarlier() throws Exception
+ {
+ Connection con = new AMQConnection("vm://:1", "guest", "guest", "consumer1", "/test");
+
+ Session consumerSession = con.createSession(false, Session.CLIENT_ACKNOWLEDGE);
+ Queue queue = new AMQQueue("someQ", "someQ", false, true);
+ MessageConsumer consumer = consumerSession.createConsumer(queue);
+ //force synch to ensure the consumer has resulted in a bound queue
+ ((AMQSession) consumerSession).declareExchangeSynch("amq.direct", "direct");
+
+ Connection con2 = new AMQConnection("vm://:1", "guest", "guest", "producer1", "/test");
+ Session producerSession = con2.createSession(false, Session.CLIENT_ACKNOWLEDGE);
+ MessageProducer producer = producerSession.createProducer(queue);
+
+ _logger.info("Sending four messages");
+ producer.send(producerSession.createTextMessage("msg1"));
+ producer.send(producerSession.createTextMessage("msg2"));
+ producer.send(producerSession.createTextMessage("msg3"));
+ producer.send(producerSession.createTextMessage("msg4"));
+
+ con2.close();
+
+ _logger.info("Starting connection");
+ con.start();
+ TextMessage tm = (TextMessage) consumer.receive();
+ TextMessage tm2 = (TextMessage) consumer.receive();
+ tm.acknowledge();
+ _logger.info("Received 2 messages, acknowledge() first message, should acknowledge both");
+
+ consumer.receive();
+ consumer.receive();
+ _logger.info("Received all four messages. Calling recover with two outstanding messages");
+ // no ack for last three messages so when I call recover I expect to get three messages back
+ consumerSession.recover();
+ TextMessage tm3 = (TextMessage) consumer.receive(3000);
+ assertEquals("msg3", tm3.getText());
+
+ TextMessage tm4 = (TextMessage) consumer.receive(3000);
+ assertEquals("msg4", tm4.getText());
+
+
+ _logger.info("Received redelivery of two messages. calling acknolwedgeThis() first of those message");
+ ((org.apache.qpid.jms.Message)tm3).acknowledgeThis();
+
+ _logger.info("Calling recover");
+ // all acked so no messages to be delivered
+ consumerSession.recover();
+
+ tm4 = (TextMessage) consumer.receive(3000);
+ assertEquals("msg4", tm4.getText());
+ ((org.apache.qpid.jms.Message)tm4).acknowledgeThis();
+
+ _logger.info("Calling recover");
+ // all acked so no messages to be delivered
+ consumerSession.recover();
+
+
+ tm = (TextMessage) consumer.receiveNoWait();
+ assertNull(tm);
+ _logger.info("No messages redelivered as is expected");
+
+ con.close();
+ }
+
+
public static junit.framework.Test suite()
{
- return new VMBrokerSetup(new junit.framework.TestSuite(RecoverTest.class));
+ return new junit.framework.TestSuite(RecoverTest.class);
}
}
Modified: incubator/qpid/branches/new_persistence/java/client/src/test/java/org/apache/qpid/test/unit/basic/FieldTableMessageTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/new_persistence/java/client/src/test/java/org/apache/qpid/test/unit/basic/FieldTableMessageTest.java?view=diff&rev=489461&r1=489460&r2=489461
==============================================================================
--- incubator/qpid/branches/new_persistence/java/client/src/test/java/org/apache/qpid/test/unit/basic/FieldTableMessageTest.java (original)
+++ incubator/qpid/branches/new_persistence/java/client/src/test/java/org/apache/qpid/test/unit/basic/FieldTableMessageTest.java Thu Dec 21 13:08:38 2006
@@ -26,7 +26,6 @@
import org.apache.qpid.client.AMQSession;
import org.apache.qpid.client.message.JMSBytesMessage;
import org.apache.qpid.framing.AMQFrameDecodingException;
-import org.apache.qpid.framing.FieldTableTest;
import org.apache.qpid.framing.FieldTable;
import org.apache.qpid.framing.FieldTableFactory;
import org.apache.qpid.testutil.VMBrokerSetup;
@@ -134,7 +133,11 @@
{
ByteBuffer buffer = ((JMSBytesMessage) m).getData();
FieldTable actual = FieldTableFactory.newFieldTable(buffer, buffer.remaining());
- new FieldTableTest().assertEquivalent(_expected, actual);
+ for (Object o : _expected.keySet())
+ {
+ String key = (String) o;
+ assertEquals("Values for " + key + " did not match", _expected.get(key), actual.get(key));
+ }
}
}
Modified: incubator/qpid/branches/new_persistence/java/client/src/test/java/org/apache/qpid/test/unit/basic/MapMessageTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/new_persistence/java/client/src/test/java/org/apache/qpid/test/unit/basic/MapMessageTest.java?view=diff&rev=489461&r1=489460&r2=489461
==============================================================================
--- incubator/qpid/branches/new_persistence/java/client/src/test/java/org/apache/qpid/test/unit/basic/MapMessageTest.java (original)
+++ incubator/qpid/branches/new_persistence/java/client/src/test/java/org/apache/qpid/test/unit/basic/MapMessageTest.java Thu Dec 21 13:08:38 2006
@@ -20,20 +20,19 @@
*/
package org.apache.qpid.test.unit.basic;
+import junit.framework.Assert;
+import junit.framework.TestCase;
+import org.apache.log4j.Logger;
import org.apache.qpid.client.AMQConnection;
import org.apache.qpid.client.AMQQueue;
import org.apache.qpid.client.AMQSession;
import org.apache.qpid.client.message.JMSMapMessage;
-import org.apache.qpid.testutil.VMBrokerSetup;
-import org.apache.log4j.Logger;
+import org.apache.qpid.client.transport.TransportConnection;
+import javax.jms.*;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
-import javax.jms.*;
-
-import junit.framework.TestCase;
-import junit.framework.Assert;
public class MapMessageTest extends TestCase implements MessageListener
{
@@ -56,7 +55,7 @@
super.setUp();
try
{
- //TransportConnection.createVMBroker(1);
+ TransportConnection.createVMBroker(1);
init(new AMQConnection(_connectionString, "guest", "guest", randomize("Client"), "/test_path"));
}
catch (Exception e)
@@ -69,7 +68,7 @@
{
_logger.info("Tearing Down unit.basic.MapMessageTest");
super.tearDown();
- //TransportConnection.killAllVMBrokers();
+ TransportConnection.killAllVMBrokers();
}
private void init(AMQConnection connection) throws Exception
@@ -166,12 +165,12 @@
testMapValues(m, count);
+ testCorrectExceptions(m);
+
testMessageWriteStatus(m);
testPropertyWriteStatus(m);
- testCorrectExceptions(m);
-
count++;
}
}
@@ -230,7 +229,7 @@
m.getChar("message");
fail("Exception Expected.");
}
- catch (MessageFormatException nfe)
+ catch (MessageFormatException npe)
{
//normal execution
}
@@ -320,7 +319,7 @@
m.getChar("short");
fail("Exception Expected.");
}
- catch (MessageFormatException nfe)
+ catch (MessageFormatException npe)
{
//normal execution
}
@@ -403,7 +402,7 @@
m.getChar("long");
fail("Exception Expected.");
}
- catch (MessageFormatException nfe)
+ catch (MessageFormatException npe)
{
//normal execution
}
@@ -494,7 +493,7 @@
m.getChar("double");
fail("Exception Expected.");
}
- catch (MessageFormatException nfe)
+ catch (MessageFormatException npe)
{
//normal execution
}
@@ -587,7 +586,7 @@
m.getChar("float");
fail("Exception Expected.");
}
- catch (MessageFormatException nfe)
+ catch (MessageFormatException npe)
{
//normal execution
}
@@ -671,7 +670,7 @@
m.getChar("int");
fail("Exception Expected.");
}
- catch (MessageFormatException nfe)
+ catch (MessageFormatException npe)
{
//normal execution
}
@@ -843,7 +842,7 @@
m.getChar("bytes");
fail("Exception Expected.");
}
- catch (MessageFormatException nfe)
+ catch (MessageFormatException npe)
{
//normal execution
}
@@ -928,7 +927,7 @@
m.getChar("byte");
fail("Exception Expected.");
}
- catch (MessageFormatException nfe)
+ catch (MessageFormatException npe)
{
//normal execution
}
@@ -1005,7 +1004,7 @@
m.getChar("odd");
fail("Exception Expected.");
}
- catch (MessageFormatException nfe)
+ catch (MessageFormatException npe)
{
//normal execution
}
@@ -1249,6 +1248,6 @@
public static junit.framework.Test suite()
{
- return new VMBrokerSetup(new junit.framework.TestSuite(MapMessageTest.class));
+ return new junit.framework.TestSuite(MapMessageTest.class);
}
}
Modified: incubator/qpid/branches/new_persistence/java/client/src/test/java/org/apache/qpid/test/unit/basic/MultipleConnectionTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/new_persistence/java/client/src/test/java/org/apache/qpid/test/unit/basic/MultipleConnectionTest.java?view=diff&rev=489461&r1=489460&r2=489461
==============================================================================
--- incubator/qpid/branches/new_persistence/java/client/src/test/java/org/apache/qpid/test/unit/basic/MultipleConnectionTest.java (original)
+++ incubator/qpid/branches/new_persistence/java/client/src/test/java/org/apache/qpid/test/unit/basic/MultipleConnectionTest.java Thu Dec 21 13:08:38 2006
@@ -6,9 +6,9 @@
* 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
@@ -19,18 +19,15 @@
*/
package org.apache.qpid.test.unit.basic;
+import junit.framework.TestCase;
import org.apache.qpid.client.AMQConnection;
import org.apache.qpid.client.AMQDestination;
import org.apache.qpid.client.AMQSession;
import org.apache.qpid.client.AMQTopic;
-import org.apache.qpid.client.vmbroker.AMQVMBrokerCreationException;
import org.apache.qpid.client.transport.TransportConnection;
-import org.apache.qpid.testutil.VMBrokerSetup;
import javax.jms.*;
-import junit.framework.TestCase;
-
public class MultipleConnectionTest extends TestCase
{
public static final String _defaultBroker = "vm://:1";
@@ -138,6 +135,19 @@
}
}
+
+ protected void setUp() throws Exception
+ {
+ super.setUp();
+ TransportConnection.createVMBroker(1);
+ }
+
+ protected void tearDown() throws Exception
+ {
+ super.tearDown();
+ TransportConnection.killAllVMBrokers();
+ }
+
private static void waitForCompletion(int expected, long wait, Receiver[] receivers) throws InterruptedException
{
for (int i = 0; i < receivers.length; i++)
@@ -209,6 +219,6 @@
public static junit.framework.Test suite()
{
- return new VMBrokerSetup(new junit.framework.TestSuite(MultipleConnectionTest.class));
+ return new junit.framework.TestSuite(MultipleConnectionTest.class);
}
}
Modified: incubator/qpid/branches/new_persistence/java/client/src/test/java/org/apache/qpid/test/unit/basic/TextMessageTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/new_persistence/java/client/src/test/java/org/apache/qpid/test/unit/basic/TextMessageTest.java?view=diff&rev=489461&r1=489460&r2=489461
==============================================================================
--- incubator/qpid/branches/new_persistence/java/client/src/test/java/org/apache/qpid/test/unit/basic/TextMessageTest.java (original)
+++ incubator/qpid/branches/new_persistence/java/client/src/test/java/org/apache/qpid/test/unit/basic/TextMessageTest.java Thu Dec 21 13:08:38 2006
@@ -21,11 +21,8 @@
package org.apache.qpid.test.unit.basic;
import org.apache.qpid.client.AMQConnection;
-import org.apache.qpid.client.AMQDestination;
import org.apache.qpid.client.AMQQueue;
import org.apache.qpid.client.AMQSession;
-import org.apache.qpid.client.vmbroker.AMQVMBrokerCreationException;
-import org.apache.qpid.client.transport.TransportConnection;
import org.apache.qpid.client.message.JMSTextMessage;
import org.apache.qpid.testutil.VMBrokerSetup;
import org.apache.log4j.Logger;
Modified: incubator/qpid/branches/new_persistence/java/client/src/test/java/org/apache/qpid/test/unit/client/message/BytesMessageTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/new_persistence/java/client/src/test/java/org/apache/qpid/test/unit/client/message/BytesMessageTest.java?view=diff&rev=489461&r1=489460&r2=489461
==============================================================================
--- incubator/qpid/branches/new_persistence/java/client/src/test/java/org/apache/qpid/test/unit/client/message/BytesMessageTest.java (original)
+++ incubator/qpid/branches/new_persistence/java/client/src/test/java/org/apache/qpid/test/unit/client/message/BytesMessageTest.java Thu Dec 21 13:08:38 2006
@@ -552,16 +552,6 @@
assertEquals((byte)0, result[2]);
}
- public void testToBodyString() throws Exception
- {
- JMSBytesMessage bm = TestMessageHelper.newJMSBytesMessage();
- final String testText = "This is a test";
- bm.writeUTF(testText);
- bm.reset();
- String result = bm.toBodyString();
- assertEquals(testText, result);
- }
-
public void testToBodyStringWithNull() throws Exception
{
JMSBytesMessage bm = TestMessageHelper.newJMSBytesMessage();
Modified: incubator/qpid/branches/new_persistence/java/client/src/test/java/org/apache/qpid/test/unit/client/message/MapMessageTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/new_persistence/java/client/src/test/java/org/apache/qpid/test/unit/client/message/MapMessageTest.java?view=diff&rev=489461&r1=489460&r2=489461
==============================================================================
--- incubator/qpid/branches/new_persistence/java/client/src/test/java/org/apache/qpid/test/unit/client/message/MapMessageTest.java (original)
+++ incubator/qpid/branches/new_persistence/java/client/src/test/java/org/apache/qpid/test/unit/client/message/MapMessageTest.java Thu Dec 21 13:08:38 2006
@@ -14,18 +14,16 @@
* "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.
+ * under the License.
+ *
*
- *
*/
package org.apache.qpid.test.unit.client.message;
-import junit.framework.TestCase;
import junit.framework.Assert;
-import org.apache.qpid.framing.PropertyFieldTable;
+import junit.framework.TestCase;
import org.apache.qpid.client.message.JMSMapMessage;
import org.apache.qpid.client.message.TestMessageHelper;
-import org.apache.log4j.Logger;
import javax.jms.JMSException;
import javax.jms.MessageFormatException;
@@ -104,6 +102,27 @@
{
Assert.fail("JMSException received." + e);
}
+
+ try
+ {
+ JMSMapMessage mm = TestMessageHelper.newJMSMapMessage();
+
+ mm.setString("value", null);
+ char c = mm.getChar("value");
+ fail("Expected NullPointerException");
+
+ }
+ catch (NullPointerException e)
+ {
+ ; // pass
+ }
+ catch (JMSException e)
+ {
+ Assert.fail("JMSException received." + e);
+ }
+
+
+
}
public void testDoubleLookup()
@@ -204,7 +223,7 @@
{
JMSMapMessage mm = TestMessageHelper.newJMSMapMessage();
mm.getByte("random");
- Assert.fail("MessageFormatException expected");
+ Assert.fail("NumberFormatException expected");
}
catch (NumberFormatException e)
{
@@ -331,7 +350,7 @@
{
JMSMapMessage mm = TestMessageHelper.newJMSMapMessage();
mm.getShort("random");
- Assert.fail("NumberFormatException should be received.");
+ Assert.fail("NumberFormatException should be received.");
}
catch (NumberFormatException e)
{
Modified: incubator/qpid/branches/new_persistence/java/client/src/test/java/org/apache/qpid/test/unit/client/message/ObjectMessageTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/new_persistence/java/client/src/test/java/org/apache/qpid/test/unit/client/message/ObjectMessageTest.java?view=diff&rev=489461&r1=489460&r2=489461
==============================================================================
--- incubator/qpid/branches/new_persistence/java/client/src/test/java/org/apache/qpid/test/unit/client/message/ObjectMessageTest.java (original)
+++ incubator/qpid/branches/new_persistence/java/client/src/test/java/org/apache/qpid/test/unit/client/message/ObjectMessageTest.java Thu Dec 21 13:08:38 2006
@@ -167,6 +167,12 @@
}
+ public void testSetObjectForNull() throws Exception
+ {
+ ObjectMessage msg = session.createObjectMessage();
+ msg.setObject(null);
+ assertNull(msg.getObject());
+ }
private void send() throws Exception
Modified: incubator/qpid/branches/new_persistence/java/client/src/test/java/org/apache/qpid/test/unit/client/message/TextMessageTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/new_persistence/java/client/src/test/java/org/apache/qpid/test/unit/client/message/TextMessageTest.java?view=diff&rev=489461&r1=489460&r2=489461
==============================================================================
--- incubator/qpid/branches/new_persistence/java/client/src/test/java/org/apache/qpid/test/unit/client/message/TextMessageTest.java (original)
+++ incubator/qpid/branches/new_persistence/java/client/src/test/java/org/apache/qpid/test/unit/client/message/TextMessageTest.java Thu Dec 21 13:08:38 2006
@@ -7,9 +7,9 @@
* 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
@@ -22,8 +22,13 @@
import org.apache.qpid.client.message.TestMessageHelper;
import org.apache.qpid.client.message.JMSTextMessage;
+import org.apache.qpid.client.message.JMSMapMessage;
import junit.framework.TestCase;
+import junit.framework.Assert;
+
+import javax.jms.JMSException;
+import javax.jms.MessageFormatException;
public class TextMessageTest extends TestCase
{
@@ -46,6 +51,248 @@
val = tm.getText();
assertEquals(val, "Banana");
}
+
+
+ public void testBooleanPropertyLookup()
+ {
+ try
+ {
+ JMSTextMessage tm = TestMessageHelper.newJMSTextMessage();
+
+ tm.setBooleanProperty("value", true);
+ Assert.assertEquals(true, tm.getBooleanProperty("value"));
+ Assert.assertEquals("true", tm.getStringProperty("value"));
+ }
+ catch (JMSException e)
+ {
+ Assert.fail("JMSException received." + e);
+ }
+ }
+
+ public void testBytePropertyLookup()
+ {
+ try
+ {
+ JMSMapMessage mm = TestMessageHelper.newJMSMapMessage();
+ mm.setByteProperty("value", Byte.MAX_VALUE);
+
+ Assert.assertEquals(Byte.MAX_VALUE, mm.getByteProperty("value"));
+ Assert.assertEquals((short) Byte.MAX_VALUE, mm.getShortProperty("value"));
+ Assert.assertEquals(Byte.MAX_VALUE, mm.getIntProperty("value"));
+ Assert.assertEquals((long) Byte.MAX_VALUE, mm.getLongProperty("value"));
+ Assert.assertEquals("" + Byte.MAX_VALUE, mm.getStringProperty("value"));
+
+ }
+ catch (JMSException e)
+ {
+ Assert.fail("JMSException received." + e);
+ }
+ }
+
+ public void testShortPropertyLookup()
+ {
+ try
+ {
+ JMSMapMessage mm = TestMessageHelper.newJMSMapMessage();
+ mm.setShortProperty("value", Short.MAX_VALUE);
+ Assert.assertEquals(Short.MAX_VALUE, mm.getShortProperty("value"));
+ Assert.assertEquals((int) Short.MAX_VALUE, mm.getIntProperty("value"));
+ Assert.assertEquals((long) Short.MAX_VALUE, mm.getLongProperty("value"));
+ Assert.assertEquals("" + Short.MAX_VALUE, mm.getStringProperty("value"));
+ }
+ catch (JMSException e)
+ {
+ Assert.fail("JMSException received." + e);
+ }
+ }
+
+ public void testDoublePropertyLookup()
+ {
+ try
+ {
+ JMSMapMessage mm = TestMessageHelper.newJMSMapMessage();
+ mm.setDoubleProperty("value", Double.MAX_VALUE);
+ Assert.assertEquals(Double.MAX_VALUE, mm.getDoubleProperty("value"));
+ Assert.assertEquals("" + Double.MAX_VALUE, mm.getStringProperty("value"));
+ }
+ catch (JMSException e)
+ {
+ Assert.fail("JMSException received." + e);
+ }
+ }
+
+ public void testFloatPropertyLookup()
+ {
+ try
+ {
+ JMSMapMessage mm = TestMessageHelper.newJMSMapMessage();
+ mm.setFloatProperty("value", Float.MAX_VALUE);
+ Assert.assertEquals(Float.MAX_VALUE, mm.getFloatProperty("value"));
+ Assert.assertEquals((double) Float.MAX_VALUE, mm.getDoubleProperty("value"));
+ Assert.assertEquals("" + Float.MAX_VALUE, mm.getStringProperty("value"));
+ }
+ catch (JMSException e)
+ {
+ Assert.fail("JMSException received." + e);
+ }
+ }
+
+ public void testIntPropertyLookup()
+ {
+ try
+ {
+ JMSMapMessage mm = TestMessageHelper.newJMSMapMessage();
+ mm.setIntProperty("value", Integer.MAX_VALUE);
+ Assert.assertEquals(Integer.MAX_VALUE, mm.getIntProperty("value"));
+ Assert.assertEquals((long) Integer.MAX_VALUE, mm.getLongProperty("value"));
+ Assert.assertEquals("" + Integer.MAX_VALUE, mm.getStringProperty("value"));
+ }
+ catch (JMSException e)
+ {
+ Assert.fail("JMSException received." + e);
+ }
+ }
+
+ public void testLongPropertyLookup()
+ {
+ try
+ {
+ JMSMapMessage mm = TestMessageHelper.newJMSMapMessage();
+ mm.setLongProperty("value", Long.MAX_VALUE);
+ Assert.assertEquals(Long.MAX_VALUE, mm.getLongProperty("value"));
+ Assert.assertEquals("" + Long.MAX_VALUE, mm.getStringProperty("value"));
+ }
+ catch (JMSException e)
+ {
+ Assert.fail("JMSException received." + e);
+ }
+ }
+
+
+ // Failed Lookups
+
+ public void testFailedBooleanPropertyLookup()
+ {
+ try
+ {
+ JMSMapMessage mm = TestMessageHelper.newJMSMapMessage();
+ Assert.assertEquals(false, mm.getBooleanProperty("int"));
+ }
+ catch (JMSException e)
+ {
+ Assert.fail("JMSException received." + e);
+ }
+ }
+
+ public void testFailedBytePropertyLookup()
+ {
+ try
+ {
+ JMSMapMessage mm = TestMessageHelper.newJMSMapMessage();
+ mm.getByteProperty("random");
+ Assert.fail("NumberFormatException expected");
+ }
+ catch (NumberFormatException e)
+ {
+ //normal execution
+ }
+ catch (JMSException e)
+ {
+ Assert.fail("JMSException received:" + e);
+ }
+
+ }
+
+ public void testFailedDoublePropertyLookup()
+ {
+ try
+ {
+ JMSMapMessage mm = TestMessageHelper.newJMSMapMessage();
+ mm.getDoubleProperty("random");
+ Assert.fail("NullPointerException should be received.");
+ }
+ catch (NullPointerException e)
+ {
+ //normal execution
+ }
+ catch (JMSException e)
+ {
+ Assert.fail("JMSException received:" + e);
+ }
+ }
+
+ public void testFailedFloatPropertyLookup()
+ {
+ try
+ {
+ JMSMapMessage mm = TestMessageHelper.newJMSMapMessage();
+ mm.getFloatProperty("random");
+ Assert.fail("NullPointerException should be received.");
+ }
+ catch (NullPointerException e)
+ {
+ //normal execution
+ }
+ catch (JMSException e)
+ {
+ Assert.fail("JMSException received:" + e);
+ }
+ }
+
+ public void testFailedIntPropertyLookup()
+ {
+ try
+ {
+ JMSMapMessage mm = TestMessageHelper.newJMSMapMessage();
+ mm.getIntProperty("random");
+ Assert.fail("NumberFormatException should be received.");
+ }
+ catch (NumberFormatException e)
+ {
+ //normal execution
+ }
+ catch (JMSException e)
+ {
+ Assert.fail("JMSException received:" + e);
+ }
+ }
+
+ public void testFailedLongPropertyLookup()
+ {
+ try
+ {
+ JMSMapMessage mm = TestMessageHelper.newJMSMapMessage();
+ mm.getLongProperty("random");
+ Assert.fail("NumberFormatException should be received.");
+ }
+ catch (NumberFormatException e)
+ {
+ //normal execution
+ }
+ catch (JMSException e)
+ {
+ Assert.fail("JMSException received:" + e);
+ }
+ }
+
+ public void testFailedShortPropertyLookup()
+ {
+ try
+ {
+ JMSMapMessage mm = TestMessageHelper.newJMSMapMessage();
+ mm.getShortProperty("random");
+ Assert.fail("NumberFormatException should be received.");
+ }
+ catch (NumberFormatException e)
+ {
+ //normal execution
+ }
+ catch (JMSException e)
+ {
+ Assert.fail("JMSException received:" + e);
+ }
+ }
+
public static junit.framework.Test suite()
{
Copied: incubator/qpid/branches/new_persistence/java/client/src/test/java/org/apache/qpid/test/unit/client/temporaryqueue/TemporaryQueueTest.java (from r487903, incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/temporaryqueue/TemporaryQueueTest.java)
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/new_persistence/java/client/src/test/java/org/apache/qpid/test/unit/client/temporaryqueue/TemporaryQueueTest.java?view=diff&rev=489461&p1=incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/temporaryqueue/TemporaryQueueTest.java&r1=487903&p2=incubator/qpid/branches/new_persistence/java/client/src/test/java/org/apache/qpid/test/unit/client/temporaryqueue/TemporaryQueueTest.java&r2=489461
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/test/java/org/apache/qpid/test/unit/client/temporaryqueue/TemporaryQueueTest.java (original)
+++ incubator/qpid/branches/new_persistence/java/client/src/test/java/org/apache/qpid/test/unit/client/temporaryqueue/TemporaryQueueTest.java Thu Dec 21 13:08:38 2006
@@ -8,7 +8,6 @@
import org.apache.qpid.AMQConnectionException;
import org.apache.qpid.AMQUnresolvedAddressException;
import org.apache.qpid.url.URLSyntaxException;
-import org.apache.qpid.test.unit.client.connection.ConnectionTest;
import javax.jms.*;
Modified: incubator/qpid/branches/new_persistence/java/client/src/test/java/org/apache/qpid/test/unit/message/StreamMessageTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/new_persistence/java/client/src/test/java/org/apache/qpid/test/unit/message/StreamMessageTest.java?view=diff&rev=489461&r1=489460&r2=489461
==============================================================================
--- incubator/qpid/branches/new_persistence/java/client/src/test/java/org/apache/qpid/test/unit/message/StreamMessageTest.java (original)
+++ incubator/qpid/branches/new_persistence/java/client/src/test/java/org/apache/qpid/test/unit/message/StreamMessageTest.java Thu Dec 21 13:08:38 2006
@@ -73,7 +73,7 @@
MessageProducer mandatoryProducer = producerSession.createProducer(queue);
// Third test - should be routed
- _logger.info("Sending routable message");
+ _logger.info("Sending isBound message");
StreamMessage msg = producerSession.createStreamMessage();
msg.setStringProperty("F1000","1");
Modified: incubator/qpid/branches/new_persistence/java/client/src/test/java/org/apache/qpid/test/unit/topic/TopicSessionTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/new_persistence/java/client/src/test/java/org/apache/qpid/test/unit/topic/TopicSessionTest.java?view=diff&rev=489461&r1=489460&r2=489461
==============================================================================
--- incubator/qpid/branches/new_persistence/java/client/src/test/java/org/apache/qpid/test/unit/topic/TopicSessionTest.java (original)
+++ incubator/qpid/branches/new_persistence/java/client/src/test/java/org/apache/qpid/test/unit/topic/TopicSessionTest.java Thu Dec 21 13:08:38 2006
@@ -26,10 +26,8 @@
import org.apache.qpid.client.AMQTopic;
import org.apache.qpid.client.transport.TransportConnection;
-import javax.jms.TopicSession;
-import javax.jms.TextMessage;
-import javax.jms.TopicPublisher;
-import javax.jms.MessageConsumer;
+import javax.jms.*;
+
/**
* @author Apache Software Foundation
@@ -46,12 +44,126 @@
{
super.tearDown();
TransportConnection.killAllVMBrokers();
+ //Thread.sleep(2000);
}
- public void testTextMessageCreation() throws Exception
+
+ public void testTopicSubscriptionUnsubscription() throws Exception
{
AMQTopic topic = new AMQTopic("MyTopic");
- AMQConnection con = new AMQConnection("vm://:1", "guest", "guest", "test", "/test");
+ AMQConnection con = new AMQConnection("vm://:1?retries='0'", "guest", "guest", "test", "/test");
+ TopicSession session1 = con.createTopicSession(false, AMQSession.NO_ACKNOWLEDGE);
+ TopicSubscriber sub = session1.createDurableSubscriber(topic,"subscription0");
+ TopicPublisher publisher = session1.createPublisher(topic);
+
+ con.start();
+
+ TextMessage tm = session1.createTextMessage("Hello");
+ publisher.publish(tm);
+
+ tm = (TextMessage) sub.receive(2000);
+ assertNotNull(tm);
+
+ session1.unsubscribe("subscription0");
+
+ try
+ {
+ session1.unsubscribe("not a subscription");
+ fail("expected InvalidDestinationException when unsubscribing from unknown subscription");
+ }
+ catch(InvalidDestinationException e)
+ {
+ ; // PASS
+ }
+ catch(Exception e)
+ {
+ fail("expected InvalidDestinationException when unsubscribing from unknown subscription, got: " + e);
+ }
+
+ con.close();
+ }
+
+ public void testSubscriptionNameReuseForDifferentTopicSingleConnection() throws Exception
+ {
+ subscriptionNameReuseForDifferentTopic(false);
+ }
+
+ public void testSubscriptionNameReuseForDifferentTopicTwoConnections() throws Exception
+ {
+ subscriptionNameReuseForDifferentTopic(true);
+ }
+
+ private void subscriptionNameReuseForDifferentTopic(boolean shutdown) throws Exception
+ {
+ AMQTopic topic = new AMQTopic("MyTopic1" + String.valueOf(shutdown));
+ AMQTopic topic2 = new AMQTopic("MyOtherTopic1" + String.valueOf(shutdown));
+ AMQConnection con = new AMQConnection("vm://:1?retries='0'", "guest", "guest", "test", "/test");
+ TopicSession session1 = con.createTopicSession(false, AMQSession.NO_ACKNOWLEDGE);
+ TopicSubscriber sub = session1.createDurableSubscriber(topic, "subscription0");
+ TopicPublisher publisher = session1.createPublisher(null);
+
+ con.start();
+
+ publisher.publish(topic, session1.createTextMessage("hello"));
+ TextMessage m = (TextMessage) sub.receive(2000);
+ assertNotNull(m);
+
+ if (shutdown)
+ {
+ session1.close();
+ con.close();
+ con = new AMQConnection("vm://:1?retries='0'", "guest", "guest", "test", "/test");
+ con.start();
+ session1 = con.createTopicSession(false, AMQSession.NO_ACKNOWLEDGE);
+ publisher = session1.createPublisher(null);
+ }
+ TopicSubscriber sub2 = session1.createDurableSubscriber(topic2, "subscription0");
+ publisher.publish(topic, session1.createTextMessage("hello"));
+ if (!shutdown)
+ {
+ m = (TextMessage) sub.receive(2000);
+ assertNull(m);
+ }
+ publisher.publish(topic2, session1.createTextMessage("goodbye"));
+ m = (TextMessage) sub2.receive(2000);
+ assertNotNull(m);
+ assertEquals("goodbye", m.getText());
+ con.close();
+ }
+
+ public void testUnsubscriptionAfterConnectionClose() throws Exception
+ {
+ AMQTopic topic = new AMQTopic("MyTopic3");
+ AMQConnection con1 = new AMQConnection("vm://:1?retries='0'", "guest", "guest", "test", "/test");
+ TopicSession session1 = con1.createTopicSession(false, AMQSession.NO_ACKNOWLEDGE);
+ TopicPublisher publisher = session1.createPublisher(topic);
+
+ AMQConnection con2 = new AMQConnection("vm://:1?retries='0'", "guest", "guest", "test2", "/test");
+ TopicSession session2 = con2.createTopicSession(false, AMQSession.NO_ACKNOWLEDGE);
+ TopicSubscriber sub = session2.createDurableSubscriber(topic, "subscription0");
+
+ con2.start();
+
+ publisher.publish(session1.createTextMessage("Hello"));
+ TextMessage tm = (TextMessage) sub.receive(2000);
+ assertNotNull(tm);
+ con2.close();
+ publisher.publish(session1.createTextMessage("Hello2"));
+ con2 = new AMQConnection("vm://:1?retries='0'", "guest", "guest", "test2", "/test");
+ session2 = con2.createTopicSession(false, AMQSession.NO_ACKNOWLEDGE);
+ sub = session2.createDurableSubscriber(topic, "subscription0");
+ con2.start();
+ tm = (TextMessage) sub.receive(2000);
+ assertNotNull(tm);
+ assertEquals("Hello2", tm.getText());
+ con1.close();
+ con2.close();
+ }
+
+ public void testTextMessageCreation() throws Exception
+ {
+ AMQTopic topic = new AMQTopic("MyTopic4");
+ AMQConnection con = new AMQConnection("vm://:1?retries='0'", "guest", "guest", "test", "/test");
TopicSession session1 = con.createTopicSession(false, AMQSession.NO_ACKNOWLEDGE);
TopicPublisher publisher = session1.createPublisher(topic);
MessageConsumer consumer1 = session1.createConsumer(topic);
@@ -85,7 +197,82 @@
tm = (TextMessage) consumer1.receive(2000);
assertNotNull(tm);
assertEquals("Empty string not returned", "", msgText);
+ con.close();
}
+
+ public void testSendingSameMessage() throws Exception
+ {
+ AMQConnection conn = new AMQConnection("vm://:1?retries='0'", "guest", "guest", "test", "/test");
+ TopicSession session = conn.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
+ TemporaryTopic topic = session.createTemporaryTopic();
+ assertNotNull(topic);
+ TopicPublisher producer = session.createPublisher(topic);
+ MessageConsumer consumer = session.createConsumer(topic);
+ conn.start();
+ TextMessage sentMessage = session.createTextMessage("Test Message");
+ producer.send(sentMessage);
+ TextMessage receivedMessage = (TextMessage) consumer.receive(2000);
+ assertNotNull(receivedMessage);
+ assertEquals(sentMessage.getText(),receivedMessage.getText());
+ producer.send(sentMessage);
+ receivedMessage = (TextMessage) consumer.receive(2000);
+ assertNotNull(receivedMessage);
+ assertEquals(sentMessage.getText(),receivedMessage.getText());
+
+
+ }
+
+ public void testTemporaryTopic() throws Exception
+ {
+ AMQConnection conn = new AMQConnection("vm://:1?retries='0'", "guest", "guest", "test", "/test");
+ TopicSession session = conn.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
+ TemporaryTopic topic = session.createTemporaryTopic();
+ assertNotNull(topic);
+ TopicPublisher producer = session.createPublisher(topic);
+ MessageConsumer consumer = session.createConsumer(topic);
+ conn.start();
+ producer.send(session.createTextMessage("hello"));
+ TextMessage tm = (TextMessage) consumer.receive(2000);
+ assertNotNull(tm);
+ assertEquals("hello",tm.getText());
+
+ try
+ {
+ topic.delete();
+ fail("Expected JMSException : should not be able to delete while there are active consumers");
+ }
+ catch(JMSException je)
+ {
+ ; //pass
+ }
+
+ consumer.close();
+
+ try
+ {
+ topic.delete();
+ }
+ catch(JMSException je)
+ {
+ fail("Unexpected Exception: " + je.getMessage());
+ }
+
+ TopicSession session2 = conn.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
+ try
+ {
+ MessageConsumer consumer2 = session2.createConsumer(topic);
+ fail("Expected a JMSException when subscribing to a temporary topic created on adifferent session");
+ }
+ catch (JMSException je)
+ {
+ ; // pass
+ }
+
+
+
+ conn.close();
+ }
+
public static junit.framework.Test suite()
{
Added: incubator/qpid/branches/new_persistence/java/client/src/test/java/org/apache/qpid/test/unit/transacted/TransactedTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/new_persistence/java/client/src/test/java/org/apache/qpid/test/unit/transacted/TransactedTest.java?view=auto&rev=489461
==============================================================================
--- incubator/qpid/branches/new_persistence/java/client/src/test/java/org/apache/qpid/test/unit/transacted/TransactedTest.java (added)
+++ incubator/qpid/branches/new_persistence/java/client/src/test/java/org/apache/qpid/test/unit/transacted/TransactedTest.java Thu Dec 21 13:08:38 2006
@@ -0,0 +1,145 @@
+/*
+ *
+ * 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.qpid.test.unit.transacted;
+
+import org.apache.qpid.client.AMQConnection;
+import org.apache.qpid.client.AMQQueue;
+import org.apache.qpid.client.AMQSession;
+import org.apache.qpid.testutil.VMBrokerSetup;
+
+import javax.jms.*;
+
+import junit.framework.TestCase;
+
+public class TransactedTest extends TestCase
+{
+ private AMQQueue queue1;
+ private AMQQueue queue2;
+
+ private AMQConnection con;
+ private Session session;
+ private MessageConsumer consumer1;
+ private MessageProducer producer2;
+
+ private AMQConnection prepCon;
+ private Session prepSession;
+ private MessageProducer prepProducer1;
+
+ private AMQConnection testCon;
+ private Session testSession;
+ private MessageConsumer testConsumer1;
+ private MessageConsumer testConsumer2;
+
+ protected void setUp() throws Exception
+ {
+ super.setUp();
+ queue1 = new AMQQueue("Q1", false);
+ queue2 = new AMQQueue("Q2", false);
+
+ con = new AMQConnection("vm://:1", "guest", "guest", "TransactedTest", "/test");
+ session = con.createSession(true, 0);
+ consumer1 = session.createConsumer(queue1);
+ //Dummy just to create the queue.
+ MessageConsumer consumer2 = session.createConsumer(queue2);
+ consumer2.close();
+ producer2 = session.createProducer(queue2);
+ con.start();
+
+ prepCon = new AMQConnection("vm://:1", "guest", "guest", "PrepConnection", "/test");
+ prepSession = prepCon.createSession(false, AMQSession.NO_ACKNOWLEDGE);
+ prepProducer1 = prepSession.createProducer(queue1);
+ prepCon.start();
+
+
+ //add some messages
+ prepProducer1.send(prepSession.createTextMessage("A"));
+ prepProducer1.send(prepSession.createTextMessage("B"));
+ prepProducer1.send(prepSession.createTextMessage("C"));
+
+ testCon = new AMQConnection("vm://:1", "guest", "guest", "TestConnection", "/test");
+ testSession = testCon.createSession(false, AMQSession.NO_ACKNOWLEDGE);
+ testConsumer1 = testSession.createConsumer(queue1);
+ testConsumer2 = testSession.createConsumer(queue2);
+ testCon.start();
+ }
+
+ protected void tearDown() throws Exception
+ {
+ con.close();
+ testCon.close();
+ prepCon.close();
+ super.tearDown();
+ }
+
+ public void testCommit() throws Exception
+ {
+ //send and receive some messages
+ producer2.send(session.createTextMessage("X"));
+ producer2.send(session.createTextMessage("Y"));
+ producer2.send(session.createTextMessage("Z"));
+ expect("A", consumer1.receive(1000));
+ expect("B", consumer1.receive(1000));
+ expect("C", consumer1.receive(1000));
+
+ //commit
+ session.commit();
+
+ //ensure sent messages can be received and received messages are gone
+ expect("X", testConsumer2.receive(1000));
+ expect("Y", testConsumer2.receive(1000));
+ expect("Z", testConsumer2.receive(1000));
+
+ assertTrue(null == testConsumer1.receive(1000));
+ assertTrue(null == testConsumer2.receive(1000));
+ }
+
+ public void testRollback() throws Exception
+ {
+ producer2.send(session.createTextMessage("X"));
+ producer2.send(session.createTextMessage("Y"));
+ producer2.send(session.createTextMessage("Z"));
+ expect("A", consumer1.receive(1000));
+ expect("B", consumer1.receive(1000));
+ expect("C", consumer1.receive(1000));
+
+ //rollback
+ session.rollback();
+
+ //ensure sent messages are not visible and received messages are requeued
+ expect("A", consumer1.receive(1000));
+ expect("B", consumer1.receive(1000));
+ expect("C", consumer1.receive(1000));
+
+ assertTrue(null == testConsumer1.receive(1000));
+ assertTrue(null == testConsumer2.receive(1000));
+ }
+
+ private void expect(String text, Message msg) throws JMSException
+ {
+ assertTrue(msg instanceof TextMessage);
+ assertEquals(text, ((TextMessage) msg).getText());
+ }
+
+ public static junit.framework.Test suite()
+ {
+ return new VMBrokerSetup(new junit.framework.TestSuite(TransactedTest.class));
+ }
+}
Propchange: incubator/qpid/branches/new_persistence/java/client/src/test/java/org/apache/qpid/test/unit/transacted/TransactedTest.java
------------------------------------------------------------------------------
svn:eol-style = native
Modified: incubator/qpid/branches/new_persistence/java/common/pom.xml
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/new_persistence/java/common/pom.xml?view=diff&rev=489461&r1=489460&r2=489461
==============================================================================
--- incubator/qpid/branches/new_persistence/java/common/pom.xml (original)
+++ incubator/qpid/branches/new_persistence/java/common/pom.xml Thu Dec 21 13:08:38 2006
@@ -90,5 +90,9 @@
<groupId>junit</groupId>
<artifactId>junit</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.apache.geronimo.specs</groupId>
+ <artifactId>geronimo-jms_1.1_spec</artifactId>
+ </dependency>
</dependencies>
</project>
Modified: incubator/qpid/branches/new_persistence/java/common/src/main/java/org/apache/qpid/framing/AMQBody.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/new_persistence/java/common/src/main/java/org/apache/qpid/framing/AMQBody.java?view=diff&rev=489461&r1=489460&r2=489461
==============================================================================
--- incubator/qpid/branches/new_persistence/java/common/src/main/java/org/apache/qpid/framing/AMQBody.java (original)
+++ incubator/qpid/branches/new_persistence/java/common/src/main/java/org/apache/qpid/framing/AMQBody.java Thu Dec 21 13:08:38 2006
@@ -24,7 +24,7 @@
public abstract class AMQBody
{
- protected abstract byte getType();
+ protected abstract byte getFrameType();
/**
* Get the size of the body
Modified: incubator/qpid/branches/new_persistence/java/common/src/main/java/org/apache/qpid/framing/AMQFrame.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/new_persistence/java/common/src/main/java/org/apache/qpid/framing/AMQFrame.java?view=diff&rev=489461&r1=489460&r2=489461
==============================================================================
--- incubator/qpid/branches/new_persistence/java/common/src/main/java/org/apache/qpid/framing/AMQFrame.java (original)
+++ incubator/qpid/branches/new_persistence/java/common/src/main/java/org/apache/qpid/framing/AMQFrame.java Thu Dec 21 13:08:38 2006
@@ -45,7 +45,7 @@
public void writePayload(ByteBuffer buffer)
{
- buffer.put(bodyFrame.getType());
+ buffer.put(bodyFrame.getFrameType());
// TODO: how does channel get populated
EncodingUtils.writeUnsignedShort(buffer, channel);
EncodingUtils.writeUnsignedInteger(buffer, bodyFrame.getSize());
Modified: incubator/qpid/branches/new_persistence/java/common/src/main/java/org/apache/qpid/framing/AMQMethodBody.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/new_persistence/java/common/src/main/java/org/apache/qpid/framing/AMQMethodBody.java?view=diff&rev=489461&r1=489460&r2=489461
==============================================================================
--- incubator/qpid/branches/new_persistence/java/common/src/main/java/org/apache/qpid/framing/AMQMethodBody.java (original)
+++ incubator/qpid/branches/new_persistence/java/common/src/main/java/org/apache/qpid/framing/AMQMethodBody.java Thu Dec 21 13:08:38 2006
@@ -42,7 +42,7 @@
protected abstract void writeMethodPayload(ByteBuffer buffer);
- protected byte getType()
+ protected byte getFrameType()
{
return TYPE;
}
Modified: incubator/qpid/branches/new_persistence/java/common/src/main/java/org/apache/qpid/framing/BasicContentHeaderProperties.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/new_persistence/java/common/src/main/java/org/apache/qpid/framing/BasicContentHeaderProperties.java?view=diff&rev=489461&r1=489460&r2=489461
==============================================================================
--- incubator/qpid/branches/new_persistence/java/common/src/main/java/org/apache/qpid/framing/BasicContentHeaderProperties.java (original)
+++ incubator/qpid/branches/new_persistence/java/common/src/main/java/org/apache/qpid/framing/BasicContentHeaderProperties.java Thu Dec 21 13:08:38 2006
@@ -57,6 +57,8 @@
private FieldTable _headers;
+ private JMSPropertyFieldTable _jmsHeaders;
+
private byte _deliveryMode;
private byte _priority;
@@ -276,6 +278,7 @@
if ((_propertyFlags & (1 << 13)) > 0)
{
_headers = EncodingUtils.readFieldTable(buffer);
+ setJMSHeaders();
}
if ((_propertyFlags & (1 << 12)) > 0)
{
@@ -358,6 +361,8 @@
if ((_propertyFlags & (1 << 13)) > 0)
{
_headers = EncodingUtils.readFieldTable(buffer);
+ setJMSHeaders();
+
}
_decodedHeaders = true;
}
@@ -446,6 +451,26 @@
clearEncodedForm();
_propertyFlags |= (1 << 13);
_headers = headers;
+ setJMSHeaders();
+ }
+
+ private void setJMSHeaders()
+ {
+ if (_jmsHeaders == null)
+ {
+ _jmsHeaders = new JMSPropertyFieldTable(_headers);
+ }
+ else
+ {
+ _jmsHeaders.setFieldTable(_headers);
+ }
+ }
+
+ public JMSPropertyFieldTable getJMSHeaders()
+ {
+ //This will ensure we have a blank header
+ getHeaders();
+ return _jmsHeaders;
}
public byte getDeliveryMode()
Modified: incubator/qpid/branches/new_persistence/java/common/src/main/java/org/apache/qpid/framing/ContentBody.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/new_persistence/java/common/src/main/java/org/apache/qpid/framing/ContentBody.java?view=diff&rev=489461&r1=489460&r2=489461
==============================================================================
--- incubator/qpid/branches/new_persistence/java/common/src/main/java/org/apache/qpid/framing/ContentBody.java (original)
+++ incubator/qpid/branches/new_persistence/java/common/src/main/java/org/apache/qpid/framing/ContentBody.java Thu Dec 21 13:08:38 2006
@@ -28,7 +28,7 @@
public ByteBuffer payload;
- protected byte getType()
+ protected byte getFrameType()
{
return TYPE;
}
Modified: incubator/qpid/branches/new_persistence/java/common/src/main/java/org/apache/qpid/framing/ContentHeaderBody.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/new_persistence/java/common/src/main/java/org/apache/qpid/framing/ContentHeaderBody.java?view=diff&rev=489461&r1=489460&r2=489461
==============================================================================
--- incubator/qpid/branches/new_persistence/java/common/src/main/java/org/apache/qpid/framing/ContentHeaderBody.java (original)
+++ incubator/qpid/branches/new_persistence/java/common/src/main/java/org/apache/qpid/framing/ContentHeaderBody.java Thu Dec 21 13:08:38 2006
@@ -53,7 +53,7 @@
this.bodySize = bodySize;
}
- protected byte getType()
+ protected byte getFrameType()
{
return TYPE;
}
Modified: incubator/qpid/branches/new_persistence/java/common/src/main/java/org/apache/qpid/framing/EncodingUtils.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/new_persistence/java/common/src/main/java/org/apache/qpid/framing/EncodingUtils.java?view=diff&rev=489461&r1=489460&r2=489461
==============================================================================
--- incubator/qpid/branches/new_persistence/java/common/src/main/java/org/apache/qpid/framing/EncodingUtils.java (original)
+++ incubator/qpid/branches/new_persistence/java/common/src/main/java/org/apache/qpid/framing/EncodingUtils.java Thu Dec 21 13:08:38 2006
@@ -461,7 +461,7 @@
//**** new methods
- // BOOLEAN_PROPERTY_PREFIX
+ // AMQP_BOOLEAN_PROPERTY_PREFIX
public static void writeBoolean(ByteBuffer buffer, Boolean aBoolean)
{
@@ -479,7 +479,7 @@
return 1;
}
- // BYTE_PROPERTY_PREFIX
+ // AMQP_BYTE_PROPERTY_PREFIX
public static void writeByte(ByteBuffer buffer, Byte aByte)
{
buffer.put(aByte);
@@ -496,7 +496,7 @@
}
- // SHORT_PROPERTY_PREFIX
+ // AMQP_SHORT_PROPERTY_PREFIX
public static void writeShort(ByteBuffer buffer, Short aShort)
{
buffer.putShort(aShort);
@@ -528,7 +528,7 @@
return 4;
}
- // LONG_PROPERTY_PREFIX
+ // AMQP_LONG_PROPERTY_PREFIX
public static void writeLong(ByteBuffer buffer, Long aLong)
{
buffer.putLong(aLong);
@@ -610,7 +610,7 @@
}
//CHAR_PROPERTY
- public static long encodedCharLength()
+ public static int encodedCharLength()
{
return encodedByteLength();
}
Modified: incubator/qpid/branches/new_persistence/java/common/src/main/java/org/apache/qpid/framing/FieldTable.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/new_persistence/java/common/src/main/java/org/apache/qpid/framing/FieldTable.java?view=diff&rev=489461&r1=489460&r2=489461
==============================================================================
--- incubator/qpid/branches/new_persistence/java/common/src/main/java/org/apache/qpid/framing/FieldTable.java (original)
+++ incubator/qpid/branches/new_persistence/java/common/src/main/java/org/apache/qpid/framing/FieldTable.java Thu Dec 21 13:08:38 2006
@@ -93,4 +93,6 @@
public Object setObject(String string, Object object);
+ public boolean isNullStringValue(String name);
+
}
Modified: incubator/qpid/branches/new_persistence/java/common/src/main/java/org/apache/qpid/framing/HeartbeatBody.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/new_persistence/java/common/src/main/java/org/apache/qpid/framing/HeartbeatBody.java?view=diff&rev=489461&r1=489460&r2=489461
==============================================================================
--- incubator/qpid/branches/new_persistence/java/common/src/main/java/org/apache/qpid/framing/HeartbeatBody.java (original)
+++ incubator/qpid/branches/new_persistence/java/common/src/main/java/org/apache/qpid/framing/HeartbeatBody.java Thu Dec 21 13:08:38 2006
@@ -27,7 +27,7 @@
public static final byte TYPE = 8;
public static AMQFrame FRAME = new HeartbeatBody().toFrame();
- protected byte getType()
+ protected byte getFrameType()
{
return TYPE;
}