7:50 p.m.
Author: clebert.suconic(a)jboss.com
Date: 2010-12-20 20:50:48 -0500 (Mon, 20 Dec 2010)
New Revision: 10063
Modified:
trunk/src/main/org/hornetq/jms/bridge/impl/JMSBridgeImpl.java
trunk/src/main/org/hornetq/jms/client/HornetQMessage.java
trunk/tests/src/org/hornetq/tests/integration/client/LargeMessageCompressTest.java
trunk/tests/src/org/hornetq/tests/integration/jms/server/management/JMSQueueControlUsingJMSTest.java
trunk/tests/src/org/hornetq/tests/integration/largemessage/LargeMessageTestBase.java
Log:
fixing tests on Large Message
Modified: trunk/src/main/org/hornetq/jms/bridge/impl/JMSBridgeImpl.java
===================================================================
--- trunk/src/main/org/hornetq/jms/bridge/impl/JMSBridgeImpl.java 2010-12-20 15:10:20 UTC
(rev 10062)
+++ trunk/src/main/org/hornetq/jms/bridge/impl/JMSBridgeImpl.java 2010-12-21 01:50:48 UTC
(rev 10063)
@@ -1730,6 +1730,13 @@
try
{
msg = sourceConsumer.receive(1000);
+
+ if (msg instanceof HornetQMessage)
+ {
+ // We need to check the buffer mainly in the case of LargeMessages
+ // As we need to reconstruct the buffer before resending the
message
+ ((HornetQMessage)msg).checkBuffer();
+ }
}
catch (JMSException jmse)
{
Modified: trunk/src/main/org/hornetq/jms/client/HornetQMessage.java
===================================================================
--- trunk/src/main/org/hornetq/jms/client/HornetQMessage.java 2010-12-20 15:10:20 UTC (rev
10062)
+++ trunk/src/main/org/hornetq/jms/client/HornetQMessage.java 2010-12-21 01:50:48 UTC (rev
10063)
@@ -879,6 +879,11 @@
{
message.getBodyBuffer().resetReaderIndex();
}
+
+ public void checkBuffer()
+ {
+ message.getBodyBuffer();
+ }
public void doBeforeReceive() throws Exception
{
Modified:
trunk/tests/src/org/hornetq/tests/integration/client/LargeMessageCompressTest.java
===================================================================
---
trunk/tests/src/org/hornetq/tests/integration/client/LargeMessageCompressTest.java 2010-12-20
15:10:20 UTC (rev 10062)
+++
trunk/tests/src/org/hornetq/tests/integration/client/LargeMessageCompressTest.java 2010-12-21
01:50:48 UTC (rev 10063)
@@ -27,6 +27,7 @@
import org.hornetq.api.core.client.ClientSessionFactory;
import org.hornetq.api.core.client.HornetQClient;
import org.hornetq.api.core.client.ServerLocator;
+import org.hornetq.tests.util.RandomUtil;
/**
* A LargeMessageCompressTest
@@ -277,6 +278,111 @@
}
+
+ public void testLargeMessageCompressionRestartAndCheckSize() throws Exception
+ {
+ final int messageSize = 1024 * 1024;
+
+ ClientSession session = null;
+
+ try
+ {
+ server = createServer(true, isNetty());
+
+ server.start();
+
+ ClientSessionFactory sf = locator.createSessionFactory();
+
+ session = sf.createSession(false, false, false);
+
+ session.createQueue(LargeMessageTest.ADDRESS, LargeMessageTest.ADDRESS, true);
+
+ ClientProducer producer = session.createProducer(LargeMessageTest.ADDRESS);
+
+ byte [] msgs = new byte[1024 * 1024];
+ for (int i = 0 ; i < msgs.length; i++)
+ {
+ msgs[i] = RandomUtil.randomByte();
+ }
+
+ Message clientFile = createLargeClientMessage(session, msgs, true);
+
+ producer.send(clientFile);
+
+ session.commit();
+
+ session.close();
+
+ sf.close();
+
+ locator.close();
+
+ server.stop();
+
+ server = createServer(true, isNetty());
+
+ server.start();
+
+ locator = createFactory(isNetty());
+
+ sf = locator.createSessionFactory();
+
+ session = sf.createSession();
+
+ session.start();
+
+ ClientConsumer consumer = session.createConsumer(LargeMessageTest.ADDRESS);
+ ClientMessage msg1 = consumer.receive(1000);
+ Assert.assertNotNull(msg1);
+
+ assertEquals(messageSize, msg1.getBodySize());
+
+ String testDir = this.getTestDir();
+ File testFile = new File(testDir, "async_large_message");
+ FileOutputStream output = new FileOutputStream(testFile);
+
+ msg1.saveToOutputStream(output);
+
+ msg1.acknowledge();
+
+ session.commit();
+
+ consumer.close();
+
+ session.close();
+
+ //verify
+ FileInputStream input = new FileInputStream(testFile);
+ for (int i = 0 ; i < messageSize; i++)
+ {
+ byte b = (byte)input.read();
+ assertEquals("position = " + i, msgs[i], b);
+ }
+
+ testFile.delete();
+ validateNoFilesOnLargeDir();
+ }
+ finally
+ {
+ try
+ {
+ server.stop();
+ }
+ catch (Throwable ignored)
+ {
+ }
+
+ try
+ {
+ session.close();
+ }
+ catch (Throwable ignored)
+ {
+ }
+ }
+ }
+
+
public void testSendServerMessage() throws Exception
{
// doesn't make sense as compressed
Modified:
trunk/tests/src/org/hornetq/tests/integration/jms/server/management/JMSQueueControlUsingJMSTest.java
===================================================================
---
trunk/tests/src/org/hornetq/tests/integration/jms/server/management/JMSQueueControlUsingJMSTest.java 2010-12-20
15:10:20 UTC (rev 10062)
+++
trunk/tests/src/org/hornetq/tests/integration/jms/server/management/JMSQueueControlUsingJMSTest.java 2010-12-21
01:50:48 UTC (rev 10063)
@@ -100,7 +100,7 @@
public long countMessages(final String filter) throws Exception
{
- return (Long)proxy.invokeOperation("countMessages", filter);
+ return ((Number)proxy.invokeOperation("countMessages",
filter)).intValue();
}
public boolean expireMessage(final String messageID) throws Exception
Modified:
trunk/tests/src/org/hornetq/tests/integration/largemessage/LargeMessageTestBase.java
===================================================================
---
trunk/tests/src/org/hornetq/tests/integration/largemessage/LargeMessageTestBase.java 2010-12-20
15:10:20 UTC (rev 10062)
+++
trunk/tests/src/org/hornetq/tests/integration/largemessage/LargeMessageTestBase.java 2010-12-21
01:50:48 UTC (rev 10063)
@@ -617,6 +617,13 @@
return createLargeClientMessage(session, numberOfBytes, true);
}
+ protected ClientMessage createLargeClientMessage (final ClientSession session, final
byte[] buffer, final boolean durable) throws Exception
+ {
+ ClientMessage msgs = session.createMessage(durable);
+ msgs.getBodyBuffer().writeBytes(buffer);
+ return msgs;
+ }
+
protected ClientMessage createLargeClientMessage(final ClientSession session,
final long numberOfBytes,
final boolean persistent) throws
Exception