10:20 p.m.
Author: clebert.suconic(a)jboss.com
Date: 2009-11-27 16:20:56 -0500 (Fri, 27 Nov 2009)
New Revision: 8433
Modified:
trunk/src/main/org/hornetq/core/persistence/OperationContext.java
trunk/src/main/org/hornetq/core/persistence/impl/journal/OperationContextImpl.java
trunk/src/main/org/hornetq/core/postoffice/impl/PostOfficeImpl.java
trunk/src/main/org/hornetq/core/server/impl/DivertImpl.java
trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java
trunk/tests/src/org/hornetq/tests/integration/divert/PersistentDivertTest.java
trunk/tests/src/org/hornetq/tests/integration/paging/PagingSendTest.java
Log:
https://jira.jboss.org/jira/browse/HORNETQ-226 - Large Message and Diverts
Modified: trunk/src/main/org/hornetq/core/persistence/OperationContext.java
===================================================================
--- trunk/src/main/org/hornetq/core/persistence/OperationContext.java 2009-11-27 21:15:05
UTC (rev 8432)
+++ trunk/src/main/org/hornetq/core/persistence/OperationContext.java 2009-11-27 21:20:56
UTC (rev 8433)
@@ -13,8 +13,6 @@
package org.hornetq.core.persistence;
-import java.util.concurrent.Executor;
-
import org.hornetq.core.journal.IOAsyncTask;
import org.hornetq.core.journal.IOCompletion;
@@ -30,10 +28,6 @@
public interface OperationContext extends IOCompletion
{
- /** The executor used on responses.
- * If this is not set, it will use the current thread. */
- void setExecutor(Executor executor);
-
/** Execute the task when all IO operations are complete,
* Or execute it immediately if nothing is pending. */
void executeOnCompletion(IOAsyncTask runnable);
Modified:
trunk/src/main/org/hornetq/core/persistence/impl/journal/OperationContextImpl.java
===================================================================
---
trunk/src/main/org/hornetq/core/persistence/impl/journal/OperationContextImpl.java 2009-11-27
21:15:05 UTC (rev 8432)
+++
trunk/src/main/org/hornetq/core/persistence/impl/journal/OperationContextImpl.java 2009-11-27
21:20:56 UTC (rev 8433)
@@ -81,7 +81,7 @@
private String errorMessage = null;
- private Executor executor;
+ private final Executor executor;
private final AtomicInteger executorsPending = new AtomicInteger(0);
@@ -102,12 +102,6 @@
replicationLineUp++;
}
- /** this method needs to be called before the executor became operational */
- public void setExecutor(Executor executor)
- {
- this.executor = executor;
- }
-
public synchronized void replicationDone()
{
replicated++;
@@ -137,25 +131,18 @@
// On this case, we can just execute the context directly
if (replicationLineUp == replicated && storeLineUp == stored)
{
- if (executor != null)
+ // We want to avoid the executor if everything is complete...
+ // However, we can't execute the context if there are executions pending
+ // We need to use the executor on this case
+ if (executorsPending.get() == 0)
{
- // We want to avoid the executor if everything is complete...
- // However, we can't execute the context if there are executions
pending
- // We need to use the executor on this case
- if (executorsPending.get() == 0)
- {
- // No need to use an executor here or a context switch
- // there are no actions pending.. hence we can just execute the task
directly on the same thread
- executeNow = true;
- }
- else
- {
- execute(completion);
- }
+ // No need to use an executor here or a context switch
+ // there are no actions pending.. hence we can just execute the task
directly on the same thread
+ executeNow = true;
}
else
{
- executeNow = true;
+ execute(completion);
}
}
else
Modified: trunk/src/main/org/hornetq/core/postoffice/impl/PostOfficeImpl.java
===================================================================
--- trunk/src/main/org/hornetq/core/postoffice/impl/PostOfficeImpl.java 2009-11-27
21:15:05 UTC (rev 8432)
+++ trunk/src/main/org/hornetq/core/postoffice/impl/PostOfficeImpl.java 2009-11-27
21:20:56 UTC (rev 8433)
@@ -550,11 +550,6 @@
throw new IllegalStateException("Message cannot be routed more than
once");
}
- if (message.getMessageID() == 0l)
- {
- generateID(message);
- }
-
RoutingContext context = new RoutingContextImpl(tx);
SimpleString address = message.getDestination();
Modified: trunk/src/main/org/hornetq/core/server/impl/DivertImpl.java
===================================================================
--- trunk/src/main/org/hornetq/core/server/impl/DivertImpl.java 2009-11-27 21:15:05 UTC
(rev 8432)
+++ trunk/src/main/org/hornetq/core/server/impl/DivertImpl.java 2009-11-27 21:20:56 UTC
(rev 8433)
@@ -17,6 +17,7 @@
import org.hornetq.core.filter.Filter;
import org.hornetq.core.logging.Logger;
+import org.hornetq.core.persistence.StorageManager;
import org.hornetq.core.postoffice.PostOffice;
import org.hornetq.core.server.Divert;
import org.hornetq.core.server.RoutingContext;
@@ -50,6 +51,8 @@
private final Filter filter;
private final Transformer transformer;
+
+ private final StorageManager storageManager;
public DivertImpl(final SimpleString forwardAddress,
final SimpleString uniqueName,
@@ -57,7 +60,8 @@
final boolean exclusive,
final Filter filter,
final Transformer transformer,
- final PostOffice postOffice)
+ final PostOffice postOffice,
+ final StorageManager storageManager)
{
this.forwardAddress = forwardAddress;
@@ -72,6 +76,8 @@
this.transformer = transformer;
this.postOffice = postOffice;
+
+ this.storageManager = storageManager;
}
public void route(final ServerMessage message, final RoutingContext context) throws
Exception
@@ -81,17 +87,16 @@
// We must make a copy of the message, otherwise things like returning credits to
the page won't work
// properly on ack, since the original destination will be overwritten
- // TODO we can optimise this so it doesn't copy if it's not routed anywhere
else
+ // TODO we can optimise this so it doesn't copy if it's not routed anywhere
else
- ServerMessage copy = message.copy();
+ long id = storageManager.generateUniqueID();
+ ServerMessage copy = message.copy(id);
- // Setting the messageID to 0. The postOffice should set a new one
- copy.setMessageID(0);
-
+ // This will set the original MessageId, and the original destination
+ copy.setOriginalHeaders(message, false);
+
copy.setDestination(forwardAddress);
- copy.putStringProperty(HDR_ORIGINAL_DESTINATION, originalDestination);
-
if (transformer != null)
{
copy = transformer.transform(copy);
Modified: trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java
===================================================================
--- trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2009-11-27 21:15:05
UTC (rev 8432)
+++ trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2009-11-27 21:20:56
UTC (rev 8433)
@@ -1414,7 +1414,8 @@
config.isExclusive(),
filter,
transformer,
- postOffice);
+ postOffice,
+ storageManager);
// pagingManager,
// storageManager);
Modified: trunk/tests/src/org/hornetq/tests/integration/divert/PersistentDivertTest.java
===================================================================
---
trunk/tests/src/org/hornetq/tests/integration/divert/PersistentDivertTest.java 2009-11-27
21:15:05 UTC (rev 8432)
+++
trunk/tests/src/org/hornetq/tests/integration/divert/PersistentDivertTest.java 2009-11-27
21:20:56 UTC (rev 8433)
@@ -45,8 +45,20 @@
{
private static final Logger log = Logger.getLogger(DivertTest.class);
+ final int minLargeMessageSize =
ClientSessionFactoryImpl.DEFAULT_MIN_LARGE_MESSAGE_SIZE * 2;
+
public void testPersistentDivert() throws Exception
{
+ doTestPersistentDivert(false);
+ }
+
+ public void testPersistentDiverLargeMessage() throws Exception
+ {
+ doTestPersistentDivert(true);
+ }
+
+ public void doTestPersistentDivert(boolean largeMessage) throws Exception
+ {
Configuration conf = createDefaultConfig();
conf.setClustered(true);
@@ -121,6 +133,11 @@
{
ClientMessage message = session.createClientMessage(true);
+ if (largeMessage)
+ {
+ message.setBodyInputStream(createFakeLargeStream(minLargeMessageSize));
+ }
+
message.putIntProperty(propKey, i);
producer.send(message);
@@ -128,12 +145,17 @@
for (int i = 0; i < numMessages; i++)
{
- ClientMessage message = consumer1.receive(200);
+ ClientMessage message = consumer1.receive(5000);
assertNotNull(message);
assertEquals((Integer)i, (Integer)message.getObjectProperty(propKey));
+ if (largeMessage)
+ {
+ checkLargeMessage(message);
+ }
+
message.acknowledge();
}
@@ -141,12 +163,17 @@
for (int i = 0; i < numMessages; i++)
{
- ClientMessage message = consumer2.receive(200);
+ ClientMessage message = consumer2.receive(5000);
assertNotNull(message);
assertEquals((Integer)i, (Integer)message.getObjectProperty(propKey));
+ if (largeMessage)
+ {
+ checkLargeMessage(message);
+ }
+
message.acknowledge();
}
@@ -154,12 +181,17 @@
for (int i = 0; i < numMessages; i++)
{
- ClientMessage message = consumer3.receive(200);
+ ClientMessage message = consumer3.receive(5000);
assertNotNull(message);
assertEquals((Integer)i, (Integer)message.getObjectProperty(propKey));
+ if (largeMessage)
+ {
+ checkLargeMessage(message);
+ }
+
message.acknowledge();
}
@@ -167,12 +199,17 @@
for (int i = 0; i < numMessages; i++)
{
- ClientMessage message = consumer4.receive(200);
+ ClientMessage message = consumer4.receive(5000);
assertNotNull(message);
assertEquals((Integer)i, (Integer)message.getObjectProperty(propKey));
+ if (largeMessage)
+ {
+ checkLargeMessage(message);
+ }
+
message.acknowledge();
}
@@ -183,9 +220,30 @@
messagingService.stop();
}
+
+ /**
+ * @param message
+ */
+ private void checkLargeMessage(ClientMessage message)
+ {
+ for (int j = 0 ; j < minLargeMessageSize; j++)
+ {
+ assertEquals(getSamplebyte(j), message.getBodyBuffer().readByte());
+ }
+ }
public void testPersistentDivertRestartBeforeConsume() throws Exception
{
+ doTestPersistentDivertRestartBeforeConsume(false);
+ }
+
+ public void testPersistentDivertRestartBeforeConsumeLargeMessage() throws Exception
+ {
+ doTestPersistentDivertRestartBeforeConsume(true);
+ }
+
+ public void doTestPersistentDivertRestartBeforeConsume(boolean largeMessage) throws
Exception
+ {
Configuration conf = createDefaultConfig();
conf.setClustered(true);
@@ -251,6 +309,12 @@
ClientMessage message = session.createClientMessage(true);
message.putIntProperty(propKey, i);
+
+ if (largeMessage)
+ {
+ message.setBodyInputStream(createFakeLargeStream(minLargeMessageSize));
+ }
+
producer.send(message);
}
@@ -281,10 +345,15 @@
for (int i = 0; i < numMessages; i++)
{
- ClientMessage message = consumer1.receive(200);
+ ClientMessage message = consumer1.receive(5000);
assertNotNull(message);
+ if (largeMessage)
+ {
+ checkLargeMessage(message);
+ }
+
assertEquals((Integer)i, (Integer)message.getObjectProperty(propKey));
message.acknowledge();
@@ -294,10 +363,15 @@
for (int i = 0; i < numMessages; i++)
{
- ClientMessage message = consumer2.receive(200);
+ ClientMessage message = consumer2.receive(5000);
assertNotNull(message);
+ if (largeMessage)
+ {
+ checkLargeMessage(message);
+ }
+
assertEquals((Integer)i, (Integer)message.getObjectProperty(propKey));
message.acknowledge();
@@ -307,10 +381,15 @@
for (int i = 0; i < numMessages; i++)
{
- ClientMessage message = consumer3.receive(200);
+ ClientMessage message = consumer3.receive(5000);
assertNotNull(message);
+ if (largeMessage)
+ {
+ checkLargeMessage(message);
+ }
+
assertEquals((Integer)i, (Integer)message.getObjectProperty(propKey));
message.acknowledge();
@@ -320,10 +399,15 @@
for (int i = 0; i < numMessages; i++)
{
- ClientMessage message = consumer4.receive(200);
+ ClientMessage message = consumer4.receive(5000);
assertNotNull(message);
+ if (largeMessage)
+ {
+ checkLargeMessage(message);
+ }
+
assertEquals((Integer)i, (Integer)message.getObjectProperty(propKey));
message.acknowledge();
Modified: trunk/tests/src/org/hornetq/tests/integration/paging/PagingSendTest.java
===================================================================
--- trunk/tests/src/org/hornetq/tests/integration/paging/PagingSendTest.java 2009-11-27
21:15:05 UTC (rev 8432)
+++ trunk/tests/src/org/hornetq/tests/integration/paging/PagingSendTest.java 2009-11-27
21:20:56 UTC (rev 8433)
@@ -112,7 +112,6 @@
for (int i = 0; i < 200; i++)
{
- System.out.println("Sent " + i);
producer.send(message);
}