4:18 a.m.
Author: timfox
Date: 2010-03-30 05:18:21 -0400 (Tue, 30 Mar 2010)
New Revision: 9013
Modified:
trunk/src/config/jboss-as-6/clustered/hornetq-jms.xml
trunk/src/config/jboss-as-6/non-clustered/hornetq-jms.xml
trunk/src/config/jboss-as/clustered/hornetq-jms.xml
trunk/src/config/jboss-as/non-clustered/hornetq-jms.xml
trunk/src/config/stand-alone/clustered/hornetq-jms.xml
trunk/src/config/stand-alone/non-clustered/hornetq-jms.xml
trunk/src/config/trunk/clustered/hornetq-jms.xml
trunk/src/config/trunk/non-clustered/hornetq-jms.xml
trunk/src/main/org/hornetq/core/client/impl/ClientConsumerImpl.java
trunk/src/main/org/hornetq/core/client/impl/ClientProducerImpl.java
trunk/src/main/org/hornetq/core/message/impl/MessageImpl.java
trunk/src/main/org/hornetq/core/message/impl/MessageInternal.java
trunk/src/main/org/hornetq/core/server/ServerMessage.java
trunk/src/main/org/hornetq/core/server/impl/ServerMessageImpl.java
trunk/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java
trunk/src/main/org/hornetq/jms/client/HornetQMessageProducer.java
trunk/tests/src/org/hornetq/tests/unit/core/postoffice/impl/BindingsImplTest.java
Log:
mainly default address optimisation
Modified: trunk/src/config/jboss-as/clustered/hornetq-jms.xml
===================================================================
--- trunk/src/config/jboss-as/clustered/hornetq-jms.xml 2010-03-30 09:11:31 UTC (rev
9012)
+++ trunk/src/config/jboss-as/clustered/hornetq-jms.xml 2010-03-30 09:18:21 UTC (rev
9013)
@@ -12,6 +12,16 @@
</entries>
</connection-factory>
+ <connection-factory name="NettyThroughputConnectionFactory">
+ <connectors>
+ <connector-ref connector-name="netty-throughput"/>
+ </connectors>
+ <entries>
+ <entry name="/ThroughputConnectionFactory"/>
+ <entry name="/XAThroughputConnectionFactory"/>
+ </entries>
+ /connection-factory>
+
<connection-factory name="InVMConnectionFactory">
<connectors>
<connector-ref connector-name="in-vm"/>
Modified: trunk/src/config/jboss-as/non-clustered/hornetq-jms.xml
===================================================================
--- trunk/src/config/jboss-as/non-clustered/hornetq-jms.xml 2010-03-30 09:11:31 UTC (rev
9012)
+++ trunk/src/config/jboss-as/non-clustered/hornetq-jms.xml 2010-03-30 09:18:21 UTC (rev
9013)
@@ -12,6 +12,16 @@
</entries>
</connection-factory>
+ <connection-factory name="NettyThroughputConnectionFactory">
+ <connectors>
+ <connector-ref connector-name="netty-throughput"/>
+ </connectors>
+ <entries>
+ <entry name="/ThroughputConnectionFactory"/>
+ <entry name="/XAThroughputConnectionFactory"/>
+ </entries>
+ </connection-factory>
+
<connection-factory name="InVMConnectionFactory">
<connectors>
<connector-ref connector-name="in-vm"/>
Modified: trunk/src/config/jboss-as-6/clustered/hornetq-jms.xml
===================================================================
--- trunk/src/config/jboss-as-6/clustered/hornetq-jms.xml 2010-03-30 09:11:31 UTC (rev
9012)
+++ trunk/src/config/jboss-as-6/clustered/hornetq-jms.xml 2010-03-30 09:18:21 UTC (rev
9013)
@@ -12,6 +12,16 @@
</entries>
</connection-factory>
+ <connection-factory name="NettyThroughputConnectionFactory">
+ <connectors>
+ <connector-ref connector-name="netty-throughput"/>
+ </connectors>
+ <entries>
+ <entry name="/ThroughputConnectionFactory"/>
+ <entry name="/XAThroughputConnectionFactory"/>
+ </entries>
+ </connection-factory>
+
<connection-factory name="InVMConnectionFactory">
<connectors>
<connector-ref connector-name="in-vm"/>
Modified: trunk/src/config/jboss-as-6/non-clustered/hornetq-jms.xml
===================================================================
--- trunk/src/config/jboss-as-6/non-clustered/hornetq-jms.xml 2010-03-30 09:11:31 UTC (rev
9012)
+++ trunk/src/config/jboss-as-6/non-clustered/hornetq-jms.xml 2010-03-30 09:18:21 UTC (rev
9013)
@@ -12,6 +12,16 @@
</entries>
</connection-factory>
+ <connection-factory name="NettyThroughputConnectionFactory">
+ <connectors>
+ <connector-ref connector-name="netty-throughput"/>
+ </connectors>
+ <entries>
+ <entry name="/ThroughputConnectionFactory"/>
+ <entry name="/XAThroughputConnectionFactory"/>
+ </entries>
+ </connection-factory>
+
<connection-factory name="InVMConnectionFactory">
<connectors>
<connector-ref connector-name="in-vm"/>
Modified: trunk/src/config/stand-alone/clustered/hornetq-jms.xml
===================================================================
--- trunk/src/config/stand-alone/clustered/hornetq-jms.xml 2010-03-30 09:11:31 UTC (rev
9012)
+++ trunk/src/config/stand-alone/clustered/hornetq-jms.xml 2010-03-30 09:18:21 UTC (rev
9013)
@@ -12,6 +12,16 @@
</entries>
</connection-factory>
+ <connection-factory name="NettyThroughputConnectionFactory">
+ <connectors>
+ <connector-ref connector-name="netty-throughput"/>
+ </connectors>
+ <entries>
+ <entry name="/ThroughputConnectionFactory"/>
+ <entry name="/XAThroughputConnectionFactory"/>
+ </entries>
+ </connection-factory>
+
<queue name="DLQ">
<entry name="/queue/DLQ"/>
</queue>
Modified: trunk/src/config/stand-alone/non-clustered/hornetq-jms.xml
===================================================================
--- trunk/src/config/stand-alone/non-clustered/hornetq-jms.xml 2010-03-30 09:11:31 UTC
(rev 9012)
+++ trunk/src/config/stand-alone/non-clustered/hornetq-jms.xml 2010-03-30 09:18:21 UTC
(rev 9013)
@@ -12,6 +12,16 @@
</entries>
</connection-factory>
+ <connection-factory name="NettyThroughputConnectionFactory">
+ <connectors>
+ <connector-ref connector-name="netty-throughput"/>
+ </connectors>
+ <entries>
+ <entry name="/ThroughputConnectionFactory"/>
+ <entry name="/XAThroughputConnectionFactory"/>
+ </entries>
+ </connection-factory>
+
<queue name="DLQ">
<entry name="/queue/DLQ"/>
</queue>
Modified: trunk/src/config/trunk/clustered/hornetq-jms.xml
===================================================================
--- trunk/src/config/trunk/clustered/hornetq-jms.xml 2010-03-30 09:11:31 UTC (rev 9012)
+++ trunk/src/config/trunk/clustered/hornetq-jms.xml 2010-03-30 09:18:21 UTC (rev 9013)
@@ -12,6 +12,16 @@
</entries>
</connection-factory>
+ <connection-factory name="NettyThroughputConnectionFactory">
+ <connectors>
+ <connector-ref connector-name="netty-throughput"/>
+ </connectors>
+ <entries>
+ <entry name="/ThroughputConnectionFactory"/>
+ <entry name="/XAThroughputConnectionFactory"/>
+ </entries>
+ </connection-factory>
+
<queue name="DLQ">
<entry name="/queue/DLQ"/>
</queue>
Modified: trunk/src/config/trunk/non-clustered/hornetq-jms.xml
===================================================================
--- trunk/src/config/trunk/non-clustered/hornetq-jms.xml 2010-03-30 09:11:31 UTC (rev
9012)
+++ trunk/src/config/trunk/non-clustered/hornetq-jms.xml 2010-03-30 09:18:21 UTC (rev
9013)
@@ -12,6 +12,18 @@
</entries>
</connection-factory>
+ <!--
+ <connection-factory name="NettyThroughputConnectionFactory">
+ <connectors>
+ <connector-ref connector-name="netty-throughput"/>
+ </connectors>
+ <entries>
+ <entry name="/ThroughputConnectionFactory"/>
+ <entry name="/XAThroughputConnectionFactory"/>
+ </entries>
+ </connection-factory>
+ -->
+
<queue name="DLQ">
<entry name="/queue/DLQ"/>
</queue>
Modified: trunk/src/main/org/hornetq/core/client/impl/ClientConsumerImpl.java
===================================================================
--- trunk/src/main/org/hornetq/core/client/impl/ClientConsumerImpl.java 2010-03-30
09:11:31 UTC (rev 9012)
+++ trunk/src/main/org/hornetq/core/client/impl/ClientConsumerImpl.java 2010-03-30
09:18:21 UTC (rev 9013)
@@ -465,6 +465,11 @@
}
ClientMessageInternal messageToHandle = message;
+
+ if (messageToHandle.getAddress() == null)
+ {
+ messageToHandle.setAddressTransient(queueInfo.getAddress());
+ }
messageToHandle.onReceipt(this);
Modified: trunk/src/main/org/hornetq/core/client/impl/ClientProducerImpl.java
===================================================================
--- trunk/src/main/org/hornetq/core/client/impl/ClientProducerImpl.java 2010-03-30
09:11:31 UTC (rev 9012)
+++ trunk/src/main/org/hornetq/core/client/impl/ClientProducerImpl.java 2010-03-30
09:18:21 UTC (rev 9013)
@@ -214,22 +214,43 @@
MessageInternal msgI = (MessageInternal)msg;
ClientProducerCredits theCredits;
+
+ boolean isLarge;
+ if (msgI.getBodyInputStream() != null || msgI.isLargeMessage())
+ {
+ isLarge = true;
+ }
+ else
+ {
+ isLarge = false;
+ }
+
if (address != null)
{
- //session.setAddress(msg, address);
+ if (!isLarge)
+ {
+ session.setAddress(msg, address);
+ }
+ else
+ {
+ msg.setAddress(address);
+ }
- msg.setAddress(address);
-
// Anonymous
theCredits = session.getCredits(address, true);
}
else
{
- //session.setAddress(msg, this.address);
+ if (!isLarge)
+ {
+ session.setAddress(msg, this.address);
+ }
+ else
+ {
+ msg.setAddress(this.address);
+ }
- msg.setAddress(this.address);
-
theCredits = credits;
}
@@ -249,17 +270,8 @@
session.workDone();
- boolean isLarge;
+
- if (msgI.getBodyInputStream() != null || msgI.isLargeMessage())
- {
- isLarge = true;
- }
- else
- {
- isLarge = false;
- }
-
if (isLarge)
{
largeMessageSend(sendBlocking, msgI, theCredits);
Modified: trunk/src/main/org/hornetq/core/message/impl/MessageImpl.java
===================================================================
--- trunk/src/main/org/hornetq/core/message/impl/MessageImpl.java 2010-03-30 09:11:31 UTC
(rev 9012)
+++ trunk/src/main/org/hornetq/core/message/impl/MessageImpl.java 2010-03-30 09:18:21 UTC
(rev 9013)
@@ -444,7 +444,13 @@
return buffer;
}
}
+
+ public void setAddressTransient(final SimpleString address)
+ {
+ this.address = address;
+ }
+
// Properties
//
---------------------------------------------------------------------------------------
Modified: trunk/src/main/org/hornetq/core/message/impl/MessageInternal.java
===================================================================
--- trunk/src/main/org/hornetq/core/message/impl/MessageInternal.java 2010-03-30 09:11:31
UTC (rev 9012)
+++ trunk/src/main/org/hornetq/core/message/impl/MessageInternal.java 2010-03-30 09:18:21
UTC (rev 9013)
@@ -18,6 +18,7 @@
import org.hornetq.api.core.HornetQBuffer;
import org.hornetq.api.core.HornetQException;
import org.hornetq.api.core.Message;
+import org.hornetq.api.core.SimpleString;
import org.hornetq.core.message.BodyEncoder;
/**
@@ -56,4 +57,6 @@
BodyEncoder getBodyEncoder() throws HornetQException;
InputStream getBodyInputStream();
+
+ void setAddressTransient(SimpleString address);
}
Modified: trunk/src/main/org/hornetq/core/server/ServerMessage.java
===================================================================
--- trunk/src/main/org/hornetq/core/server/ServerMessage.java 2010-03-30 09:11:31 UTC (rev
9012)
+++ trunk/src/main/org/hornetq/core/server/ServerMessage.java 2010-03-30 09:18:21 UTC (rev
9013)
@@ -13,6 +13,7 @@
package org.hornetq.core.server;
+import org.hornetq.api.core.SimpleString;
import org.hornetq.core.journal.EncodingSupport;
import org.hornetq.core.message.impl.MessageInternal;
import org.hornetq.core.paging.PagingStore;
Modified: trunk/src/main/org/hornetq/core/server/impl/ServerMessageImpl.java
===================================================================
--- trunk/src/main/org/hornetq/core/server/impl/ServerMessageImpl.java 2010-03-30 09:11:31
UTC (rev 9012)
+++ trunk/src/main/org/hornetq/core/server/impl/ServerMessageImpl.java 2010-03-30 09:18:21
UTC (rev 9013)
@@ -316,5 +316,5 @@
buffer.setLong(buffer.getInt(MessageImpl.BUFFER_HEADER_SPACE) +
DataConstants.SIZE_INT, messageID);
}
-
+
}
Modified: trunk/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java
===================================================================
--- trunk/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java 2010-03-30 09:11:31
UTC (rev 9012)
+++ trunk/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java 2010-03-30 09:18:21
UTC (rev 9013)
@@ -131,6 +131,8 @@
private final RoutingContext routingContext = new RoutingContextImpl(null);
private final SessionCallback callback;
+
+ private volatile SimpleString defaultAddress;
// Constructors
---------------------------------------------------------------------------------
@@ -905,8 +907,6 @@
currentLargeMessage = msg;
}
- private volatile SimpleString defaultAddress;
-
public void send(final ServerMessage message) throws Exception
{
long id = storageManager.generateUniqueID();
@@ -916,7 +916,17 @@
if (message.getAddress() == null)
{
- message.setAddress(defaultAddress);
+ if (message.isDurable())
+ {
+ //We need to force a re-encode when the message gets persisted or when it
gets reloaded
+ //it will have no address
+ message.setAddress(defaultAddress);
+ }
+ else
+ {
+ //We don't want to force a re-encode when the message gets sent to the
consumer
+ message.setAddressTransient(defaultAddress);
+ }
}
if (message.getAddress().equals(managementAddress))
Modified: trunk/src/main/org/hornetq/jms/client/HornetQMessageProducer.java
===================================================================
--- trunk/src/main/org/hornetq/jms/client/HornetQMessageProducer.java 2010-03-30 09:11:31
UTC (rev 9012)
+++ trunk/src/main/org/hornetq/jms/client/HornetQMessageProducer.java 2010-03-30 09:18:21
UTC (rev 9013)
@@ -413,6 +413,36 @@
msg.resetMessageID(null);
}
+
+// if (!disableMessageID)
+// {
+// // Generate an id
+//
+// UUID uid = UUIDGenerator.getInstance().generateUUID();
+//
+// byte[] bytes = uid.asBytes();
+//
+// long id1 = bytes[0] << 56 | bytes[1] << 48 |
+// bytes[2] << 40 |
+// bytes[3] << 32 |
+// bytes[4] << 24 |
+// bytes[5] << 16 |
+// bytes[6] << 8 |
+// bytes[7];
+//
+// long id2 = bytes[8] << 56 | bytes[9] << 48 |
+// bytes[10] << 40 |
+// bytes[11] << 32 |
+// bytes[12] << 24 |
+// bytes[13] << 16 |
+// bytes[14] << 8 |
+// bytes[15];
+//
+// //We store it as two longs in the message, as it's a more compact format
+// msg.getCoreMessage().setClientMessageID(id1, id2);
+//
+// msg.resetMessageID(null);
+// }
if (foreign)
{
Modified:
trunk/tests/src/org/hornetq/tests/unit/core/postoffice/impl/BindingsImplTest.java
===================================================================
---
trunk/tests/src/org/hornetq/tests/unit/core/postoffice/impl/BindingsImplTest.java 2010-03-30
09:11:31 UTC (rev 9012)
+++
trunk/tests/src/org/hornetq/tests/unit/core/postoffice/impl/BindingsImplTest.java 2010-03-30
09:18:21 UTC (rev 9013)
@@ -880,6 +880,12 @@
}
+ public void setAddressTransient(SimpleString address)
+ {
+ // TODO Auto-generated method stub
+
+ }
+
}
class FakeFilter implements Filter