December 2010 - hornetq-commits - Jboss List Archives

JBoss hornetq SVN: r10011 - in trunk/examples/jms: non-transaction-failover and 3 other directories.

by do-not-reply＠jboss.org

Author: ataylor Date: 2010-12-08 02:57:53 -0500 (Wed, 08 Dec 2010) New Revision: 10011 Added: trunk/examples/jms/failover-manual-stop/ Modified: trunk/examples/jms/non-transaction-failover/build.xml trunk/examples/jms/non-transaction-failover/server0/hornetq-configuration.xml trunk/examples/jms/non-transaction-failover/server0/hornetq-jms.xml trunk/examples/jms/non-transaction-failover/server1/hornetq-configuration.xml trunk/examples/jms/non-transaction-failover/server1/hornetq-jms.xml trunk/examples/jms/non-transaction-failover/src/org/hornetq/jms/example/NonTransactionFailoverExample.java Log: fixed non transaction failover test Modified: trunk/examples/jms/non-transaction-failover/build.xml =================================================================== --- trunk/examples/jms/non-transaction-failover/build.xml 2010-12-08 06:53:58 UTC (rev 10010) +++ trunk/examples/jms/non-transaction-failover/build.xml 2010-12-08 07:57:53 UTC (rev 10011) @@ -34,7 +34,7 @@ </target> <target name="delete-files" depends="clean"> - <delete file="./server1/KILL_ME"/> + <delete file="./server0/KILL_ME"/> </target> </project> Modified: trunk/examples/jms/non-transaction-failover/server0/hornetq-configuration.xml =================================================================== --- trunk/examples/jms/non-transaction-failover/server0/hornetq-configuration.xml 2010-12-08 06:53:58 UTC (rev 10010) +++ trunk/examples/jms/non-transaction-failover/server0/hornetq-configuration.xml 2010-12-08 07:57:53 UTC (rev 10011) @@ -2,7 +2,13 @@ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="urn:hornetq /schema/hornetq-configuration.xsd"> - <live-connector-ref connector-name="live-connector"/> + <clustered>true</clustered> + +  + <failover-on-shutdown>false</failover-on-shutdown> + + <shared-store>true</shared-store> +  <connectors> @@ -10,10 +16,6 @@ <factory-class>org.hornetq.core.remoting.impl.netty.NettyConnectorFactory</factory-class> <param key="port" value="5445"/> </connector> - <connector name="live-connector"> - <factory-class>org.hornetq.core.remoting.impl.netty.NettyConnectorFactory</factory-class> - <param key="port" value="5446"/> - </connector> </connectors>  @@ -23,7 +25,31 @@ <param key="port" value="5445"/> </acceptor> </acceptors> - + + <broadcast-groups> + <broadcast-group name="bg-group1"> + <group-address>231.7.7.7</group-address> + <group-port>9876</group-port> + <broadcast-period>1000</broadcast-period> + <connector-ref>netty-connector</connector-ref> + </broadcast-group> + </broadcast-groups> + + <discovery-groups> + <discovery-group name="dg-group1"> + <group-address>231.7.7.7</group-address> + <group-port>9876</group-port> + <refresh-timeout>60000</refresh-timeout> + </discovery-group> + </discovery-groups> + + <cluster-connections> + <cluster-connection name="my-cluster"> + <address>jms</address> + <connector-ref>netty-connector</connector-ref> + <discovery-group-ref discovery-group-name="dg-group1"/> + </cluster-connection> + </cluster-connections>  <security-settings> Modified: trunk/examples/jms/non-transaction-failover/server0/hornetq-jms.xml =================================================================== --- trunk/examples/jms/non-transaction-failover/server0/hornetq-jms.xml 2010-12-08 06:53:58 UTC (rev 10010) +++ trunk/examples/jms/non-transaction-failover/server0/hornetq-jms.xml 2010-12-08 07:57:53 UTC (rev 10011) @@ -10,6 +10,18 @@ <entries> <entry name="ConnectionFactory"/> </entries> + + <ha>true</ha> +  + <retry-interval>1000</retry-interval> + +  + <retry-interval-multiplier>1.0</retry-interval-multiplier> + +  + <reconnect-attempts>-1</reconnect-attempts> </connection-factory>  Modified: trunk/examples/jms/non-transaction-failover/server1/hornetq-configuration.xml =================================================================== --- trunk/examples/jms/non-transaction-failover/server1/hornetq-configuration.xml 2010-12-08 06:53:58 UTC (rev 10010) +++ trunk/examples/jms/non-transaction-failover/server1/hornetq-configuration.xml 2010-12-08 07:57:53 UTC (rev 10011) @@ -1,6 +1,16 @@ <configuration xmlns="urn:hornetq" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="urn:hornetq ../../../src/schema/hornetq-configuration.xsd"> + + <clustered>true</clustered> + + <backup>true</backup> + +  + <failover-on-shutdown>false</failover-on-shutdown> + + <shared-store>true</shared-store> +  <connectors> @@ -17,7 +27,37 @@ <param key="port" value="5446"/> </acceptor> </acceptors> - + + <broadcast-groups> + <broadcast-group name="bg-group1"> + <group-address>231.7.7.7</group-address> + <group-port>9876</group-port> + <broadcast-period>1000</broadcast-period> + <connector-ref>netty-connector</connector-ref> + </broadcast-group> + </broadcast-groups> + + <discovery-groups> + <discovery-group name="dg-group1"> + <group-address>231.7.7.7</group-address> + <group-port>9876</group-port> + <refresh-timeout>60000</refresh-timeout> + </discovery-group> + </discovery-groups> + + <cluster-connections> + <cluster-connection name="my-cluster"> + <address>jms</address> + <connector-ref>netty-connector</connector-ref> + <discovery-group-ref discovery-group-name="dg-group1"/> + </cluster-connection> + </cluster-connections> + + <large-messages-directory>../server0/data/large-messages</large-messages-directory> + <bindings-directory>../server0/data/bindings</bindings-directory> + <journal-directory>../server0/data/journal</journal-directory> + <paging-directory>../server0/data/paging</paging-directory> +  <security-settings> Modified: trunk/examples/jms/non-transaction-failover/server1/hornetq-jms.xml =================================================================== --- trunk/examples/jms/non-transaction-failover/server1/hornetq-jms.xml 2010-12-08 06:53:58 UTC (rev 10010) +++ trunk/examples/jms/non-transaction-failover/server1/hornetq-jms.xml 2010-12-08 07:57:53 UTC (rev 10011) @@ -10,6 +10,18 @@ <entries> <entry name="ConnectionFactory"/> </entries> + + <ha>true</ha> +  + <retry-interval>1000</retry-interval> + +  + <retry-interval-multiplier>1.0</retry-interval-multiplier> + +  + <reconnect-attempts>-1</reconnect-attempts> </connection-factory>  Modified: trunk/examples/jms/non-transaction-failover/src/org/hornetq/jms/example/NonTransactionFailoverExample.java =================================================================== --- trunk/examples/jms/non-transaction-failover/src/org/hornetq/jms/example/NonTransactionFailoverExample.java 2010-12-08 06:53:58 UTC (rev 10010) +++ trunk/examples/jms/non-transaction-failover/src/org/hornetq/jms/example/NonTransactionFailoverExample.java 2010-12-08 07:57:53 UTC (rev 10011) @@ -49,7 +49,7 @@ try { // Step 1. Get an initial context for looking up JNDI from the server #1 - initialContext = getContext(1); + initialContext = getContext(0); // Step 2. Look up the JMS resources from JNDI Queue queue = (Queue)initialContext.lookup("/queue/exampleQueue"); @@ -94,8 +94,8 @@ // Step 10. Crash server #1, the live server, and wait a little while to make sure // it has really crashed - killServer(1); - Thread.sleep(5000); + Thread.sleep(2000); + killServer(0); // Step 11. Acknowledging the 2nd half of the sent messages will fail as failover to the // backup server has occured

15 years, 5 months

1
0
0 / 0

JBoss hornetq SVN: r10010 - in trunk: examples/jms/multiple-failover/server1 and 18 other directories.

by do-not-reply＠jboss.org

Author: ataylor Date: 2010-12-08 01:53:58 -0500 (Wed, 08 Dec 2010) New Revision: 10010 Modified: trunk/examples/jms/multiple-failover-failback/server0/hornetq-configuration.xml trunk/examples/jms/multiple-failover-failback/server1/hornetq-configuration.xml trunk/examples/jms/multiple-failover-failback/server2/hornetq-configuration.xml trunk/examples/jms/multiple-failover/server0/hornetq-configuration.xml trunk/examples/jms/multiple-failover/server1/hornetq-configuration.xml trunk/examples/jms/multiple-failover/server2/hornetq-configuration.xml trunk/src/config/common/schema/hornetq-configuration.xsd trunk/src/config/trunk/clustered/hornetq-configuration.xml trunk/src/main/org/hornetq/api/core/management/HornetQServerControl.java trunk/src/main/org/hornetq/core/config/Configuration.java trunk/src/main/org/hornetq/core/config/impl/ConfigurationImpl.java trunk/src/main/org/hornetq/core/deployers/impl/FileConfigurationParser.java trunk/src/main/org/hornetq/core/management/impl/HornetQServerControlImpl.java trunk/src/main/org/hornetq/core/server/HornetQServer.java trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java trunk/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeReconnectTest.java trunk/tests/src/org/hornetq/tests/integration/cluster/failover/ClusterWithBackupFailoverTestBase.java trunk/tests/src/org/hornetq/tests/integration/cluster/failover/GroupingFailoverTestBase.java trunk/tests/src/org/hornetq/tests/integration/cluster/util/SameProcessHornetQServer.java trunk/tests/src/org/hornetq/tests/integration/jms/server/management/JMSUtil.java trunk/tests/src/org/hornetq/tests/integration/management/HornetQServerControlUsingCoreTest.java Log: added failoveronshutdown flag to configuration Modified: trunk/examples/jms/multiple-failover/server0/hornetq-configuration.xml =================================================================== --- trunk/examples/jms/multiple-failover/server0/hornetq-configuration.xml 2010-12-07 22:24:31 UTC (rev 10009) +++ trunk/examples/jms/multiple-failover/server0/hornetq-configuration.xml 2010-12-08 06:53:58 UTC (rev 10010) @@ -4,6 +4,9 @@ <clustered>true</clustered> +  + <failover-on-shutdown>false</failover-on-shutdown> + <shared-store>true</shared-store>  Modified: trunk/examples/jms/multiple-failover/server1/hornetq-configuration.xml =================================================================== --- trunk/examples/jms/multiple-failover/server1/hornetq-configuration.xml 2010-12-07 22:24:31 UTC (rev 10009) +++ trunk/examples/jms/multiple-failover/server1/hornetq-configuration.xml 2010-12-08 06:53:58 UTC (rev 10010) @@ -4,6 +4,9 @@ <clustered>true</clustered> +  + <failover-on-shutdown>false</failover-on-shutdown> + <backup>true</backup> <shared-store>true</shared-store> Modified: trunk/examples/jms/multiple-failover/server2/hornetq-configuration.xml =================================================================== --- trunk/examples/jms/multiple-failover/server2/hornetq-configuration.xml 2010-12-07 22:24:31 UTC (rev 10009) +++ trunk/examples/jms/multiple-failover/server2/hornetq-configuration.xml 2010-12-08 06:53:58 UTC (rev 10010) @@ -20,6 +20,9 @@ <backup>true</backup> +  + <failover-on-shutdown>false</failover-on-shutdown> + <shared-store>true</shared-store> <connectors> Modified: trunk/examples/jms/multiple-failover-failback/server0/hornetq-configuration.xml =================================================================== --- trunk/examples/jms/multiple-failover-failback/server0/hornetq-configuration.xml 2010-12-07 22:24:31 UTC (rev 10009) +++ trunk/examples/jms/multiple-failover-failback/server0/hornetq-configuration.xml 2010-12-08 06:53:58 UTC (rev 10010) @@ -19,6 +19,9 @@ <shared-store>true</shared-store> +  + <failover-on-shutdown>false</failover-on-shutdown> + <allow-failback>true</allow-failback>  Modified: trunk/examples/jms/multiple-failover-failback/server1/hornetq-configuration.xml =================================================================== --- trunk/examples/jms/multiple-failover-failback/server1/hornetq-configuration.xml 2010-12-07 22:24:31 UTC (rev 10009) +++ trunk/examples/jms/multiple-failover-failback/server1/hornetq-configuration.xml 2010-12-08 06:53:58 UTC (rev 10010) @@ -19,6 +19,9 @@ <backup>true</backup> +  + <failover-on-shutdown>false</failover-on-shutdown> + <shared-store>true</shared-store> <allow-failback>true</allow-failback> Modified: trunk/examples/jms/multiple-failover-failback/server2/hornetq-configuration.xml =================================================================== --- trunk/examples/jms/multiple-failover-failback/server2/hornetq-configuration.xml 2010-12-07 22:24:31 UTC (rev 10009) +++ trunk/examples/jms/multiple-failover-failback/server2/hornetq-configuration.xml 2010-12-08 06:53:58 UTC (rev 10010) @@ -20,6 +20,9 @@ <backup>true</backup> +  + <failover-on-shutdown>false</failover-on-shutdown> + <shared-store>true</shared-store> <allow-failback>true</allow-failback> Modified: trunk/src/config/common/schema/hornetq-configuration.xsd =================================================================== --- trunk/src/config/common/schema/hornetq-configuration.xsd 2010-12-07 22:24:31 UTC (rev 10009) +++ trunk/src/config/common/schema/hornetq-configuration.xsd 2010-12-08 06:53:58 UTC (rev 10010) @@ -69,6 +69,8 @@ </xsd:element> <xsd:element maxOccurs="1" minOccurs="0" name="allow-failback" type="xsd:boolean"> </xsd:element> + <xsd:element maxOccurs="1" minOccurs="0" name="failover-on-shutdown" type="xsd:boolean"> + </xsd:element> <xsd:element maxOccurs="1" minOccurs="0" name="shared-store" type="xsd:boolean"> </xsd:element> <xsd:element maxOccurs="1" minOccurs="0" name="persist-delivery-count-before-delivery" type="xsd:boolean"> Modified: trunk/src/config/trunk/clustered/hornetq-configuration.xml =================================================================== --- trunk/src/config/trunk/clustered/hornetq-configuration.xml 2010-12-07 22:24:31 UTC (rev 10009) +++ trunk/src/config/trunk/clustered/hornetq-configuration.xml 2010-12-08 06:53:58 UTC (rev 10010) @@ -4,6 +4,8 @@ <clustered>true</clustered> + <failover-on-shutdown>false</failover-on-shutdown> + <shared-store>true</shared-store> <journal-min-files>10</journal-min-files> Modified: trunk/src/main/org/hornetq/api/core/management/HornetQServerControl.java =================================================================== --- trunk/src/main/org/hornetq/api/core/management/HornetQServerControl.java 2010-12-07 22:24:31 UTC (rev 10009) +++ trunk/src/main/org/hornetq/api/core/management/HornetQServerControl.java 2010-12-08 06:53:58 UTC (rev 10010) @@ -589,4 +589,7 @@ @Operation(desc= "Destroy a bridge", impact = MBeanOperationInfo.ACTION) void destroyBridge(@Parameter(name="name", desc="Name of the bridge") String name) throws Exception; + + @Operation(desc = "force the server to stop and notify clients to failover", impact = MBeanOperationInfo.UNKNOWN) + void forceFailover() throws Exception; } Modified: trunk/src/main/org/hornetq/core/config/Configuration.java =================================================================== --- trunk/src/main/org/hornetq/core/config/Configuration.java 2010-12-07 22:24:31 UTC (rev 10009) +++ trunk/src/main/org/hornetq/core/config/Configuration.java 2010-12-08 06:53:58 UTC (rev 10010) @@ -400,6 +400,17 @@ String getClusterPassword(); /** + * should we notify any clients on close that they should failover + * @return true if clients should failover + */ + boolean isFailoverOnServerShutdown(); + + /* + * set to allow clients to failover on server shutdown + * */ + void setFailoverOnServerShutdown(boolean failoverOnServerShutdown); + + /** * Sets the cluster password for this server. */ void setClusterPassword(String password); Modified: trunk/src/main/org/hornetq/core/config/impl/ConfigurationImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/config/impl/ConfigurationImpl.java 2010-12-07 22:24:31 UTC (rev 10009) +++ trunk/src/main/org/hornetq/core/config/impl/ConfigurationImpl.java 2010-12-08 06:53:58 UTC (rev 10010) @@ -174,6 +174,8 @@ public static final long DEFAULT_SERVER_DUMP_INTERVAL = -1; + private static final boolean DEFAULT_FAILOVER_ON_SERVER_SHUTDOWN = false; + public static final int DEFAULT_MEMORY_WARNING_THRESHOLD = 25; public static final long DEFAULT_MEMORY_MEASURE_INTERVAL = -1; // in milliseconds @@ -316,6 +318,8 @@ protected long serverDumpInterval = ConfigurationImpl.DEFAULT_SERVER_DUMP_INTERVAL; + protected boolean failoverOnServerShutdown = ConfigurationImpl.DEFAULT_FAILOVER_ON_SERVER_SHUTDOWN; + // percentage of free memory which triggers warning from the memory manager protected int memoryWarningThreshold = ConfigurationImpl.DEFAULT_MEMORY_WARNING_THRESHOLD; @@ -879,6 +883,16 @@ return clusterPassword; } + public boolean isFailoverOnServerShutdown() + { + return failoverOnServerShutdown; + } + + public void setFailoverOnServerShutdown(boolean failoverOnServerShutdown) + { + this.failoverOnServerShutdown = failoverOnServerShutdown; + } + public void setClusterPassword(final String theclusterPassword) { clusterPassword = theclusterPassword; @@ -1189,6 +1203,10 @@ { return false; } + if (failoverOnServerShutdown != other.isFailoverOnServerShutdown()) + { + return false; + } if (clusterPassword == null) { if (other.clusterPassword != null) Modified: trunk/src/main/org/hornetq/core/deployers/impl/FileConfigurationParser.java =================================================================== --- trunk/src/main/org/hornetq/core/deployers/impl/FileConfigurationParser.java 2010-12-07 22:24:31 UTC (rev 10009) +++ trunk/src/main/org/hornetq/core/deployers/impl/FileConfigurationParser.java 2010-12-08 06:53:58 UTC (rev 10010) @@ -164,6 +164,8 @@ config.setClustered(XMLConfigurationUtil.getBoolean(e, "clustered", config.isClustered())); config.setAllowAutoFailBack(XMLConfigurationUtil.getBoolean(e, "allow-failback", config.isClustered())); + + config.setFailoverOnServerShutdown(XMLConfigurationUtil.getBoolean(e, "failover-on-shutdown", config.isFailoverOnServerShutdown())); config.setBackup(XMLConfigurationUtil.getBoolean(e, "backup", config.isBackup())); Modified: trunk/src/main/org/hornetq/core/management/impl/HornetQServerControlImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/management/impl/HornetQServerControlImpl.java 2010-12-07 22:24:31 UTC (rev 10009) +++ trunk/src/main/org/hornetq/core/management/impl/HornetQServerControlImpl.java 2010-12-08 06:53:58 UTC (rev 10010) @@ -1715,6 +1715,21 @@ } } + public void forceFailover() throws Exception + { + checkStarted(); + + clearIO(); + + try + { + server.stop(true); + } + finally + { + blockOnIO(); + } + } // NotificationEmitter implementation ---------------------------- public void removeNotificationListener(final NotificationListener listener, Modified: trunk/src/main/org/hornetq/core/server/HornetQServer.java =================================================================== --- trunk/src/main/org/hornetq/core/server/HornetQServer.java 2010-12-07 22:24:31 UTC (rev 10009) +++ trunk/src/main/org/hornetq/core/server/HornetQServer.java 2010-12-08 06:53:58 UTC (rev 10010) @@ -162,5 +162,5 @@ ServerSession getSessionByID(String sessionID); - void kill() throws Exception; + void stop(boolean failoverOnServerShutdown) throws Exception; } Modified: trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2010-12-07 22:24:31 UTC (rev 10009) +++ trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2010-12-08 06:53:58 UTC (rev 10010) @@ -581,14 +581,9 @@ super.finalize(); } - public void kill() throws Exception - { - stop(true); - } - public void stop() throws Exception { - stop(false); + stop(configuration.isFailoverOnServerShutdown()); } public void stop(boolean failoverOnServerShutdown) throws Exception Modified: trunk/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeReconnectTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeReconnectTest.java 2010-12-07 22:24:31 UTC (rev 10009) +++ trunk/tests/src/org/hornetq/tests/integration/cluster/bridge/BridgeReconnectTest.java 2010-12-08 06:53:58 UTC (rev 10010) @@ -155,10 +155,7 @@ BridgeReconnectTest.log.info("** failing connection"); // Now we will simulate a failure of the bridge connection between server0 and server1 - /*Bridge bridge = server0.getClusterManager().getBridges().get(bridgeName); - RemotingConnection forwardingConnection = getForwardingConnection(bridge); - forwardingConnection.fail(new HornetQException(HornetQException.NOT_CONNECTED));*/ - server0.kill(); + server0.stop(true); waitForServerStart(service2); @@ -300,7 +297,7 @@ server1.start(); server0.start(); // Now we will simulate a failure of the bridge connection between server0 and server1 - server0.kill(); + server0.stop(true); locator = HornetQClient.createServerLocatorWithHA(server2tc); Modified: trunk/tests/src/org/hornetq/tests/integration/cluster/failover/ClusterWithBackupFailoverTestBase.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/cluster/failover/ClusterWithBackupFailoverTestBase.java 2010-12-07 22:24:31 UTC (rev 10009) +++ trunk/tests/src/org/hornetq/tests/integration/cluster/failover/ClusterWithBackupFailoverTestBase.java 2010-12-08 06:53:58 UTC (rev 10010) @@ -348,9 +348,8 @@ ClusterManagerImpl clusterManager = (ClusterManagerImpl) server.getClusterManager(); clusterManager.clear(); - //FailoverManagerImpl.failAllConnectionsForConnector(serverTC); - server.kill(); + server.stop(true); } public void testFailAllNodes() throws Exception Modified: trunk/tests/src/org/hornetq/tests/integration/cluster/failover/GroupingFailoverTestBase.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/cluster/failover/GroupingFailoverTestBase.java 2010-12-07 22:24:31 UTC (rev 10009) +++ trunk/tests/src/org/hornetq/tests/integration/cluster/failover/GroupingFailoverTestBase.java 2010-12-08 06:53:58 UTC (rev 10010) @@ -87,7 +87,7 @@ closeSessionFactory(0); - servers[0].kill(); + servers[0].stop(true); waitForServerRestart(2); @@ -176,7 +176,7 @@ closeSessionFactory(0); - servers[0].kill(); + servers[0].stop(true); waitForServerRestart(2); Modified: trunk/tests/src/org/hornetq/tests/integration/cluster/util/SameProcessHornetQServer.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/cluster/util/SameProcessHornetQServer.java 2010-12-07 22:24:31 UTC (rev 10009) +++ trunk/tests/src/org/hornetq/tests/integration/cluster/util/SameProcessHornetQServer.java 2010-12-08 06:53:58 UTC (rev 10010) @@ -97,16 +97,10 @@ { session.addFailureListener(new MyListener()); } - /*Set<RemotingConnection> connections = server.getRemotingService().getConnections(); - for (RemotingConnection remotingConnection : connections) - { - remotingConnection.destroy(); - server.getRemotingService().removeConnection(remotingConnection.getID()); - }*/ ClusterManagerImpl clusterManager = (ClusterManagerImpl) server.getClusterManager(); clusterManager.clear(); - server.kill(); + server.stop(true); // Wait to be informed of failure Modified: trunk/tests/src/org/hornetq/tests/integration/jms/server/management/JMSUtil.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/jms/server/management/JMSUtil.java 2010-12-07 22:24:31 UTC (rev 10009) +++ trunk/tests/src/org/hornetq/tests/integration/jms/server/management/JMSUtil.java 2010-12-08 06:53:58 UTC (rev 10010) @@ -235,16 +235,10 @@ { session.addFailureListener(new MyListener()); } - /*Set<RemotingConnection> connections = server.getRemotingService().getConnections(); - for (RemotingConnection remotingConnection : connections) - { - remotingConnection.destroy(); - server.getRemotingService().removeConnection(remotingConnection.getID()); - }*/ ClusterManagerImpl clusterManager = (ClusterManagerImpl)server.getClusterManager(); clusterManager.clear(); - server.kill(); + server.stop(true); // Wait to be informed of failure boolean ok = latch.await(10000, TimeUnit.MILLISECONDS); Modified: trunk/tests/src/org/hornetq/tests/integration/management/HornetQServerControlUsingCoreTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/management/HornetQServerControlUsingCoreTest.java 2010-12-07 22:24:31 UTC (rev 10009) +++ trunk/tests/src/org/hornetq/tests/integration/management/HornetQServerControlUsingCoreTest.java 2010-12-08 06:53:58 UTC (rev 10010) @@ -562,6 +562,11 @@ } + public void forceFailover() throws Exception + { + proxy.invokeOperation("forceFailover"); + } + public String getLiveConnectorName() throws Exception { return (String)proxy.retrieveAttributeValue("liveConnectorName");

15 years, 5 months

1
0
0 / 0

JBoss hornetq SVN: r10009 - trunk/src/main/org/hornetq/core/client/impl.

by do-not-reply＠jboss.org

Author: clebert.suconic(a)jboss.com Date: 2010-12-07 17:24:31 -0500 (Tue, 07 Dec 2010) New Revision: 10009 Added: trunk/src/main/org/hornetq/core/client/impl/CompressedLargeMessageControllerImpl.java Removed: trunk/src/main/org/hornetq/core/client/impl/DecompressedLargeMessageControllerImpl.java Modified: trunk/src/main/org/hornetq/core/client/impl/ClientConsumerImpl.java Log: Simple rename Modified: trunk/src/main/org/hornetq/core/client/impl/ClientConsumerImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/ClientConsumerImpl.java 2010-12-07 22:22:35 UTC (rev 10008) +++ trunk/src/main/org/hornetq/core/client/impl/ClientConsumerImpl.java 2010-12-07 22:24:31 UTC (rev 10009) @@ -557,7 +557,7 @@ if (currentChunkMessage.isCompressed()) { - currentChunkMessage.setBuffer(new DecompressedLargeMessageControllerImpl(currentLargeMessageController)); + currentChunkMessage.setBuffer(new CompressedLargeMessageControllerImpl(currentLargeMessageController)); } else { Copied: trunk/src/main/org/hornetq/core/client/impl/CompressedLargeMessageControllerImpl.java (from rev 10008, trunk/src/main/org/hornetq/core/client/impl/DecompressedLargeMessageControllerImpl.java) =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/CompressedLargeMessageControllerImpl.java (rev 0) +++ trunk/src/main/org/hornetq/core/client/impl/CompressedLargeMessageControllerImpl.java 2010-12-07 22:24:31 UTC (rev 10009) @@ -0,0 +1,1094 @@ +/* + * Copyright 2010 Red Hat, Inc. + * Red Hat 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.hornetq.core.client.impl; + +import java.io.DataInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.nio.ByteBuffer; +import java.nio.channels.GatheringByteChannel; +import java.nio.channels.ScatteringByteChannel; + +import org.hornetq.api.core.HornetQBuffer; +import org.hornetq.api.core.HornetQBuffers; +import org.hornetq.api.core.HornetQException; +import org.hornetq.api.core.SimpleString; +import org.hornetq.core.protocol.core.impl.wireformat.SessionReceiveContinuationMessage; +import org.hornetq.utils.DataConstants; +import org.hornetq.utils.HornetQBufferInputStream; +import org.hornetq.utils.InflaterReader; +import org.hornetq.utils.InflaterWriter; +import org.hornetq.utils.UTF8Util; +import org.jboss.netty.buffer.ChannelBuffer; + +/** + * A DecompressedHornetQBuffer + * + * @author <a href="mailto:clebert.suconic@jboss.org">Clebert Suconic</a> + * + * + */ +public class CompressedLargeMessageControllerImpl implements LargeMessageController +{ + + // Constants ----------------------------------------------------- + + private static final String OPERATION_NOT_SUPPORTED = "Operation not supported"; + + private static final String READ_ONLY_ERROR_MESSAGE = "This is a read-only buffer, setOperations are not supported"; + + // Attributes ---------------------------------------------------- + + final LargeMessageController bufferDelegate; + + // Static -------------------------------------------------------- + + // Constructors -------------------------------------------------- + + public CompressedLargeMessageControllerImpl(final LargeMessageController bufferDelegate) + { + this.bufferDelegate = bufferDelegate; + } + + + // Public -------------------------------------------------------- + + /** + * + */ + public void discardUnusedPackets() + { + bufferDelegate.discardUnusedPackets(); + } + + /** + * Add a buff to the List, or save it to the OutputStream if set + * @param packet + */ + public void addPacket(final SessionReceiveContinuationMessage packet) + { + bufferDelegate.addPacket(packet); + } + + public synchronized void cancel() + { + bufferDelegate.cancel(); + } + + public synchronized void close() + { + bufferDelegate.cancel(); + } + + public void setOutputStream(final OutputStream output) throws HornetQException + { + bufferDelegate.setOutputStream(new InflaterWriter(output)); + } + + public synchronized void saveBuffer(final OutputStream output) throws HornetQException + { + setOutputStream(output); + waitCompletion(0); + } + + /** + * + * @param timeWait Milliseconds to Wait. 0 means forever + * @throws Exception + */ + public synchronized boolean waitCompletion(final long timeWait) throws HornetQException + { + return bufferDelegate.waitCompletion(timeWait); + } + + // Channel Buffer Implementation --------------------------------- + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#array() + */ + public byte[] array() + { + throw new IllegalAccessError("array not supported on LargeMessageBufferImpl"); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#capacity() + */ + public int capacity() + { + return -1; + } + + DataInputStream dataInput = null; + + private DataInputStream getStream() + { + if (dataInput == null) + { + try + { + InputStream input = new HornetQBufferInputStream(bufferDelegate); + + dataInput = new DataInputStream(new InflaterReader(input)); + } + catch (Exception e) + { + throw new RuntimeException (e.getMessage(), e); + } + + } + return dataInput; + } + + private void positioningNotSupported() + { + throw new IllegalStateException("Position not supported over compressed large messages"); + } + + public byte readByte() + { + try + { + return getStream().readByte(); + } + catch (Exception e) + { + throw new RuntimeException (e.getMessage(), e); + } + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#getByte(int) + */ + public byte getByte(final int index) + { + positioningNotSupported(); + return 0; + } + + private byte getByte(final long index) + { + positioningNotSupported(); + return 0; + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, org.hornetq.api.core.buffers.ChannelBuffer, int, int) + */ + public void getBytes(final int index, final HornetQBuffer dst, final int dstIndex, final int length) + { + positioningNotSupported(); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, org.hornetq.api.core.buffers.ChannelBuffer, int, int) + */ + public void getBytes(final long index, final HornetQBuffer dst, final int dstIndex, final int length) + { + positioningNotSupported(); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, byte[], int, int) + */ + public void getBytes(final int index, final byte[] dst, final int dstIndex, final int length) + { + positioningNotSupported(); + } + + public void getBytes(final long index, final byte[] dst, final int dstIndex, final int length) + { + positioningNotSupported(); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, java.nio.ByteBuffer) + */ + public void getBytes(final int index, final ByteBuffer dst) + { + positioningNotSupported(); + } + + public void getBytes(final long index, final ByteBuffer dst) + { + positioningNotSupported(); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, java.io.OutputStream, int) + */ + public void getBytes(final int index, final OutputStream out, final int length) throws IOException + { + positioningNotSupported(); + } + + public void getBytes(final long index, final OutputStream out, final int length) throws IOException + { + positioningNotSupported(); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, java.nio.channels.GatheringByteChannel, int) + */ + public int getBytes(final int index, final GatheringByteChannel out, final int length) throws IOException + { + positioningNotSupported(); + return 0; + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#getInt(int) + */ + public int getInt(final int index) + { + positioningNotSupported(); + return 0; + } + + public int getInt(final long index) + { + positioningNotSupported(); + return 0; + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#getLong(int) + */ + public long getLong(final int index) + { + positioningNotSupported(); + return 0; + } + + public long getLong(final long index) + { + positioningNotSupported(); + return 0; + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#getShort(int) + */ + public short getShort(final int index) + { + positioningNotSupported(); + return 0; + } + + public short getShort(final long index) + { + return (short)(getByte(index) << 8 | getByte(index + 1) & 0xFF); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#getUnsignedMedium(int) + */ + public int getUnsignedMedium(final int index) + { + positioningNotSupported(); + return 0; + } + + + + public int getUnsignedMedium(final long index) + { + positioningNotSupported(); + return 0; + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#setByte(int, byte) + */ + public void setByte(final int index, final byte value) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#setBytes(int, org.hornetq.api.core.buffers.ChannelBuffer, int, int) + */ + public void setBytes(final int index, final HornetQBuffer src, final int srcIndex, final int length) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#setBytes(int, byte[], int, int) + */ + public void setBytes(final int index, final byte[] src, final int srcIndex, final int length) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#setBytes(int, java.nio.ByteBuffer) + */ + public void setBytes(final int index, final ByteBuffer src) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#setBytes(int, java.io.InputStream, int) + */ + public int setBytes(final int index, final InputStream in, final int length) throws IOException + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#setBytes(int, java.nio.channels.ScatteringByteChannel, int) + */ + public int setBytes(final int index, final ScatteringByteChannel in, final int length) throws IOException + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#setInt(int, int) + */ + public void setInt(final int index, final int value) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#setLong(int, long) + */ + public void setLong(final int index, final long value) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#setMedium(int, int) + */ + public void setMedium(final int index, final int value) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#setShort(int, short) + */ + public void setShort(final int index, final short value) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#toByteBuffer(int, int) + */ + public ByteBuffer toByteBuffer(final int index, final int length) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#toString(int, int, java.lang.String) + */ + public String toString(final int index, final int length, final String charsetName) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public int readerIndex() + { + return 0; + } + + public void readerIndex(final int readerIndex) + { + // TODO + } + + public int writerIndex() + { + // TODO + return 0; + } + + public long getSize() + { + // TODO + return 0; + } + + public void writerIndex(final int writerIndex) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void setIndex(final int readerIndex, final int writerIndex) + { + positioningNotSupported(); + } + + public void clear() + { + } + + public boolean readable() + { + return true; + } + + public boolean writable() + { + return false; + } + + public int readableBytes() + { + return 1; + } + + public int writableBytes() + { + return 0; + } + + public void markReaderIndex() + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void resetReaderIndex() + { + // TODO: reset positioning if possible + } + + public void markWriterIndex() + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void resetWriterIndex() + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void discardReadBytes() + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public short getUnsignedByte(final int index) + { + return (short)(getByte(index) & 0xFF); + } + + public int getUnsignedShort(final int index) + { + return getShort(index) & 0xFFFF; + } + + public int getMedium(final int index) + { + int value = getUnsignedMedium(index); + if ((value & 0x800000) != 0) + { + value |= 0xff000000; + } + return value; + } + + public long getUnsignedInt(final int index) + { + return getInt(index) & 0xFFFFFFFFL; + } + + public void getBytes(int index, final byte[] dst) + { + // TODO: optimize this by using System.arraycopy + for (int i = 0; i < dst.length; i++) + { + dst[i] = getByte(index++); + } + } + + public void getBytes(long index, final byte[] dst) + { + // TODO: optimize this by using System.arraycopy + for (int i = 0; i < dst.length; i++) + { + dst[i] = getByte(index++); + } + } + + public void getBytes(final int index, final HornetQBuffer dst) + { + getBytes(index, dst, dst.writableBytes()); + } + + public void getBytes(final int index, final HornetQBuffer dst, final int length) + { + if (length > dst.writableBytes()) + { + throw new IndexOutOfBoundsException(); + } + getBytes(index, dst, dst.writerIndex(), length); + dst.writerIndex(dst.writerIndex() + length); + } + + public void setBytes(final int index, final byte[] src) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void setBytes(final int index, final HornetQBuffer src) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void setBytes(final int index, final HornetQBuffer src, final int length) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void setZero(final int index, final int length) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public short readUnsignedByte() + { + try + { + return (short)getStream().readUnsignedByte(); + } + catch (Exception e) + { + throw new IllegalStateException (e.getMessage(), e); + } + } + + public short readShort() + { + try + { + return (short)getStream().readShort(); + } + catch (Exception e) + { + throw new IllegalStateException (e.getMessage(), e); + } + } + + public int readUnsignedShort() + { + try + { + return (int)getStream().readUnsignedShort(); + } + catch (Exception e) + { + throw new IllegalStateException (e.getMessage(), e); + } + } + + public int readMedium() + { + int value = readUnsignedMedium(); + if ((value & 0x800000) != 0) + { + value |= 0xff000000; + } + return value; + } + + + public int readUnsignedMedium() + { + return (readByte() & 0xff) << 16 | (readByte() & 0xff) << 8 | (readByte() & 0xff) << 0; + } + + public int readInt() + { + try + { + return getStream().readInt(); + } + catch (Exception e) + { + throw new IllegalStateException(e.getMessage(), e); + } + } + + public int readInt(final int pos) + { + positioningNotSupported(); + return 0; + } + + public long readUnsignedInt() + { + return readInt() & 0xFFFFFFFFL; + } + + public long readLong() + { + try + { + return getStream().readLong(); + } + catch (Exception e) + { + throw new IllegalStateException(e.getMessage(), e); + } + } + + public void readBytes(final byte[] dst, final int dstIndex, final int length) + { + try + { + getStream().read(dst, dstIndex, length); + } + catch (Exception e) + { + throw new IllegalStateException(e.getMessage(), e); + } + } + + public void readBytes(final byte[] dst) + { + readBytes(dst, 0, dst.length); + } + + public void readBytes(final HornetQBuffer dst) + { + readBytes(dst, dst.writableBytes()); + } + + public void readBytes(final HornetQBuffer dst, final int length) + { + if (length > dst.writableBytes()) + { + throw new IndexOutOfBoundsException(); + } + readBytes(dst, dst.writerIndex(), length); + dst.writerIndex(dst.writerIndex() + length); + } + + public void readBytes(final HornetQBuffer dst, final int dstIndex, final int length) + { + byte[] destBytes = new byte[length]; + readBytes(destBytes); + dst.setBytes(dstIndex, destBytes); + } + + public void readBytes(final ByteBuffer dst) + { + byte bytesToGet[] = new byte[dst.remaining()]; + readBytes(bytesToGet); + dst.put(bytesToGet); + } + + public int readBytes(final GatheringByteChannel out, final int length) throws IOException + { + throw new IllegalStateException("Not implemented!"); + } + + public void readBytes(final OutputStream out, final int length) throws IOException + { + throw new IllegalStateException("Not implemented!"); + } + + public void skipBytes(final int length) + { + + try + { + for (int i = 0 ; i < length; i++) + { + getStream().read(); + } + } + catch (Exception e) + { + throw new IllegalStateException(e.getMessage(), e); + } + } + + public void writeByte(final byte value) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void writeShort(final short value) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void writeMedium(final int value) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void writeInt(final int value) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void writeLong(final long value) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void writeBytes(final byte[] src, final int srcIndex, final int length) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void writeBytes(final byte[] src) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void writeBytes(final HornetQBuffer src) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void writeBytes(final HornetQBuffer src, final int length) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void writeBytes(final ByteBuffer src) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public int writeBytes(final InputStream in, final int length) throws IOException + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public int writeBytes(final ScatteringByteChannel in, final int length) throws IOException + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void writeZero(final int length) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public ByteBuffer toByteBuffer() + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public ByteBuffer[] toByteBuffers() + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public ByteBuffer[] toByteBuffers(final int index, final int length) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public String toString(final String charsetName) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public Object getUnderlyingBuffer() + { + return this; + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#readBoolean() + */ + public boolean readBoolean() + { + return readByte() != 0; + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#readChar() + */ + public char readChar() + { + return (char)readShort(); + } + + public char getChar(final int index) + { + return (char)getShort(index); + } + + public double getDouble(final int index) + { + return Double.longBitsToDouble(getLong(index)); + } + + public float getFloat(final int index) + { + return Float.intBitsToFloat(getInt(index)); + } + + public HornetQBuffer readBytes(final int length) + { + byte bytesToGet[] = new byte[length]; + readBytes(bytesToGet); + return HornetQBuffers.wrappedBuffer(bytesToGet); + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#readDouble() + */ + public double readDouble() + { + return Double.longBitsToDouble(readLong()); + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#readFloat() + */ + public float readFloat() + { + return Float.intBitsToFloat(readInt()); + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#readNullableSimpleString() + */ + public SimpleString readNullableSimpleString() + { + int b = readByte(); + if (b == DataConstants.NULL) + { + return null; + } + else + { + return readSimpleString(); + } + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#readNullableString() + */ + public String readNullableString() + { + int b = readByte(); + if (b == DataConstants.NULL) + { + return null; + } + else + { + return readString(); + } + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#readSimpleString() + */ + public SimpleString readSimpleString() + { + int len = readInt(); + byte[] data = new byte[len]; + readBytes(data); + return new SimpleString(data); + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#readString() + */ + public String readString() + { + int len = readInt(); + + if (len < 9) + { + char[] chars = new char[len]; + for (int i = 0; i < len; i++) + { + chars[i] = (char)readShort(); + } + return new String(chars); + } + else if (len < 0xfff) + { + return readUTF(); + } + else + { + return readSimpleString().toString(); + } + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#readUTF() + */ + public String readUTF() + { + return UTF8Util.readUTF(this); + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeBoolean(boolean) + */ + public void writeBoolean(final boolean val) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeChar(char) + */ + public void writeChar(final char val) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeDouble(double) + */ + public void writeDouble(final double val) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeFloat(float) + */ + public void writeFloat(final float val) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeNullableSimpleString(org.hornetq.util.SimpleString) + */ + public void writeNullableSimpleString(final SimpleString val) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeNullableString(java.lang.String) + */ + public void writeNullableString(final String val) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeSimpleString(org.hornetq.util.SimpleString) + */ + public void writeSimpleString(final SimpleString val) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeString(java.lang.String) + */ + public void writeString(final String val) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeUTF(java.lang.String) + */ + public void writeUTF(final String utf) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#compareTo(org.hornetq.api.core.buffers.ChannelBuffer) + */ + public int compareTo(final HornetQBuffer buffer) + { + return -1; + } + + public HornetQBuffer copy() + { + throw new UnsupportedOperationException(); + } + + public HornetQBuffer slice(final int index, final int length) + { + throw new UnsupportedOperationException(); + } + + // Package protected --------------------------------------------- + + // Protected ----------------------------------------------------- + + // Private ------------------------------------------------------- + + /** + * @param body + */ + // Inner classes ------------------------------------------------- + + public ChannelBuffer channelBuffer() + { + return null; + } + + public HornetQBuffer copy(final int index, final int length) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public HornetQBuffer duplicate() + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public HornetQBuffer readSlice(final int length) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void setChar(final int index, final char value) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void setDouble(final int index, final double value) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void setFloat(final int index, final float value) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public HornetQBuffer slice() + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void writeBytes(final HornetQBuffer src, final int srcIndex, final int length) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } +} Deleted: trunk/src/main/org/hornetq/core/client/impl/DecompressedLargeMessageControllerImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/DecompressedLargeMessageControllerImpl.java 2010-12-07 22:22:35 UTC (rev 10008) +++ trunk/src/main/org/hornetq/core/client/impl/DecompressedLargeMessageControllerImpl.java 2010-12-07 22:24:31 UTC (rev 10009) @@ -1,1094 +0,0 @@ -/* - * Copyright 2010 Red Hat, Inc. - * Red Hat 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.hornetq.core.client.impl; - -import java.io.DataInputStream; -import java.io.IOException; -import java.io.InputStream; -import java.io.OutputStream; -import java.nio.ByteBuffer; -import java.nio.channels.GatheringByteChannel; -import java.nio.channels.ScatteringByteChannel; - -import org.hornetq.api.core.HornetQBuffer; -import org.hornetq.api.core.HornetQBuffers; -import org.hornetq.api.core.HornetQException; -import org.hornetq.api.core.SimpleString; -import org.hornetq.core.protocol.core.impl.wireformat.SessionReceiveContinuationMessage; -import org.hornetq.utils.DataConstants; -import org.hornetq.utils.HornetQBufferInputStream; -import org.hornetq.utils.InflaterReader; -import org.hornetq.utils.InflaterWriter; -import org.hornetq.utils.UTF8Util; -import org.jboss.netty.buffer.ChannelBuffer; - -/** - * A DecompressedHornetQBuffer - * - * @author <a href="mailto:clebert.suconic@jboss.org">Clebert Suconic</a> - * - * - */ -public class DecompressedLargeMessageControllerImpl implements LargeMessageController -{ - - // Constants ----------------------------------------------------- - - private static final String OPERATION_NOT_SUPPORTED = "Operation not supported"; - - private static final String READ_ONLY_ERROR_MESSAGE = "This is a read-only buffer, setOperations are not supported"; - - // Attributes ---------------------------------------------------- - - final LargeMessageController bufferDelegate; - - // Static -------------------------------------------------------- - - // Constructors -------------------------------------------------- - - public DecompressedLargeMessageControllerImpl(final LargeMessageController bufferDelegate) - { - this.bufferDelegate = bufferDelegate; - } - - - // Public -------------------------------------------------------- - - /** - * - */ - public void discardUnusedPackets() - { - bufferDelegate.discardUnusedPackets(); - } - - /** - * Add a buff to the List, or save it to the OutputStream if set - * @param packet - */ - public void addPacket(final SessionReceiveContinuationMessage packet) - { - bufferDelegate.addPacket(packet); - } - - public synchronized void cancel() - { - bufferDelegate.cancel(); - } - - public synchronized void close() - { - bufferDelegate.cancel(); - } - - public void setOutputStream(final OutputStream output) throws HornetQException - { - bufferDelegate.setOutputStream(new InflaterWriter(output)); - } - - public synchronized void saveBuffer(final OutputStream output) throws HornetQException - { - setOutputStream(output); - waitCompletion(0); - } - - /** - * - * @param timeWait Milliseconds to Wait. 0 means forever - * @throws Exception - */ - public synchronized boolean waitCompletion(final long timeWait) throws HornetQException - { - return bufferDelegate.waitCompletion(timeWait); - } - - // Channel Buffer Implementation --------------------------------- - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#array() - */ - public byte[] array() - { - throw new IllegalAccessError("array not supported on LargeMessageBufferImpl"); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#capacity() - */ - public int capacity() - { - return -1; - } - - DataInputStream dataInput = null; - - private DataInputStream getStream() - { - if (dataInput == null) - { - try - { - InputStream input = new HornetQBufferInputStream(bufferDelegate); - - dataInput = new DataInputStream(new InflaterReader(input)); - } - catch (Exception e) - { - throw new RuntimeException (e.getMessage(), e); - } - - } - return dataInput; - } - - private void positioningNotSupported() - { - throw new IllegalStateException("Position not supported over compressed large messages"); - } - - public byte readByte() - { - try - { - return getStream().readByte(); - } - catch (Exception e) - { - throw new RuntimeException (e.getMessage(), e); - } - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#getByte(int) - */ - public byte getByte(final int index) - { - positioningNotSupported(); - return 0; - } - - private byte getByte(final long index) - { - positioningNotSupported(); - return 0; - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, org.hornetq.api.core.buffers.ChannelBuffer, int, int) - */ - public void getBytes(final int index, final HornetQBuffer dst, final int dstIndex, final int length) - { - positioningNotSupported(); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, org.hornetq.api.core.buffers.ChannelBuffer, int, int) - */ - public void getBytes(final long index, final HornetQBuffer dst, final int dstIndex, final int length) - { - positioningNotSupported(); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, byte[], int, int) - */ - public void getBytes(final int index, final byte[] dst, final int dstIndex, final int length) - { - positioningNotSupported(); - } - - public void getBytes(final long index, final byte[] dst, final int dstIndex, final int length) - { - positioningNotSupported(); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, java.nio.ByteBuffer) - */ - public void getBytes(final int index, final ByteBuffer dst) - { - positioningNotSupported(); - } - - public void getBytes(final long index, final ByteBuffer dst) - { - positioningNotSupported(); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, java.io.OutputStream, int) - */ - public void getBytes(final int index, final OutputStream out, final int length) throws IOException - { - positioningNotSupported(); - } - - public void getBytes(final long index, final OutputStream out, final int length) throws IOException - { - positioningNotSupported(); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, java.nio.channels.GatheringByteChannel, int) - */ - public int getBytes(final int index, final GatheringByteChannel out, final int length) throws IOException - { - positioningNotSupported(); - return 0; - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#getInt(int) - */ - public int getInt(final int index) - { - positioningNotSupported(); - return 0; - } - - public int getInt(final long index) - { - positioningNotSupported(); - return 0; - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#getLong(int) - */ - public long getLong(final int index) - { - positioningNotSupported(); - return 0; - } - - public long getLong(final long index) - { - positioningNotSupported(); - return 0; - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#getShort(int) - */ - public short getShort(final int index) - { - positioningNotSupported(); - return 0; - } - - public short getShort(final long index) - { - return (short)(getByte(index) << 8 | getByte(index + 1) & 0xFF); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#getUnsignedMedium(int) - */ - public int getUnsignedMedium(final int index) - { - positioningNotSupported(); - return 0; - } - - - - public int getUnsignedMedium(final long index) - { - positioningNotSupported(); - return 0; - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#setByte(int, byte) - */ - public void setByte(final int index, final byte value) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#setBytes(int, org.hornetq.api.core.buffers.ChannelBuffer, int, int) - */ - public void setBytes(final int index, final HornetQBuffer src, final int srcIndex, final int length) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#setBytes(int, byte[], int, int) - */ - public void setBytes(final int index, final byte[] src, final int srcIndex, final int length) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#setBytes(int, java.nio.ByteBuffer) - */ - public void setBytes(final int index, final ByteBuffer src) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#setBytes(int, java.io.InputStream, int) - */ - public int setBytes(final int index, final InputStream in, final int length) throws IOException - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#setBytes(int, java.nio.channels.ScatteringByteChannel, int) - */ - public int setBytes(final int index, final ScatteringByteChannel in, final int length) throws IOException - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#setInt(int, int) - */ - public void setInt(final int index, final int value) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#setLong(int, long) - */ - public void setLong(final int index, final long value) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#setMedium(int, int) - */ - public void setMedium(final int index, final int value) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#setShort(int, short) - */ - public void setShort(final int index, final short value) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#toByteBuffer(int, int) - */ - public ByteBuffer toByteBuffer(final int index, final int length) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#toString(int, int, java.lang.String) - */ - public String toString(final int index, final int length, final String charsetName) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public int readerIndex() - { - return 0; - } - - public void readerIndex(final int readerIndex) - { - // TODO - } - - public int writerIndex() - { - // TODO - return 0; - } - - public long getSize() - { - // TODO - return 0; - } - - public void writerIndex(final int writerIndex) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void setIndex(final int readerIndex, final int writerIndex) - { - positioningNotSupported(); - } - - public void clear() - { - } - - public boolean readable() - { - return true; - } - - public boolean writable() - { - return false; - } - - public int readableBytes() - { - return 1; - } - - public int writableBytes() - { - return 0; - } - - public void markReaderIndex() - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void resetReaderIndex() - { - // TODO: reset positioning if possible - } - - public void markWriterIndex() - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void resetWriterIndex() - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void discardReadBytes() - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public short getUnsignedByte(final int index) - { - return (short)(getByte(index) & 0xFF); - } - - public int getUnsignedShort(final int index) - { - return getShort(index) & 0xFFFF; - } - - public int getMedium(final int index) - { - int value = getUnsignedMedium(index); - if ((value & 0x800000) != 0) - { - value |= 0xff000000; - } - return value; - } - - public long getUnsignedInt(final int index) - { - return getInt(index) & 0xFFFFFFFFL; - } - - public void getBytes(int index, final byte[] dst) - { - // TODO: optimize this by using System.arraycopy - for (int i = 0; i < dst.length; i++) - { - dst[i] = getByte(index++); - } - } - - public void getBytes(long index, final byte[] dst) - { - // TODO: optimize this by using System.arraycopy - for (int i = 0; i < dst.length; i++) - { - dst[i] = getByte(index++); - } - } - - public void getBytes(final int index, final HornetQBuffer dst) - { - getBytes(index, dst, dst.writableBytes()); - } - - public void getBytes(final int index, final HornetQBuffer dst, final int length) - { - if (length > dst.writableBytes()) - { - throw new IndexOutOfBoundsException(); - } - getBytes(index, dst, dst.writerIndex(), length); - dst.writerIndex(dst.writerIndex() + length); - } - - public void setBytes(final int index, final byte[] src) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void setBytes(final int index, final HornetQBuffer src) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void setBytes(final int index, final HornetQBuffer src, final int length) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void setZero(final int index, final int length) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public short readUnsignedByte() - { - try - { - return (short)getStream().readUnsignedByte(); - } - catch (Exception e) - { - throw new IllegalStateException (e.getMessage(), e); - } - } - - public short readShort() - { - try - { - return (short)getStream().readShort(); - } - catch (Exception e) - { - throw new IllegalStateException (e.getMessage(), e); - } - } - - public int readUnsignedShort() - { - try - { - return (int)getStream().readUnsignedShort(); - } - catch (Exception e) - { - throw new IllegalStateException (e.getMessage(), e); - } - } - - public int readMedium() - { - int value = readUnsignedMedium(); - if ((value & 0x800000) != 0) - { - value |= 0xff000000; - } - return value; - } - - - public int readUnsignedMedium() - { - return (readByte() & 0xff) << 16 | (readByte() & 0xff) << 8 | (readByte() & 0xff) << 0; - } - - public int readInt() - { - try - { - return getStream().readInt(); - } - catch (Exception e) - { - throw new IllegalStateException(e.getMessage(), e); - } - } - - public int readInt(final int pos) - { - positioningNotSupported(); - return 0; - } - - public long readUnsignedInt() - { - return readInt() & 0xFFFFFFFFL; - } - - public long readLong() - { - try - { - return getStream().readLong(); - } - catch (Exception e) - { - throw new IllegalStateException(e.getMessage(), e); - } - } - - public void readBytes(final byte[] dst, final int dstIndex, final int length) - { - try - { - getStream().read(dst, dstIndex, length); - } - catch (Exception e) - { - throw new IllegalStateException(e.getMessage(), e); - } - } - - public void readBytes(final byte[] dst) - { - readBytes(dst, 0, dst.length); - } - - public void readBytes(final HornetQBuffer dst) - { - readBytes(dst, dst.writableBytes()); - } - - public void readBytes(final HornetQBuffer dst, final int length) - { - if (length > dst.writableBytes()) - { - throw new IndexOutOfBoundsException(); - } - readBytes(dst, dst.writerIndex(), length); - dst.writerIndex(dst.writerIndex() + length); - } - - public void readBytes(final HornetQBuffer dst, final int dstIndex, final int length) - { - byte[] destBytes = new byte[length]; - readBytes(destBytes); - dst.setBytes(dstIndex, destBytes); - } - - public void readBytes(final ByteBuffer dst) - { - byte bytesToGet[] = new byte[dst.remaining()]; - readBytes(bytesToGet); - dst.put(bytesToGet); - } - - public int readBytes(final GatheringByteChannel out, final int length) throws IOException - { - throw new IllegalStateException("Not implemented!"); - } - - public void readBytes(final OutputStream out, final int length) throws IOException - { - throw new IllegalStateException("Not implemented!"); - } - - public void skipBytes(final int length) - { - - try - { - for (int i = 0 ; i < length; i++) - { - getStream().read(); - } - } - catch (Exception e) - { - throw new IllegalStateException(e.getMessage(), e); - } - } - - public void writeByte(final byte value) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void writeShort(final short value) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void writeMedium(final int value) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void writeInt(final int value) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void writeLong(final long value) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void writeBytes(final byte[] src, final int srcIndex, final int length) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void writeBytes(final byte[] src) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void writeBytes(final HornetQBuffer src) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void writeBytes(final HornetQBuffer src, final int length) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void writeBytes(final ByteBuffer src) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public int writeBytes(final InputStream in, final int length) throws IOException - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public int writeBytes(final ScatteringByteChannel in, final int length) throws IOException - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void writeZero(final int length) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public ByteBuffer toByteBuffer() - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public ByteBuffer[] toByteBuffers() - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public ByteBuffer[] toByteBuffers(final int index, final int length) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public String toString(final String charsetName) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public Object getUnderlyingBuffer() - { - return this; - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#readBoolean() - */ - public boolean readBoolean() - { - return readByte() != 0; - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#readChar() - */ - public char readChar() - { - return (char)readShort(); - } - - public char getChar(final int index) - { - return (char)getShort(index); - } - - public double getDouble(final int index) - { - return Double.longBitsToDouble(getLong(index)); - } - - public float getFloat(final int index) - { - return Float.intBitsToFloat(getInt(index)); - } - - public HornetQBuffer readBytes(final int length) - { - byte bytesToGet[] = new byte[length]; - readBytes(bytesToGet); - return HornetQBuffers.wrappedBuffer(bytesToGet); - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#readDouble() - */ - public double readDouble() - { - return Double.longBitsToDouble(readLong()); - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#readFloat() - */ - public float readFloat() - { - return Float.intBitsToFloat(readInt()); - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#readNullableSimpleString() - */ - public SimpleString readNullableSimpleString() - { - int b = readByte(); - if (b == DataConstants.NULL) - { - return null; - } - else - { - return readSimpleString(); - } - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#readNullableString() - */ - public String readNullableString() - { - int b = readByte(); - if (b == DataConstants.NULL) - { - return null; - } - else - { - return readString(); - } - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#readSimpleString() - */ - public SimpleString readSimpleString() - { - int len = readInt(); - byte[] data = new byte[len]; - readBytes(data); - return new SimpleString(data); - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#readString() - */ - public String readString() - { - int len = readInt(); - - if (len < 9) - { - char[] chars = new char[len]; - for (int i = 0; i < len; i++) - { - chars[i] = (char)readShort(); - } - return new String(chars); - } - else if (len < 0xfff) - { - return readUTF(); - } - else - { - return readSimpleString().toString(); - } - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#readUTF() - */ - public String readUTF() - { - return UTF8Util.readUTF(this); - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeBoolean(boolean) - */ - public void writeBoolean(final boolean val) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeChar(char) - */ - public void writeChar(final char val) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeDouble(double) - */ - public void writeDouble(final double val) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeFloat(float) - */ - public void writeFloat(final float val) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeNullableSimpleString(org.hornetq.util.SimpleString) - */ - public void writeNullableSimpleString(final SimpleString val) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeNullableString(java.lang.String) - */ - public void writeNullableString(final String val) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeSimpleString(org.hornetq.util.SimpleString) - */ - public void writeSimpleString(final SimpleString val) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeString(java.lang.String) - */ - public void writeString(final String val) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeUTF(java.lang.String) - */ - public void writeUTF(final String utf) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#compareTo(org.hornetq.api.core.buffers.ChannelBuffer) - */ - public int compareTo(final HornetQBuffer buffer) - { - return -1; - } - - public HornetQBuffer copy() - { - throw new UnsupportedOperationException(); - } - - public HornetQBuffer slice(final int index, final int length) - { - throw new UnsupportedOperationException(); - } - - // Package protected --------------------------------------------- - - // Protected ----------------------------------------------------- - - // Private ------------------------------------------------------- - - /** - * @param body - */ - // Inner classes ------------------------------------------------- - - public ChannelBuffer channelBuffer() - { - return null; - } - - public HornetQBuffer copy(final int index, final int length) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public HornetQBuffer duplicate() - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public HornetQBuffer readSlice(final int length) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void setChar(final int index, final char value) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void setDouble(final int index, final double value) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void setFloat(final int index, final float value) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public HornetQBuffer slice() - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void writeBytes(final HornetQBuffer src, final int srcIndex, final int length) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } -}

15 years, 5 months

1
0
0 / 0

JBoss hornetq SVN: r10008 - in trunk: src/main/org/hornetq/core/message/impl and 14 other directories.

by do-not-reply＠jboss.org

Author: clebert.suconic(a)jboss.com Date: 2010-12-07 17:22:35 -0500 (Tue, 07 Dec 2010) New Revision: 10008 Added: trunk/src/main/org/hornetq/core/client/impl/DecompressedLargeMessageControllerImpl.java trunk/src/main/org/hornetq/core/client/impl/LargeMessageController.java trunk/src/main/org/hornetq/core/client/impl/LargeMessageControllerImpl.java Removed: trunk/src/main/org/hornetq/core/client/impl/LargeMessageBufferImpl.java trunk/src/main/org/hornetq/core/client/impl/LargeMessageBufferInternal.java trunk/src/main/org/hornetq/utils/DecompressedLargeMessageBuffer.java Modified: trunk/src/main/org/hornetq/core/client/impl/ClientConsumerImpl.java trunk/src/main/org/hornetq/core/client/impl/ClientMessageImpl.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/persistence/StorageManager.java trunk/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java trunk/src/main/org/hornetq/core/persistence/impl/nullpm/NullStorageManager.java trunk/src/main/org/hornetq/core/protocol/core/ServerSessionPacketHandler.java trunk/src/main/org/hornetq/core/protocol/core/impl/CoreSessionCallback.java trunk/src/main/org/hornetq/core/protocol/core/impl/PacketDecoder.java trunk/src/main/org/hornetq/core/protocol/core/impl/wireformat/SessionReceiveLargeMessage.java trunk/src/main/org/hornetq/core/protocol/core/impl/wireformat/SessionSendLargeMessage.java trunk/src/main/org/hornetq/core/protocol/core/impl/wireformat/SessionSendMessage.java trunk/src/main/org/hornetq/core/protocol/stomp/StompSession.java trunk/src/main/org/hornetq/core/server/ServerSession.java trunk/src/main/org/hornetq/core/server/impl/ServerConsumerImpl.java trunk/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java trunk/src/main/org/hornetq/spi/core/protocol/SessionCallback.java trunk/tests/src/org/hornetq/tests/unit/core/client/impl/LargeMessageBufferTest.java trunk/tests/src/org/hornetq/tests/unit/core/paging/impl/PagingStoreImplTest.java trunk/tests/src/org/hornetq/tests/unit/core/postoffice/impl/BindingsImplTest.java Log: HORNETQ-446 - Improvements on Large message - first step (optimizing encoding - removing two phase encoding) Modified: trunk/src/main/org/hornetq/core/client/impl/ClientConsumerImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/ClientConsumerImpl.java 2010-12-07 18:05:17 UTC (rev 10007) +++ trunk/src/main/org/hornetq/core/client/impl/ClientConsumerImpl.java 2010-12-07 22:22:35 UTC (rev 10008) @@ -29,7 +29,6 @@ import org.hornetq.core.protocol.core.impl.wireformat.SessionQueueQueryResponseMessage; import org.hornetq.core.protocol.core.impl.wireformat.SessionReceiveContinuationMessage; import org.hornetq.core.protocol.core.impl.wireformat.SessionReceiveLargeMessage; -import org.hornetq.utils.DecompressedLargeMessageBuffer; import org.hornetq.utils.Future; import org.hornetq.utils.PriorityLinkedList; import org.hornetq.utils.PriorityLinkedListImpl; @@ -83,7 +82,7 @@ private final Runner runner = new Runner(); - private LargeMessageBufferImpl currentLargeMessageBuffer; + private LargeMessageControllerImpl currentLargeMessageController; // When receiving LargeMessages, the user may choose to not read the body, on this case we need to discard the body // before moving to the next message. @@ -539,14 +538,10 @@ flowControl(packet.getPacketSize(), false); - ClientMessageInternal currentChunkMessage = new ClientMessageImpl(); + ClientMessageInternal currentChunkMessage = (ClientMessageInternal)packet.getLargeMessage(); currentChunkMessage.setDeliveryCount(packet.getDeliveryCount()); - // FIXME - this is really inefficient - decoding from a buffer to a byte[] then from the byte[] to another buffer - // which is then decoded to form the message! Clebert, what were you thinking? - currentChunkMessage.decodeHeadersAndProperties(HornetQBuffers.wrappedBuffer(packet.getLargeMessageHeader())); - currentChunkMessage.setLargeMessage(true); File largeMessageCache = null; @@ -558,15 +553,15 @@ largeMessageCache.deleteOnExit(); } - currentLargeMessageBuffer = new LargeMessageBufferImpl(this, packet.getLargeMessageSize(), 60, largeMessageCache); + currentLargeMessageController = new LargeMessageControllerImpl(this, packet.getLargeMessageSize(), 60, largeMessageCache); if (currentChunkMessage.isCompressed()) { - currentChunkMessage.setBuffer(new DecompressedLargeMessageBuffer(currentLargeMessageBuffer)); + currentChunkMessage.setBuffer(new DecompressedLargeMessageControllerImpl(currentLargeMessageController)); } else { - currentChunkMessage.setBuffer(currentLargeMessageBuffer); + currentChunkMessage.setBuffer(currentLargeMessageController); } currentChunkMessage.setFlowControlSize(0); @@ -580,7 +575,7 @@ { return; } - currentLargeMessageBuffer.addPacket(chunk); + currentLargeMessageController.addPacket(chunk); } public void clear(boolean waitForOnMessage) throws HornetQException @@ -904,10 +899,10 @@ // Now we wait for any current handler runners to run. waitForOnMessageToComplete(true); - if (currentLargeMessageBuffer != null) + if (currentLargeMessageController != null) { - currentLargeMessageBuffer.cancel(); - currentLargeMessageBuffer = null; + currentLargeMessageController.cancel(); + currentLargeMessageController = null; } closed = true; Modified: trunk/src/main/org/hornetq/core/client/impl/ClientMessageImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/ClientMessageImpl.java 2010-12-07 18:05:17 UTC (rev 10007) +++ trunk/src/main/org/hornetq/core/client/impl/ClientMessageImpl.java 2010-12-07 22:22:35 UTC (rev 10008) @@ -47,6 +47,9 @@ private ClientConsumerInternal consumer; private boolean largeMessage; + + // Used only when receiving large messages + private LargeMessageController largeMessageController; private int flowControlSize = -1; @@ -159,7 +162,7 @@ { if (largeMessage) { - ((LargeMessageBufferInternal)getWholeBuffer()).saveBuffer(out); + ((LargeMessageController)getWholeBuffer()).saveBuffer(out); } else { @@ -184,7 +187,7 @@ { if (largeMessage) { - ((LargeMessageBufferInternal)getWholeBuffer()).setOutputStream(out); + ((LargeMessageController)getWholeBuffer()).setOutputStream(out); } else { @@ -200,7 +203,7 @@ { if (largeMessage) { - return ((LargeMessageBufferInternal)getWholeBuffer()).waitCompletion(timeMilliseconds); + return ((LargeMessageController)getWholeBuffer()).waitCompletion(timeMilliseconds); } else { @@ -215,7 +218,7 @@ { if (largeMessage) { - ((LargeMessageBufferInternal)getWholeBuffer()).discardUnusedPackets(); + ((LargeMessageController)getWholeBuffer()).discardUnusedPackets(); } } Modified: trunk/src/main/org/hornetq/core/client/impl/ClientProducerImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/ClientProducerImpl.java 2010-12-07 18:05:17 UTC (rev 10007) +++ trunk/src/main/org/hornetq/core/client/impl/ClientProducerImpl.java 2010-12-07 22:22:35 UTC (rev 10008) @@ -335,7 +335,7 @@ int headerSize = msgI.getHeadersAndPropertiesEncodeSize(); - if (headerSize >= minLargeMessageSize) + if (msgI.getHeadersAndPropertiesEncodeSize() >= minLargeMessageSize) { throw new HornetQException(HornetQException.ILLEGAL_STATE, "Header size (" + headerSize + ") is too big, use the messageBody for large data, or increase minLargeMessageSize"); @@ -347,11 +347,8 @@ msgI.getWholeBuffer().readerIndex(0); } - HornetQBuffer headerBuffer = HornetQBuffers.fixedBuffer(headerSize); + SessionSendLargeMessage initialChunk = new SessionSendLargeMessage(msgI); - msgI.encodeHeadersAndProperties(headerBuffer); - SessionSendLargeMessage initialChunk = new SessionSendLargeMessage(headerBuffer.toByteBuffer().array()); - channel.send(initialChunk); try Copied: trunk/src/main/org/hornetq/core/client/impl/DecompressedLargeMessageControllerImpl.java (from rev 10006, trunk/src/main/org/hornetq/utils/DecompressedLargeMessageBuffer.java) =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/DecompressedLargeMessageControllerImpl.java (rev 0) +++ trunk/src/main/org/hornetq/core/client/impl/DecompressedLargeMessageControllerImpl.java 2010-12-07 22:22:35 UTC (rev 10008) @@ -0,0 +1,1094 @@ +/* + * Copyright 2010 Red Hat, Inc. + * Red Hat 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.hornetq.core.client.impl; + +import java.io.DataInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.nio.ByteBuffer; +import java.nio.channels.GatheringByteChannel; +import java.nio.channels.ScatteringByteChannel; + +import org.hornetq.api.core.HornetQBuffer; +import org.hornetq.api.core.HornetQBuffers; +import org.hornetq.api.core.HornetQException; +import org.hornetq.api.core.SimpleString; +import org.hornetq.core.protocol.core.impl.wireformat.SessionReceiveContinuationMessage; +import org.hornetq.utils.DataConstants; +import org.hornetq.utils.HornetQBufferInputStream; +import org.hornetq.utils.InflaterReader; +import org.hornetq.utils.InflaterWriter; +import org.hornetq.utils.UTF8Util; +import org.jboss.netty.buffer.ChannelBuffer; + +/** + * A DecompressedHornetQBuffer + * + * @author <a href="mailto:clebert.suconic@jboss.org">Clebert Suconic</a> + * + * + */ +public class DecompressedLargeMessageControllerImpl implements LargeMessageController +{ + + // Constants ----------------------------------------------------- + + private static final String OPERATION_NOT_SUPPORTED = "Operation not supported"; + + private static final String READ_ONLY_ERROR_MESSAGE = "This is a read-only buffer, setOperations are not supported"; + + // Attributes ---------------------------------------------------- + + final LargeMessageController bufferDelegate; + + // Static -------------------------------------------------------- + + // Constructors -------------------------------------------------- + + public DecompressedLargeMessageControllerImpl(final LargeMessageController bufferDelegate) + { + this.bufferDelegate = bufferDelegate; + } + + + // Public -------------------------------------------------------- + + /** + * + */ + public void discardUnusedPackets() + { + bufferDelegate.discardUnusedPackets(); + } + + /** + * Add a buff to the List, or save it to the OutputStream if set + * @param packet + */ + public void addPacket(final SessionReceiveContinuationMessage packet) + { + bufferDelegate.addPacket(packet); + } + + public synchronized void cancel() + { + bufferDelegate.cancel(); + } + + public synchronized void close() + { + bufferDelegate.cancel(); + } + + public void setOutputStream(final OutputStream output) throws HornetQException + { + bufferDelegate.setOutputStream(new InflaterWriter(output)); + } + + public synchronized void saveBuffer(final OutputStream output) throws HornetQException + { + setOutputStream(output); + waitCompletion(0); + } + + /** + * + * @param timeWait Milliseconds to Wait. 0 means forever + * @throws Exception + */ + public synchronized boolean waitCompletion(final long timeWait) throws HornetQException + { + return bufferDelegate.waitCompletion(timeWait); + } + + // Channel Buffer Implementation --------------------------------- + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#array() + */ + public byte[] array() + { + throw new IllegalAccessError("array not supported on LargeMessageBufferImpl"); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#capacity() + */ + public int capacity() + { + return -1; + } + + DataInputStream dataInput = null; + + private DataInputStream getStream() + { + if (dataInput == null) + { + try + { + InputStream input = new HornetQBufferInputStream(bufferDelegate); + + dataInput = new DataInputStream(new InflaterReader(input)); + } + catch (Exception e) + { + throw new RuntimeException (e.getMessage(), e); + } + + } + return dataInput; + } + + private void positioningNotSupported() + { + throw new IllegalStateException("Position not supported over compressed large messages"); + } + + public byte readByte() + { + try + { + return getStream().readByte(); + } + catch (Exception e) + { + throw new RuntimeException (e.getMessage(), e); + } + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#getByte(int) + */ + public byte getByte(final int index) + { + positioningNotSupported(); + return 0; + } + + private byte getByte(final long index) + { + positioningNotSupported(); + return 0; + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, org.hornetq.api.core.buffers.ChannelBuffer, int, int) + */ + public void getBytes(final int index, final HornetQBuffer dst, final int dstIndex, final int length) + { + positioningNotSupported(); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, org.hornetq.api.core.buffers.ChannelBuffer, int, int) + */ + public void getBytes(final long index, final HornetQBuffer dst, final int dstIndex, final int length) + { + positioningNotSupported(); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, byte[], int, int) + */ + public void getBytes(final int index, final byte[] dst, final int dstIndex, final int length) + { + positioningNotSupported(); + } + + public void getBytes(final long index, final byte[] dst, final int dstIndex, final int length) + { + positioningNotSupported(); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, java.nio.ByteBuffer) + */ + public void getBytes(final int index, final ByteBuffer dst) + { + positioningNotSupported(); + } + + public void getBytes(final long index, final ByteBuffer dst) + { + positioningNotSupported(); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, java.io.OutputStream, int) + */ + public void getBytes(final int index, final OutputStream out, final int length) throws IOException + { + positioningNotSupported(); + } + + public void getBytes(final long index, final OutputStream out, final int length) throws IOException + { + positioningNotSupported(); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, java.nio.channels.GatheringByteChannel, int) + */ + public int getBytes(final int index, final GatheringByteChannel out, final int length) throws IOException + { + positioningNotSupported(); + return 0; + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#getInt(int) + */ + public int getInt(final int index) + { + positioningNotSupported(); + return 0; + } + + public int getInt(final long index) + { + positioningNotSupported(); + return 0; + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#getLong(int) + */ + public long getLong(final int index) + { + positioningNotSupported(); + return 0; + } + + public long getLong(final long index) + { + positioningNotSupported(); + return 0; + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#getShort(int) + */ + public short getShort(final int index) + { + positioningNotSupported(); + return 0; + } + + public short getShort(final long index) + { + return (short)(getByte(index) << 8 | getByte(index + 1) & 0xFF); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#getUnsignedMedium(int) + */ + public int getUnsignedMedium(final int index) + { + positioningNotSupported(); + return 0; + } + + + + public int getUnsignedMedium(final long index) + { + positioningNotSupported(); + return 0; + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#setByte(int, byte) + */ + public void setByte(final int index, final byte value) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#setBytes(int, org.hornetq.api.core.buffers.ChannelBuffer, int, int) + */ + public void setBytes(final int index, final HornetQBuffer src, final int srcIndex, final int length) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#setBytes(int, byte[], int, int) + */ + public void setBytes(final int index, final byte[] src, final int srcIndex, final int length) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#setBytes(int, java.nio.ByteBuffer) + */ + public void setBytes(final int index, final ByteBuffer src) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#setBytes(int, java.io.InputStream, int) + */ + public int setBytes(final int index, final InputStream in, final int length) throws IOException + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#setBytes(int, java.nio.channels.ScatteringByteChannel, int) + */ + public int setBytes(final int index, final ScatteringByteChannel in, final int length) throws IOException + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#setInt(int, int) + */ + public void setInt(final int index, final int value) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#setLong(int, long) + */ + public void setLong(final int index, final long value) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#setMedium(int, int) + */ + public void setMedium(final int index, final int value) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#setShort(int, short) + */ + public void setShort(final int index, final short value) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#toByteBuffer(int, int) + */ + public ByteBuffer toByteBuffer(final int index, final int length) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#toString(int, int, java.lang.String) + */ + public String toString(final int index, final int length, final String charsetName) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public int readerIndex() + { + return 0; + } + + public void readerIndex(final int readerIndex) + { + // TODO + } + + public int writerIndex() + { + // TODO + return 0; + } + + public long getSize() + { + // TODO + return 0; + } + + public void writerIndex(final int writerIndex) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void setIndex(final int readerIndex, final int writerIndex) + { + positioningNotSupported(); + } + + public void clear() + { + } + + public boolean readable() + { + return true; + } + + public boolean writable() + { + return false; + } + + public int readableBytes() + { + return 1; + } + + public int writableBytes() + { + return 0; + } + + public void markReaderIndex() + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void resetReaderIndex() + { + // TODO: reset positioning if possible + } + + public void markWriterIndex() + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void resetWriterIndex() + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void discardReadBytes() + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public short getUnsignedByte(final int index) + { + return (short)(getByte(index) & 0xFF); + } + + public int getUnsignedShort(final int index) + { + return getShort(index) & 0xFFFF; + } + + public int getMedium(final int index) + { + int value = getUnsignedMedium(index); + if ((value & 0x800000) != 0) + { + value |= 0xff000000; + } + return value; + } + + public long getUnsignedInt(final int index) + { + return getInt(index) & 0xFFFFFFFFL; + } + + public void getBytes(int index, final byte[] dst) + { + // TODO: optimize this by using System.arraycopy + for (int i = 0; i < dst.length; i++) + { + dst[i] = getByte(index++); + } + } + + public void getBytes(long index, final byte[] dst) + { + // TODO: optimize this by using System.arraycopy + for (int i = 0; i < dst.length; i++) + { + dst[i] = getByte(index++); + } + } + + public void getBytes(final int index, final HornetQBuffer dst) + { + getBytes(index, dst, dst.writableBytes()); + } + + public void getBytes(final int index, final HornetQBuffer dst, final int length) + { + if (length > dst.writableBytes()) + { + throw new IndexOutOfBoundsException(); + } + getBytes(index, dst, dst.writerIndex(), length); + dst.writerIndex(dst.writerIndex() + length); + } + + public void setBytes(final int index, final byte[] src) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void setBytes(final int index, final HornetQBuffer src) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void setBytes(final int index, final HornetQBuffer src, final int length) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void setZero(final int index, final int length) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public short readUnsignedByte() + { + try + { + return (short)getStream().readUnsignedByte(); + } + catch (Exception e) + { + throw new IllegalStateException (e.getMessage(), e); + } + } + + public short readShort() + { + try + { + return (short)getStream().readShort(); + } + catch (Exception e) + { + throw new IllegalStateException (e.getMessage(), e); + } + } + + public int readUnsignedShort() + { + try + { + return (int)getStream().readUnsignedShort(); + } + catch (Exception e) + { + throw new IllegalStateException (e.getMessage(), e); + } + } + + public int readMedium() + { + int value = readUnsignedMedium(); + if ((value & 0x800000) != 0) + { + value |= 0xff000000; + } + return value; + } + + + public int readUnsignedMedium() + { + return (readByte() & 0xff) << 16 | (readByte() & 0xff) << 8 | (readByte() & 0xff) << 0; + } + + public int readInt() + { + try + { + return getStream().readInt(); + } + catch (Exception e) + { + throw new IllegalStateException(e.getMessage(), e); + } + } + + public int readInt(final int pos) + { + positioningNotSupported(); + return 0; + } + + public long readUnsignedInt() + { + return readInt() & 0xFFFFFFFFL; + } + + public long readLong() + { + try + { + return getStream().readLong(); + } + catch (Exception e) + { + throw new IllegalStateException(e.getMessage(), e); + } + } + + public void readBytes(final byte[] dst, final int dstIndex, final int length) + { + try + { + getStream().read(dst, dstIndex, length); + } + catch (Exception e) + { + throw new IllegalStateException(e.getMessage(), e); + } + } + + public void readBytes(final byte[] dst) + { + readBytes(dst, 0, dst.length); + } + + public void readBytes(final HornetQBuffer dst) + { + readBytes(dst, dst.writableBytes()); + } + + public void readBytes(final HornetQBuffer dst, final int length) + { + if (length > dst.writableBytes()) + { + throw new IndexOutOfBoundsException(); + } + readBytes(dst, dst.writerIndex(), length); + dst.writerIndex(dst.writerIndex() + length); + } + + public void readBytes(final HornetQBuffer dst, final int dstIndex, final int length) + { + byte[] destBytes = new byte[length]; + readBytes(destBytes); + dst.setBytes(dstIndex, destBytes); + } + + public void readBytes(final ByteBuffer dst) + { + byte bytesToGet[] = new byte[dst.remaining()]; + readBytes(bytesToGet); + dst.put(bytesToGet); + } + + public int readBytes(final GatheringByteChannel out, final int length) throws IOException + { + throw new IllegalStateException("Not implemented!"); + } + + public void readBytes(final OutputStream out, final int length) throws IOException + { + throw new IllegalStateException("Not implemented!"); + } + + public void skipBytes(final int length) + { + + try + { + for (int i = 0 ; i < length; i++) + { + getStream().read(); + } + } + catch (Exception e) + { + throw new IllegalStateException(e.getMessage(), e); + } + } + + public void writeByte(final byte value) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void writeShort(final short value) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void writeMedium(final int value) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void writeInt(final int value) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void writeLong(final long value) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void writeBytes(final byte[] src, final int srcIndex, final int length) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void writeBytes(final byte[] src) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void writeBytes(final HornetQBuffer src) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void writeBytes(final HornetQBuffer src, final int length) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void writeBytes(final ByteBuffer src) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public int writeBytes(final InputStream in, final int length) throws IOException + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public int writeBytes(final ScatteringByteChannel in, final int length) throws IOException + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void writeZero(final int length) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public ByteBuffer toByteBuffer() + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public ByteBuffer[] toByteBuffers() + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public ByteBuffer[] toByteBuffers(final int index, final int length) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public String toString(final String charsetName) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public Object getUnderlyingBuffer() + { + return this; + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#readBoolean() + */ + public boolean readBoolean() + { + return readByte() != 0; + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#readChar() + */ + public char readChar() + { + return (char)readShort(); + } + + public char getChar(final int index) + { + return (char)getShort(index); + } + + public double getDouble(final int index) + { + return Double.longBitsToDouble(getLong(index)); + } + + public float getFloat(final int index) + { + return Float.intBitsToFloat(getInt(index)); + } + + public HornetQBuffer readBytes(final int length) + { + byte bytesToGet[] = new byte[length]; + readBytes(bytesToGet); + return HornetQBuffers.wrappedBuffer(bytesToGet); + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#readDouble() + */ + public double readDouble() + { + return Double.longBitsToDouble(readLong()); + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#readFloat() + */ + public float readFloat() + { + return Float.intBitsToFloat(readInt()); + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#readNullableSimpleString() + */ + public SimpleString readNullableSimpleString() + { + int b = readByte(); + if (b == DataConstants.NULL) + { + return null; + } + else + { + return readSimpleString(); + } + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#readNullableString() + */ + public String readNullableString() + { + int b = readByte(); + if (b == DataConstants.NULL) + { + return null; + } + else + { + return readString(); + } + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#readSimpleString() + */ + public SimpleString readSimpleString() + { + int len = readInt(); + byte[] data = new byte[len]; + readBytes(data); + return new SimpleString(data); + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#readString() + */ + public String readString() + { + int len = readInt(); + + if (len < 9) + { + char[] chars = new char[len]; + for (int i = 0; i < len; i++) + { + chars[i] = (char)readShort(); + } + return new String(chars); + } + else if (len < 0xfff) + { + return readUTF(); + } + else + { + return readSimpleString().toString(); + } + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#readUTF() + */ + public String readUTF() + { + return UTF8Util.readUTF(this); + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeBoolean(boolean) + */ + public void writeBoolean(final boolean val) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeChar(char) + */ + public void writeChar(final char val) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeDouble(double) + */ + public void writeDouble(final double val) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeFloat(float) + */ + public void writeFloat(final float val) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeNullableSimpleString(org.hornetq.util.SimpleString) + */ + public void writeNullableSimpleString(final SimpleString val) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeNullableString(java.lang.String) + */ + public void writeNullableString(final String val) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeSimpleString(org.hornetq.util.SimpleString) + */ + public void writeSimpleString(final SimpleString val) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeString(java.lang.String) + */ + public void writeString(final String val) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeUTF(java.lang.String) + */ + public void writeUTF(final String utf) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#compareTo(org.hornetq.api.core.buffers.ChannelBuffer) + */ + public int compareTo(final HornetQBuffer buffer) + { + return -1; + } + + public HornetQBuffer copy() + { + throw new UnsupportedOperationException(); + } + + public HornetQBuffer slice(final int index, final int length) + { + throw new UnsupportedOperationException(); + } + + // Package protected --------------------------------------------- + + // Protected ----------------------------------------------------- + + // Private ------------------------------------------------------- + + /** + * @param body + */ + // Inner classes ------------------------------------------------- + + public ChannelBuffer channelBuffer() + { + return null; + } + + public HornetQBuffer copy(final int index, final int length) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public HornetQBuffer duplicate() + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public HornetQBuffer readSlice(final int length) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void setChar(final int index, final char value) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void setDouble(final int index, final double value) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void setFloat(final int index, final float value) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public HornetQBuffer slice() + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void writeBytes(final HornetQBuffer src, final int srcIndex, final int length) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } +} Deleted: trunk/src/main/org/hornetq/core/client/impl/LargeMessageBufferImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/LargeMessageBufferImpl.java 2010-12-07 18:05:17 UTC (rev 10007) +++ trunk/src/main/org/hornetq/core/client/impl/LargeMessageBufferImpl.java 2010-12-07 22:22:35 UTC (rev 10008) @@ -1,1535 +0,0 @@ -/* - * Copyright 2009 Red Hat, Inc. - * Red Hat 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.hornetq.core.client.impl; - -import java.io.File; -import java.io.FileNotFoundException; -import java.io.IOException; -import java.io.InputStream; -import java.io.OutputStream; -import java.io.RandomAccessFile; -import java.nio.ByteBuffer; -import java.nio.channels.FileChannel; -import java.nio.channels.GatheringByteChannel; -import java.nio.channels.ScatteringByteChannel; -import java.util.concurrent.LinkedBlockingQueue; -import java.util.concurrent.TimeUnit; - -import org.hornetq.api.core.HornetQBuffer; -import org.hornetq.api.core.HornetQBuffers; -import org.hornetq.api.core.HornetQException; -import org.hornetq.api.core.SimpleString; -import org.hornetq.core.logging.Logger; -import org.hornetq.core.protocol.core.impl.wireformat.SessionReceiveContinuationMessage; -import org.hornetq.utils.DataConstants; -import org.hornetq.utils.UTF8Util; -import org.jboss.netty.buffer.ChannelBuffer; - -/** - * This class aggregates several SessionReceiveContinuationMessages as it was being handled by a single buffer. - * This buffer can be consumed as messages are arriving, and it will hold the packets until they are read using the ChannelBuffer interface, or the setOutputStream or saveStream are called. - * - * @author <a href="mailto:clebert.suconic@jboss.org">Clebert Suconic</a> - * - * - */ -public class LargeMessageBufferImpl implements LargeMessageBufferInternal -{ - // Constants ----------------------------------------------------- - - private static final String READ_ONLY_ERROR_MESSAGE = "This is a read-only buffer, setOperations are not supported"; - - // Attributes ---------------------------------------------------- - - private static final Logger log = Logger.getLogger(LargeMessageBufferImpl.class); - - private final ClientConsumerInternal consumerInternal; - - private final LinkedBlockingQueue<SessionReceiveContinuationMessage> packets = new LinkedBlockingQueue<SessionReceiveContinuationMessage>(); - - private volatile SessionReceiveContinuationMessage currentPacket = null; - - private final long totalSize; - - private final int bufferSize; - - private boolean streamEnded = false; - - private boolean streamClosed = false; - - private final int readTimeout; - - private long readerIndex = 0; - - private long packetPosition = -1; - - private long lastIndex = 0; - - private long packetLastPosition = -1; - - private OutputStream outStream; - - private Exception handledException; - - private final FileCache fileCache; - - // Static -------------------------------------------------------- - - // Constructors -------------------------------------------------- - - public LargeMessageBufferImpl(final ClientConsumerInternal consumerInternal, - final long totalSize, - final int readTimeout) - { - this(consumerInternal, totalSize, readTimeout, null); - } - - public LargeMessageBufferImpl(final ClientConsumerInternal consumerInternal, - final long totalSize, - final int readTimeout, - final File cachedFile) - { - this(consumerInternal, totalSize, readTimeout, cachedFile, 10 * 1024); - } - - public LargeMessageBufferImpl(final ClientConsumerInternal consumerInternal, - final long totalSize, - final int readTimeout, - final File cachedFile, - final int bufferSize) - { - this.consumerInternal = consumerInternal; - this.readTimeout = readTimeout; - this.totalSize = totalSize; - if (cachedFile == null) - { - fileCache = null; - } - else - { - fileCache = new FileCache(cachedFile); - } - this.bufferSize = bufferSize; - } - - // Public -------------------------------------------------------- - - public synchronized Exception getHandledException() - { - return handledException; - } - - /** - * - */ - public void discardUnusedPackets() - { - if (outStream == null) - { - try - { - checkForPacket(totalSize - 1); - } - catch (Exception ignored) - { - } - } - } - - /** - * Add a buff to the List, or save it to the OutputStream if set - * @param packet - */ - public void addPacket(final SessionReceiveContinuationMessage packet) - { - int flowControlCredit = 0; - boolean continues = false; - - synchronized (this) - { - if (outStream != null) - { - try - { - if (!packet.isContinues()) - { - streamEnded = true; - } - - if (fileCache != null) - { - fileCache.cachePackage(packet.getBody()); - } - - outStream.write(packet.getBody()); - - flowControlCredit = packet.getPacketSize(); - - continues = packet.isContinues(); - - notifyAll(); - - if (streamEnded) - { - outStream.close(); - } - } - catch (Exception e) - { - LargeMessageBufferImpl.log.warn(e.getMessage(), e); - handledException = e; - } - } - else - { - if (fileCache != null) - { - try - { - fileCache.cachePackage(packet.getBody()); - } - catch (Exception e) - { - LargeMessageBufferImpl.log.warn(e.getMessage(), e); - handledException = e; - } - } - - - packets.offer(packet); - } - } - - if (flowControlCredit != 0) - { - try - { - consumerInternal.flowControl(flowControlCredit, !continues); - } - catch (Exception e) - { - LargeMessageBufferImpl.log.warn(e.getMessage(), e); - handledException = e; - } - } - } - - public synchronized void cancel() - { - packets.offer(new SessionReceiveContinuationMessage()); - streamEnded = true; - streamClosed = true; - - notifyAll(); - } - - public synchronized void close() - { - if (fileCache != null) - { - fileCache.close(); - } - } - - public void setOutputStream(final OutputStream output) throws HornetQException - { - - int totalFlowControl = 0; - boolean continues = false; - - synchronized (this) - { - if (currentPacket != null) - { - sendPacketToOutput(output, currentPacket); - currentPacket = null; - } - while (true) - { - SessionReceiveContinuationMessage packet = packets.poll(); - if (packet == null) - { - break; - } - totalFlowControl += packet.getPacketSize(); - - continues = packet.isContinues(); - sendPacketToOutput(output, packet); - } - - outStream = output; - } - - consumerInternal.flowControl(totalFlowControl, !continues); - } - - public synchronized void saveBuffer(final OutputStream output) throws HornetQException - { - setOutputStream(output); - waitCompletion(0); - } - - /** - * - * @param timeWait Milliseconds to Wait. 0 means forever - * @throws Exception - */ - public synchronized boolean waitCompletion(final long timeWait) throws HornetQException - { - if (outStream == null) - { - // There is no stream.. it will never achieve the end of streaming - return false; - } - - long timeOut = System.currentTimeMillis() + timeWait; - while (!streamEnded && handledException == null) - { - try - { - this.wait(readTimeout == 0 ? 1 : readTimeout * 1000); - } - catch (InterruptedException e) - { - throw new HornetQException(HornetQException.INTERNAL_ERROR, e.getMessage(), e); - } - - if (timeWait > 0 && System.currentTimeMillis() > timeOut) - { - throw new HornetQException(HornetQException.LARGE_MESSAGE_ERROR_BODY, - "Timeout waiting for LargeMessage Body"); - } - } - - if (handledException != null) - { - throw new HornetQException(HornetQException.LARGE_MESSAGE_ERROR_BODY, - "Error on saving LargeMessageBufferImpl", - handledException); - } - - return streamEnded; - - } - - // Channel Buffer Implementation --------------------------------- - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#array() - */ - public byte[] array() - { - throw new IllegalAccessError("array not supported on LargeMessageBufferImpl"); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#capacity() - */ - public int capacity() - { - return -1; - } - - public byte readByte() - { - return getByte(readerIndex++); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#getByte(int) - */ - public byte getByte(final int index) - { - return getByte((long)index); - } - - private byte getByte(final long index) - { - checkForPacket(index); - - if (fileCache != null && index < packetPosition) - { - return fileCache.getByteFromCache(index); - } - else - { - return currentPacket.getBody()[(int)(index - packetPosition)]; - } - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, org.hornetq.api.core.buffers.ChannelBuffer, int, int) - */ - public void getBytes(final int index, final HornetQBuffer dst, final int dstIndex, final int length) - { - byte[] destBytes = new byte[length]; - getBytes(index, destBytes); - dst.setBytes(dstIndex, destBytes); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, org.hornetq.api.core.buffers.ChannelBuffer, int, int) - */ - public void getBytes(final long index, final HornetQBuffer dst, final int dstIndex, final int length) - { - byte[] destBytes = new byte[length]; - getBytes(index, destBytes); - dst.setBytes(dstIndex, destBytes); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, byte[], int, int) - */ - public void getBytes(final int index, final byte[] dst, final int dstIndex, final int length) - { - byte bytesToGet[] = new byte[length]; - - getBytes(index, bytesToGet); - - System.arraycopy(bytesToGet, 0, dst, dstIndex, length); - } - - public void getBytes(final long index, final byte[] dst, final int dstIndex, final int length) - { - byte bytesToGet[] = new byte[length]; - - getBytes(index, bytesToGet); - - System.arraycopy(bytesToGet, 0, dst, dstIndex, length); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, java.nio.ByteBuffer) - */ - public void getBytes(final int index, final ByteBuffer dst) - { - byte bytesToGet[] = new byte[dst.remaining()]; - getBytes(index, bytesToGet); - dst.put(bytesToGet); - } - - public void getBytes(final long index, final ByteBuffer dst) - { - byte bytesToGet[] = new byte[dst.remaining()]; - getBytes(index, bytesToGet); - dst.put(bytesToGet); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, java.io.OutputStream, int) - */ - public void getBytes(final int index, final OutputStream out, final int length) throws IOException - { - byte bytesToGet[] = new byte[length]; - getBytes(index, bytesToGet); - out.write(bytesToGet); - } - - public void getBytes(final long index, final OutputStream out, final int length) throws IOException - { - byte bytesToGet[] = new byte[length]; - getBytes(index, bytesToGet); - out.write(bytesToGet); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, java.nio.channels.GatheringByteChannel, int) - */ - public int getBytes(final int index, final GatheringByteChannel out, final int length) throws IOException - { - byte bytesToGet[] = new byte[length]; - getBytes(index, bytesToGet); - return out.write(ByteBuffer.wrap(bytesToGet)); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#getInt(int) - */ - public int getInt(final int index) - { - return (getByte(index) & 0xff) << 24 | (getByte(index + 1) & 0xff) << 16 | - (getByte(index + 2) & 0xff) << 8 | - (getByte(index + 3) & 0xff) << 0; - } - - public int getInt(final long index) - { - return (getByte(index) & 0xff) << 24 | (getByte(index + 1) & 0xff) << 16 | - (getByte(index + 2) & 0xff) << 8 | - (getByte(index + 3) & 0xff) << 0; - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#getLong(int) - */ - public long getLong(final int index) - { - return ((long)getByte(index) & 0xff) << 56 | ((long)getByte(index + 1) & 0xff) << 48 | - ((long)getByte(index + 2) & 0xff) << 40 | - ((long)getByte(index + 3) & 0xff) << 32 | - ((long)getByte(index + 4) & 0xff) << 24 | - ((long)getByte(index + 5) & 0xff) << 16 | - ((long)getByte(index + 6) & 0xff) << 8 | - ((long)getByte(index + 7) & 0xff) << 0; - } - - public long getLong(final long index) - { - return ((long)getByte(index) & 0xff) << 56 | ((long)getByte(index + 1) & 0xff) << 48 | - ((long)getByte(index + 2) & 0xff) << 40 | - ((long)getByte(index + 3) & 0xff) << 32 | - ((long)getByte(index + 4) & 0xff) << 24 | - ((long)getByte(index + 5) & 0xff) << 16 | - ((long)getByte(index + 6) & 0xff) << 8 | - ((long)getByte(index + 7) & 0xff) << 0; - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#getShort(int) - */ - public short getShort(final int index) - { - return (short)(getByte(index) << 8 | getByte(index + 1) & 0xFF); - } - - public short getShort(final long index) - { - return (short)(getByte(index) << 8 | getByte(index + 1) & 0xFF); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#getUnsignedMedium(int) - */ - public int getUnsignedMedium(final int index) - { - return (getByte(index) & 0xff) << 16 | (getByte(index + 1) & 0xff) << 8 | (getByte(index + 2) & 0xff) << 0; - } - - public int getUnsignedMedium(final long index) - { - return (getByte(index) & 0xff) << 16 | (getByte(index + 1) & 0xff) << 8 | (getByte(index + 2) & 0xff) << 0; - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#setByte(int, byte) - */ - public void setByte(final int index, final byte value) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#setBytes(int, org.hornetq.api.core.buffers.ChannelBuffer, int, int) - */ - public void setBytes(final int index, final HornetQBuffer src, final int srcIndex, final int length) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#setBytes(int, byte[], int, int) - */ - public void setBytes(final int index, final byte[] src, final int srcIndex, final int length) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#setBytes(int, java.nio.ByteBuffer) - */ - public void setBytes(final int index, final ByteBuffer src) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#setBytes(int, java.io.InputStream, int) - */ - public int setBytes(final int index, final InputStream in, final int length) throws IOException - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#setBytes(int, java.nio.channels.ScatteringByteChannel, int) - */ - public int setBytes(final int index, final ScatteringByteChannel in, final int length) throws IOException - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#setInt(int, int) - */ - public void setInt(final int index, final int value) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#setLong(int, long) - */ - public void setLong(final int index, final long value) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#setMedium(int, int) - */ - public void setMedium(final int index, final int value) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#setShort(int, short) - */ - public void setShort(final int index, final short value) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#toByteBuffer(int, int) - */ - public ByteBuffer toByteBuffer(final int index, final int length) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#toString(int, int, java.lang.String) - */ - public String toString(final int index, final int length, final String charsetName) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public int readerIndex() - { - return (int)readerIndex; - } - - public void readerIndex(final int readerIndex) - { - try - { - checkForPacket(readerIndex); - } - catch (Exception e) - { - LargeMessageBufferImpl.log.warn(e.getMessage(), e); - throw new RuntimeException(e.getMessage(), e); - } - this.readerIndex = readerIndex; - } - - public int writerIndex() - { - return (int)totalSize; - } - - public long getSize() - { - return totalSize; - } - - public void writerIndex(final int writerIndex) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public void setIndex(final int readerIndex, final int writerIndex) - { - try - { - checkForPacket(readerIndex); - } - catch (Exception e) - { - LargeMessageBufferImpl.log.warn(e.getMessage(), e); - throw new RuntimeException(e.getMessage(), e); - } - this.readerIndex = readerIndex; - } - - public void clear() - { - } - - public boolean readable() - { - return true; - } - - public boolean writable() - { - return false; - } - - public int readableBytes() - { - long readableBytes = totalSize - readerIndex; - - if (readableBytes > Integer.MAX_VALUE) - { - return Integer.MAX_VALUE; - } - else - { - return (int)(totalSize - readerIndex); - } - } - - public int writableBytes() - { - return 0; - } - - public void markReaderIndex() - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public void resetReaderIndex() - { - try - { - checkForPacket(0); - } - catch (Exception e) - { - LargeMessageBufferImpl.log.warn(e.getMessage(), e); - throw new RuntimeException(e.getMessage(), e); - } - } - - public void markWriterIndex() - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public void resetWriterIndex() - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public void discardReadBytes() - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public short getUnsignedByte(final int index) - { - return (short)(getByte(index) & 0xFF); - } - - public int getUnsignedShort(final int index) - { - return getShort(index) & 0xFFFF; - } - - public int getMedium(final int index) - { - int value = getUnsignedMedium(index); - if ((value & 0x800000) != 0) - { - value |= 0xff000000; - } - return value; - } - - public long getUnsignedInt(final int index) - { - return getInt(index) & 0xFFFFFFFFL; - } - - public void getBytes(int index, final byte[] dst) - { - // TODO: optimize this by using System.arraycopy - for (int i = 0; i < dst.length; i++) - { - dst[i] = getByte(index++); - } - } - - public void getBytes(long index, final byte[] dst) - { - // TODO: optimize this by using System.arraycopy - for (int i = 0; i < dst.length; i++) - { - dst[i] = getByte(index++); - } - } - - public void getBytes(final int index, final HornetQBuffer dst) - { - getBytes(index, dst, dst.writableBytes()); - } - - public void getBytes(final int index, final HornetQBuffer dst, final int length) - { - if (length > dst.writableBytes()) - { - throw new IndexOutOfBoundsException(); - } - getBytes(index, dst, dst.writerIndex(), length); - dst.writerIndex(dst.writerIndex() + length); - } - - public void setBytes(final int index, final byte[] src) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public void setBytes(final int index, final HornetQBuffer src) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public void setBytes(final int index, final HornetQBuffer src, final int length) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public void setZero(final int index, final int length) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public short readUnsignedByte() - { - return (short)(readByte() & 0xFF); - } - - public short readShort() - { - short v = getShort(readerIndex); - readerIndex += 2; - return v; - } - - public int readUnsignedShort() - { - return readShort() & 0xFFFF; - } - - public int readMedium() - { - int value = readUnsignedMedium(); - if ((value & 0x800000) != 0) - { - value |= 0xff000000; - } - return value; - } - - public int readUnsignedMedium() - { - int v = getUnsignedMedium(readerIndex); - readerIndex += 3; - return v; - } - - public int readInt() - { - int v = getInt(readerIndex); - readerIndex += 4; - return v; - } - - public int readInt(final int pos) - { - int v = getInt(pos); - return v; - } - - public long readUnsignedInt() - { - return readInt() & 0xFFFFFFFFL; - } - - public long readLong() - { - long v = getLong(readerIndex); - readerIndex += 8; - return v; - } - - public void readBytes(final byte[] dst, final int dstIndex, final int length) - { - getBytes(readerIndex, dst, dstIndex, length); - readerIndex += length; - } - - public void readBytes(final byte[] dst) - { - readBytes(dst, 0, dst.length); - } - - public void readBytes(final HornetQBuffer dst) - { - readBytes(dst, dst.writableBytes()); - } - - public void readBytes(final HornetQBuffer dst, final int length) - { - if (length > dst.writableBytes()) - { - throw new IndexOutOfBoundsException(); - } - readBytes(dst, dst.writerIndex(), length); - dst.writerIndex(dst.writerIndex() + length); - } - - public void readBytes(final HornetQBuffer dst, final int dstIndex, final int length) - { - getBytes(readerIndex, dst, dstIndex, length); - readerIndex += length; - } - - public void readBytes(final ByteBuffer dst) - { - int length = dst.remaining(); - getBytes(readerIndex, dst); - readerIndex += length; - } - - public int readBytes(final GatheringByteChannel out, final int length) throws IOException - { - int readBytes = getBytes((int)readerIndex, out, length); - readerIndex += readBytes; - return readBytes; - } - - public void readBytes(final OutputStream out, final int length) throws IOException - { - getBytes(readerIndex, out, length); - readerIndex += length; - } - - public void skipBytes(final int length) - { - - long newReaderIndex = readerIndex + length; - checkForPacket(newReaderIndex); - readerIndex = newReaderIndex; - } - - public void writeByte(final byte value) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public void writeShort(final short value) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public void writeMedium(final int value) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public void writeInt(final int value) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public void writeLong(final long value) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public void writeBytes(final byte[] src, final int srcIndex, final int length) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public void writeBytes(final byte[] src) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public void writeBytes(final HornetQBuffer src) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public void writeBytes(final HornetQBuffer src, final int length) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public void writeBytes(final ByteBuffer src) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public int writeBytes(final InputStream in, final int length) throws IOException - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public int writeBytes(final ScatteringByteChannel in, final int length) throws IOException - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public void writeZero(final int length) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public ByteBuffer toByteBuffer() - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public ByteBuffer[] toByteBuffers() - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public ByteBuffer[] toByteBuffers(final int index, final int length) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public String toString(final String charsetName) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public Object getUnderlyingBuffer() - { - return this; - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#readBoolean() - */ - public boolean readBoolean() - { - return readByte() != 0; - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#readChar() - */ - public char readChar() - { - return (char)readShort(); - } - - public char getChar(final int index) - { - return (char)getShort(index); - } - - public double getDouble(final int index) - { - return Double.longBitsToDouble(getLong(index)); - } - - public float getFloat(final int index) - { - return Float.intBitsToFloat(getInt(index)); - } - - public HornetQBuffer readBytes(final int length) - { - byte bytesToGet[] = new byte[length]; - getBytes(readerIndex, bytesToGet); - readerIndex += length; - return HornetQBuffers.wrappedBuffer(bytesToGet); - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#readDouble() - */ - public double readDouble() - { - return Double.longBitsToDouble(readLong()); - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#readFloat() - */ - public float readFloat() - { - return Float.intBitsToFloat(readInt()); - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#readNullableSimpleString() - */ - public SimpleString readNullableSimpleString() - { - int b = readByte(); - if (b == DataConstants.NULL) - { - return null; - } - else - { - return readSimpleString(); - } - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#readNullableString() - */ - public String readNullableString() - { - int b = readByte(); - if (b == DataConstants.NULL) - { - return null; - } - else - { - return readString(); - } - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#readSimpleString() - */ - public SimpleString readSimpleString() - { - int len = readInt(); - byte[] data = new byte[len]; - readBytes(data); - return new SimpleString(data); - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#readString() - */ - public String readString() - { - int len = readInt(); - - if (len < 9) - { - char[] chars = new char[len]; - for (int i = 0; i < len; i++) - { - chars[i] = (char)readShort(); - } - return new String(chars); - } - else if (len < 0xfff) - { - return readUTF(); - } - else - { - return readSimpleString().toString(); - } - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#readUTF() - */ - public String readUTF() - { - return UTF8Util.readUTF(this); - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeBoolean(boolean) - */ - public void writeBoolean(final boolean val) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeChar(char) - */ - public void writeChar(final char val) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeDouble(double) - */ - public void writeDouble(final double val) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeFloat(float) - */ - public void writeFloat(final float val) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeNullableSimpleString(org.hornetq.util.SimpleString) - */ - public void writeNullableSimpleString(final SimpleString val) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeNullableString(java.lang.String) - */ - public void writeNullableString(final String val) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeSimpleString(org.hornetq.util.SimpleString) - */ - public void writeSimpleString(final SimpleString val) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeString(java.lang.String) - */ - public void writeString(final String val) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeUTF(java.lang.String) - */ - public void writeUTF(final String utf) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#compareTo(org.hornetq.api.core.buffers.ChannelBuffer) - */ - public int compareTo(final HornetQBuffer buffer) - { - return -1; - } - - public HornetQBuffer copy() - { - throw new UnsupportedOperationException(); - } - - public HornetQBuffer slice(final int index, final int length) - { - throw new UnsupportedOperationException(); - } - - // Package protected --------------------------------------------- - - // Protected ----------------------------------------------------- - - // Private ------------------------------------------------------- - - /** - * @param output - * @param packet - * @throws HornetQException - */ - private void sendPacketToOutput(final OutputStream output, final SessionReceiveContinuationMessage packet) throws HornetQException - { - try - { - output.write(packet.getBody()); - if (!packet.isContinues()) - { - streamEnded = true; - output.close(); - } - } - catch (IOException e) - { - throw new HornetQException(HornetQException.LARGE_MESSAGE_ERROR_BODY, "Error writing body of message", e); - } - } - - private void popPacket() - { - try - { - - if (streamEnded) - { - // no more packets, we are over the last one already - throw new IndexOutOfBoundsException(); - } - - int sizeToAdd = currentPacket != null ? currentPacket.getBody().length : 1; - currentPacket = packets.poll(readTimeout, TimeUnit.SECONDS); - if (currentPacket == null) - { - throw new IndexOutOfBoundsException(); - } - - if (currentPacket.getBody() == null) // Empty packet as a signal to interruption - { - currentPacket = null; - streamEnded = true; - throw new IndexOutOfBoundsException(); - } - - consumerInternal.flowControl(currentPacket.getPacketSize(), !currentPacket.isContinues()); - - packetPosition += sizeToAdd; - - packetLastPosition = packetPosition + currentPacket.getBody().length; - } - catch (IndexOutOfBoundsException e) - { - throw e; - } - catch (Exception e) - { - throw new RuntimeException(e); - } - } - - private void checkForPacket(final long index) - { - if (outStream != null) - { - throw new IllegalAccessError("Can't read the messageBody after setting outputStream"); - } - - if (index >= totalSize) - { - throw new IndexOutOfBoundsException(); - } - - if (streamClosed) - { - throw new IllegalAccessError("The consumer associated with this large message was closed before the body was read"); - } - - if (fileCache == null) - { - if (index < lastIndex) - { - throw new IllegalAccessError("LargeMessage have read-only and one-way buffers"); - } - lastIndex = index; - } - - while (index >= packetLastPosition && !streamEnded) - { - popPacket(); - } - } - - /** - * @param body - */ - // Inner classes ------------------------------------------------- - private class FileCache - { - - public FileCache(final File cachedFile) - { - this.cachedFile = cachedFile; - } - - ByteBuffer readCache; - - long readCachePositionStart = Integer.MAX_VALUE; - - long readCachePositionEnd = -1; - - private final File cachedFile; - - private volatile RandomAccessFile cachedRAFile; - - private volatile FileChannel cachedChannel; - - private synchronized void readCache(final long position) - { - - try - { - if (position < readCachePositionStart || position > readCachePositionEnd) - { - - checkOpen(); - - if (position > cachedChannel.size()) - { - throw new ArrayIndexOutOfBoundsException("position > " + cachedChannel.size()); - } - - readCachePositionStart = position / bufferSize * bufferSize; - - cachedChannel.position(readCachePositionStart); - - if (readCache == null) - { - readCache = ByteBuffer.allocate(bufferSize); - } - - readCache.clear(); - - readCachePositionEnd = readCachePositionStart + cachedChannel.read(readCache) - 1; - } - } - catch (Exception e) - { - LargeMessageBufferImpl.log.warn(e.getMessage(), e); - throw new RuntimeException(e.getMessage(), e); - } - finally - { - close(); - } - } - - public synchronized byte getByteFromCache(final long position) - { - readCache(position); - - return readCache.get((int)(position - readCachePositionStart)); - - } - - public void cachePackage(final byte[] body) throws Exception - { - checkOpen(); - - cachedChannel.position(cachedChannel.size()); - cachedChannel.write(ByteBuffer.wrap(body)); - - close(); - } - - /** - * @throws FileNotFoundException - */ - public void checkOpen() throws FileNotFoundException - { - if (cachedFile != null || !cachedChannel.isOpen()) - { - cachedRAFile = new RandomAccessFile(cachedFile, "rw"); - - cachedChannel = cachedRAFile.getChannel(); - } - } - - public void close() - { - if (cachedChannel != null && cachedChannel.isOpen()) - { - try - { - cachedChannel.close(); - } - catch (Exception e) - { - LargeMessageBufferImpl.log.warn(e.getMessage(), e); - } - cachedChannel = null; - } - - if (cachedRAFile != null) - { - try - { - cachedRAFile.close(); - } - catch (Exception e) - { - LargeMessageBufferImpl.log.warn(e.getMessage(), e); - } - cachedRAFile = null; - } - - } - - @Override - protected void finalize() - { - close(); - if (cachedFile != null && cachedFile.exists()) - { - try - { - cachedFile.delete(); - } - catch (Exception e) - { - LargeMessageBufferImpl.log.warn("Exception during finalization for LargeMessage file cache", e); - } - } - } - - } - - public ChannelBuffer channelBuffer() - { - return null; - } - - public HornetQBuffer copy(final int index, final int length) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public HornetQBuffer duplicate() - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public HornetQBuffer readSlice(final int length) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public void setChar(final int index, final char value) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public void setDouble(final int index, final double value) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public void setFloat(final int index, final float value) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public HornetQBuffer slice() - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public void writeBytes(final HornetQBuffer src, final int srcIndex, final int length) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - -} Deleted: trunk/src/main/org/hornetq/core/client/impl/LargeMessageBufferInternal.java =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/LargeMessageBufferInternal.java 2010-12-07 18:05:17 UTC (rev 10007) +++ trunk/src/main/org/hornetq/core/client/impl/LargeMessageBufferInternal.java 2010-12-07 22:22:35 UTC (rev 10008) @@ -1,67 +0,0 @@ -/* - * Copyright 2009 Red Hat, Inc. - * Red Hat 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.hornetq.core.client.impl; - -import java.io.OutputStream; - -import org.hornetq.api.core.HornetQBuffer; -import org.hornetq.api.core.HornetQException; -import org.hornetq.core.protocol.core.impl.wireformat.SessionReceiveContinuationMessage; - -/** - * A LargeMessageBufferInternal - * - * @author <a href="mailto:jmesnil@redhat.com">Jeff Mesnil</a> - */ -public interface LargeMessageBufferInternal extends HornetQBuffer -{ - /** - * Returns the size of this buffer. - - */ - long getSize(); - - /** - * Discards packets unused by this buffer. - */ - void discardUnusedPackets(); - - /** - * Closes this buffer. - */ - void close(); - - /** - * Cancels this buffer. - */ - void cancel(); - - /** - * Sets the OutputStream of this buffer to the specified output. - */ - void setOutputStream(final OutputStream output) throws HornetQException; - - /** - * Saves this buffer to the specified output. - */ - void saveBuffer(final OutputStream output) throws HornetQException; - - public void addPacket(final SessionReceiveContinuationMessage packet); - - /** - * Waits for the completion for the specified waiting time (in milliseconds). - */ - boolean waitCompletion(long timeWait) throws HornetQException; - -} Copied: trunk/src/main/org/hornetq/core/client/impl/LargeMessageController.java (from rev 10006, trunk/src/main/org/hornetq/core/client/impl/LargeMessageBufferInternal.java) =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/LargeMessageController.java (rev 0) +++ trunk/src/main/org/hornetq/core/client/impl/LargeMessageController.java 2010-12-07 22:22:35 UTC (rev 10008) @@ -0,0 +1,67 @@ +/* + * Copyright 2009 Red Hat, Inc. + * Red Hat 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.hornetq.core.client.impl; + +import java.io.OutputStream; + +import org.hornetq.api.core.HornetQBuffer; +import org.hornetq.api.core.HornetQException; +import org.hornetq.core.protocol.core.impl.wireformat.SessionReceiveContinuationMessage; + +/** + * A LargeMessageBufferInternal + * + * @author <a href="mailto:jmesnil@redhat.com">Jeff Mesnil</a> + */ +public interface LargeMessageController extends HornetQBuffer +{ + /** + * Returns the size of this buffer. + + */ + long getSize(); + + /** + * Discards packets unused by this buffer. + */ + void discardUnusedPackets(); + + /** + * Closes this buffer. + */ + void close(); + + /** + * Cancels this buffer. + */ + void cancel(); + + /** + * Sets the OutputStream of this buffer to the specified output. + */ + void setOutputStream(final OutputStream output) throws HornetQException; + + /** + * Saves this buffer to the specified output. + */ + void saveBuffer(final OutputStream output) throws HornetQException; + + public void addPacket(final SessionReceiveContinuationMessage packet); + + /** + * Waits for the completion for the specified waiting time (in milliseconds). + */ + boolean waitCompletion(long timeWait) throws HornetQException; + +} Copied: trunk/src/main/org/hornetq/core/client/impl/LargeMessageControllerImpl.java (from rev 10006, trunk/src/main/org/hornetq/core/client/impl/LargeMessageBufferImpl.java) =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/LargeMessageControllerImpl.java (rev 0) +++ trunk/src/main/org/hornetq/core/client/impl/LargeMessageControllerImpl.java 2010-12-07 22:22:35 UTC (rev 10008) @@ -0,0 +1,1535 @@ +/* + * Copyright 2009 Red Hat, Inc. + * Red Hat 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.hornetq.core.client.impl; + +import java.io.File; +import java.io.FileNotFoundException; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.io.RandomAccessFile; +import java.nio.ByteBuffer; +import java.nio.channels.FileChannel; +import java.nio.channels.GatheringByteChannel; +import java.nio.channels.ScatteringByteChannel; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; + +import org.hornetq.api.core.HornetQBuffer; +import org.hornetq.api.core.HornetQBuffers; +import org.hornetq.api.core.HornetQException; +import org.hornetq.api.core.SimpleString; +import org.hornetq.core.logging.Logger; +import org.hornetq.core.protocol.core.impl.wireformat.SessionReceiveContinuationMessage; +import org.hornetq.utils.DataConstants; +import org.hornetq.utils.UTF8Util; +import org.jboss.netty.buffer.ChannelBuffer; + +/** + * This class aggregates several SessionReceiveContinuationMessages as it was being handled by a single buffer. + * This buffer can be consumed as messages are arriving, and it will hold the packets until they are read using the ChannelBuffer interface, or the setOutputStream or saveStream are called. + * + * @author <a href="mailto:clebert.suconic@jboss.org">Clebert Suconic</a> + * + * + */ +public class LargeMessageControllerImpl implements LargeMessageController +{ + // Constants ----------------------------------------------------- + + private static final String READ_ONLY_ERROR_MESSAGE = "This is a read-only buffer, setOperations are not supported"; + + // Attributes ---------------------------------------------------- + + private static final Logger log = Logger.getLogger(LargeMessageControllerImpl.class); + + private final ClientConsumerInternal consumerInternal; + + private final LinkedBlockingQueue<SessionReceiveContinuationMessage> packets = new LinkedBlockingQueue<SessionReceiveContinuationMessage>(); + + private volatile SessionReceiveContinuationMessage currentPacket = null; + + private final long totalSize; + + private final int bufferSize; + + private boolean streamEnded = false; + + private boolean streamClosed = false; + + private final int readTimeout; + + private long readerIndex = 0; + + private long packetPosition = -1; + + private long lastIndex = 0; + + private long packetLastPosition = -1; + + private OutputStream outStream; + + private Exception handledException; + + private final FileCache fileCache; + + // Static -------------------------------------------------------- + + // Constructors -------------------------------------------------- + + public LargeMessageControllerImpl(final ClientConsumerInternal consumerInternal, + final long totalSize, + final int readTimeout) + { + this(consumerInternal, totalSize, readTimeout, null); + } + + public LargeMessageControllerImpl(final ClientConsumerInternal consumerInternal, + final long totalSize, + final int readTimeout, + final File cachedFile) + { + this(consumerInternal, totalSize, readTimeout, cachedFile, 10 * 1024); + } + + public LargeMessageControllerImpl(final ClientConsumerInternal consumerInternal, + final long totalSize, + final int readTimeout, + final File cachedFile, + final int bufferSize) + { + this.consumerInternal = consumerInternal; + this.readTimeout = readTimeout; + this.totalSize = totalSize; + if (cachedFile == null) + { + fileCache = null; + } + else + { + fileCache = new FileCache(cachedFile); + } + this.bufferSize = bufferSize; + } + + // Public -------------------------------------------------------- + + public synchronized Exception getHandledException() + { + return handledException; + } + + /** + * + */ + public void discardUnusedPackets() + { + if (outStream == null) + { + try + { + checkForPacket(totalSize - 1); + } + catch (Exception ignored) + { + } + } + } + + /** + * Add a buff to the List, or save it to the OutputStream if set + * @param packet + */ + public void addPacket(final SessionReceiveContinuationMessage packet) + { + int flowControlCredit = 0; + boolean continues = false; + + synchronized (this) + { + if (outStream != null) + { + try + { + if (!packet.isContinues()) + { + streamEnded = true; + } + + if (fileCache != null) + { + fileCache.cachePackage(packet.getBody()); + } + + outStream.write(packet.getBody()); + + flowControlCredit = packet.getPacketSize(); + + continues = packet.isContinues(); + + notifyAll(); + + if (streamEnded) + { + outStream.close(); + } + } + catch (Exception e) + { + LargeMessageControllerImpl.log.warn(e.getMessage(), e); + handledException = e; + } + } + else + { + if (fileCache != null) + { + try + { + fileCache.cachePackage(packet.getBody()); + } + catch (Exception e) + { + LargeMessageControllerImpl.log.warn(e.getMessage(), e); + handledException = e; + } + } + + + packets.offer(packet); + } + } + + if (flowControlCredit != 0) + { + try + { + consumerInternal.flowControl(flowControlCredit, !continues); + } + catch (Exception e) + { + LargeMessageControllerImpl.log.warn(e.getMessage(), e); + handledException = e; + } + } + } + + public synchronized void cancel() + { + packets.offer(new SessionReceiveContinuationMessage()); + streamEnded = true; + streamClosed = true; + + notifyAll(); + } + + public synchronized void close() + { + if (fileCache != null) + { + fileCache.close(); + } + } + + public void setOutputStream(final OutputStream output) throws HornetQException + { + + int totalFlowControl = 0; + boolean continues = false; + + synchronized (this) + { + if (currentPacket != null) + { + sendPacketToOutput(output, currentPacket); + currentPacket = null; + } + while (true) + { + SessionReceiveContinuationMessage packet = packets.poll(); + if (packet == null) + { + break; + } + totalFlowControl += packet.getPacketSize(); + + continues = packet.isContinues(); + sendPacketToOutput(output, packet); + } + + outStream = output; + } + + consumerInternal.flowControl(totalFlowControl, !continues); + } + + public synchronized void saveBuffer(final OutputStream output) throws HornetQException + { + setOutputStream(output); + waitCompletion(0); + } + + /** + * + * @param timeWait Milliseconds to Wait. 0 means forever + * @throws Exception + */ + public synchronized boolean waitCompletion(final long timeWait) throws HornetQException + { + if (outStream == null) + { + // There is no stream.. it will never achieve the end of streaming + return false; + } + + long timeOut = System.currentTimeMillis() + timeWait; + while (!streamEnded && handledException == null) + { + try + { + this.wait(readTimeout == 0 ? 1 : readTimeout * 1000); + } + catch (InterruptedException e) + { + throw new HornetQException(HornetQException.INTERNAL_ERROR, e.getMessage(), e); + } + + if (timeWait > 0 && System.currentTimeMillis() > timeOut) + { + throw new HornetQException(HornetQException.LARGE_MESSAGE_ERROR_BODY, + "Timeout waiting for LargeMessage Body"); + } + } + + if (handledException != null) + { + throw new HornetQException(HornetQException.LARGE_MESSAGE_ERROR_BODY, + "Error on saving LargeMessageBufferImpl", + handledException); + } + + return streamEnded; + + } + + // Channel Buffer Implementation --------------------------------- + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#array() + */ + public byte[] array() + { + throw new IllegalAccessError("array not supported on LargeMessageBufferImpl"); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#capacity() + */ + public int capacity() + { + return -1; + } + + public byte readByte() + { + return getByte(readerIndex++); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#getByte(int) + */ + public byte getByte(final int index) + { + return getByte((long)index); + } + + private byte getByte(final long index) + { + checkForPacket(index); + + if (fileCache != null && index < packetPosition) + { + return fileCache.getByteFromCache(index); + } + else + { + return currentPacket.getBody()[(int)(index - packetPosition)]; + } + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, org.hornetq.api.core.buffers.ChannelBuffer, int, int) + */ + public void getBytes(final int index, final HornetQBuffer dst, final int dstIndex, final int length) + { + byte[] destBytes = new byte[length]; + getBytes(index, destBytes); + dst.setBytes(dstIndex, destBytes); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, org.hornetq.api.core.buffers.ChannelBuffer, int, int) + */ + public void getBytes(final long index, final HornetQBuffer dst, final int dstIndex, final int length) + { + byte[] destBytes = new byte[length]; + getBytes(index, destBytes); + dst.setBytes(dstIndex, destBytes); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, byte[], int, int) + */ + public void getBytes(final int index, final byte[] dst, final int dstIndex, final int length) + { + byte bytesToGet[] = new byte[length]; + + getBytes(index, bytesToGet); + + System.arraycopy(bytesToGet, 0, dst, dstIndex, length); + } + + public void getBytes(final long index, final byte[] dst, final int dstIndex, final int length) + { + byte bytesToGet[] = new byte[length]; + + getBytes(index, bytesToGet); + + System.arraycopy(bytesToGet, 0, dst, dstIndex, length); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, java.nio.ByteBuffer) + */ + public void getBytes(final int index, final ByteBuffer dst) + { + byte bytesToGet[] = new byte[dst.remaining()]; + getBytes(index, bytesToGet); + dst.put(bytesToGet); + } + + public void getBytes(final long index, final ByteBuffer dst) + { + byte bytesToGet[] = new byte[dst.remaining()]; + getBytes(index, bytesToGet); + dst.put(bytesToGet); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, java.io.OutputStream, int) + */ + public void getBytes(final int index, final OutputStream out, final int length) throws IOException + { + byte bytesToGet[] = new byte[length]; + getBytes(index, bytesToGet); + out.write(bytesToGet); + } + + public void getBytes(final long index, final OutputStream out, final int length) throws IOException + { + byte bytesToGet[] = new byte[length]; + getBytes(index, bytesToGet); + out.write(bytesToGet); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, java.nio.channels.GatheringByteChannel, int) + */ + public int getBytes(final int index, final GatheringByteChannel out, final int length) throws IOException + { + byte bytesToGet[] = new byte[length]; + getBytes(index, bytesToGet); + return out.write(ByteBuffer.wrap(bytesToGet)); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#getInt(int) + */ + public int getInt(final int index) + { + return (getByte(index) & 0xff) << 24 | (getByte(index + 1) & 0xff) << 16 | + (getByte(index + 2) & 0xff) << 8 | + (getByte(index + 3) & 0xff) << 0; + } + + public int getInt(final long index) + { + return (getByte(index) & 0xff) << 24 | (getByte(index + 1) & 0xff) << 16 | + (getByte(index + 2) & 0xff) << 8 | + (getByte(index + 3) & 0xff) << 0; + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#getLong(int) + */ + public long getLong(final int index) + { + return ((long)getByte(index) & 0xff) << 56 | ((long)getByte(index + 1) & 0xff) << 48 | + ((long)getByte(index + 2) & 0xff) << 40 | + ((long)getByte(index + 3) & 0xff) << 32 | + ((long)getByte(index + 4) & 0xff) << 24 | + ((long)getByte(index + 5) & 0xff) << 16 | + ((long)getByte(index + 6) & 0xff) << 8 | + ((long)getByte(index + 7) & 0xff) << 0; + } + + public long getLong(final long index) + { + return ((long)getByte(index) & 0xff) << 56 | ((long)getByte(index + 1) & 0xff) << 48 | + ((long)getByte(index + 2) & 0xff) << 40 | + ((long)getByte(index + 3) & 0xff) << 32 | + ((long)getByte(index + 4) & 0xff) << 24 | + ((long)getByte(index + 5) & 0xff) << 16 | + ((long)getByte(index + 6) & 0xff) << 8 | + ((long)getByte(index + 7) & 0xff) << 0; + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#getShort(int) + */ + public short getShort(final int index) + { + return (short)(getByte(index) << 8 | getByte(index + 1) & 0xFF); + } + + public short getShort(final long index) + { + return (short)(getByte(index) << 8 | getByte(index + 1) & 0xFF); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#getUnsignedMedium(int) + */ + public int getUnsignedMedium(final int index) + { + return (getByte(index) & 0xff) << 16 | (getByte(index + 1) & 0xff) << 8 | (getByte(index + 2) & 0xff) << 0; + } + + public int getUnsignedMedium(final long index) + { + return (getByte(index) & 0xff) << 16 | (getByte(index + 1) & 0xff) << 8 | (getByte(index + 2) & 0xff) << 0; + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#setByte(int, byte) + */ + public void setByte(final int index, final byte value) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#setBytes(int, org.hornetq.api.core.buffers.ChannelBuffer, int, int) + */ + public void setBytes(final int index, final HornetQBuffer src, final int srcIndex, final int length) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#setBytes(int, byte[], int, int) + */ + public void setBytes(final int index, final byte[] src, final int srcIndex, final int length) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#setBytes(int, java.nio.ByteBuffer) + */ + public void setBytes(final int index, final ByteBuffer src) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#setBytes(int, java.io.InputStream, int) + */ + public int setBytes(final int index, final InputStream in, final int length) throws IOException + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#setBytes(int, java.nio.channels.ScatteringByteChannel, int) + */ + public int setBytes(final int index, final ScatteringByteChannel in, final int length) throws IOException + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#setInt(int, int) + */ + public void setInt(final int index, final int value) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#setLong(int, long) + */ + public void setLong(final int index, final long value) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#setMedium(int, int) + */ + public void setMedium(final int index, final int value) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#setShort(int, short) + */ + public void setShort(final int index, final short value) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#toByteBuffer(int, int) + */ + public ByteBuffer toByteBuffer(final int index, final int length) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#toString(int, int, java.lang.String) + */ + public String toString(final int index, final int length, final String charsetName) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public int readerIndex() + { + return (int)readerIndex; + } + + public void readerIndex(final int readerIndex) + { + try + { + checkForPacket(readerIndex); + } + catch (Exception e) + { + LargeMessageControllerImpl.log.warn(e.getMessage(), e); + throw new RuntimeException(e.getMessage(), e); + } + this.readerIndex = readerIndex; + } + + public int writerIndex() + { + return (int)totalSize; + } + + public long getSize() + { + return totalSize; + } + + public void writerIndex(final int writerIndex) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public void setIndex(final int readerIndex, final int writerIndex) + { + try + { + checkForPacket(readerIndex); + } + catch (Exception e) + { + LargeMessageControllerImpl.log.warn(e.getMessage(), e); + throw new RuntimeException(e.getMessage(), e); + } + this.readerIndex = readerIndex; + } + + public void clear() + { + } + + public boolean readable() + { + return true; + } + + public boolean writable() + { + return false; + } + + public int readableBytes() + { + long readableBytes = totalSize - readerIndex; + + if (readableBytes > Integer.MAX_VALUE) + { + return Integer.MAX_VALUE; + } + else + { + return (int)(totalSize - readerIndex); + } + } + + public int writableBytes() + { + return 0; + } + + public void markReaderIndex() + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public void resetReaderIndex() + { + try + { + checkForPacket(0); + } + catch (Exception e) + { + LargeMessageControllerImpl.log.warn(e.getMessage(), e); + throw new RuntimeException(e.getMessage(), e); + } + } + + public void markWriterIndex() + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public void resetWriterIndex() + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public void discardReadBytes() + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public short getUnsignedByte(final int index) + { + return (short)(getByte(index) & 0xFF); + } + + public int getUnsignedShort(final int index) + { + return getShort(index) & 0xFFFF; + } + + public int getMedium(final int index) + { + int value = getUnsignedMedium(index); + if ((value & 0x800000) != 0) + { + value |= 0xff000000; + } + return value; + } + + public long getUnsignedInt(final int index) + { + return getInt(index) & 0xFFFFFFFFL; + } + + public void getBytes(int index, final byte[] dst) + { + // TODO: optimize this by using System.arraycopy + for (int i = 0; i < dst.length; i++) + { + dst[i] = getByte(index++); + } + } + + public void getBytes(long index, final byte[] dst) + { + // TODO: optimize this by using System.arraycopy + for (int i = 0; i < dst.length; i++) + { + dst[i] = getByte(index++); + } + } + + public void getBytes(final int index, final HornetQBuffer dst) + { + getBytes(index, dst, dst.writableBytes()); + } + + public void getBytes(final int index, final HornetQBuffer dst, final int length) + { + if (length > dst.writableBytes()) + { + throw new IndexOutOfBoundsException(); + } + getBytes(index, dst, dst.writerIndex(), length); + dst.writerIndex(dst.writerIndex() + length); + } + + public void setBytes(final int index, final byte[] src) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public void setBytes(final int index, final HornetQBuffer src) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public void setBytes(final int index, final HornetQBuffer src, final int length) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public void setZero(final int index, final int length) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public short readUnsignedByte() + { + return (short)(readByte() & 0xFF); + } + + public short readShort() + { + short v = getShort(readerIndex); + readerIndex += 2; + return v; + } + + public int readUnsignedShort() + { + return readShort() & 0xFFFF; + } + + public int readMedium() + { + int value = readUnsignedMedium(); + if ((value & 0x800000) != 0) + { + value |= 0xff000000; + } + return value; + } + + public int readUnsignedMedium() + { + int v = getUnsignedMedium(readerIndex); + readerIndex += 3; + return v; + } + + public int readInt() + { + int v = getInt(readerIndex); + readerIndex += 4; + return v; + } + + public int readInt(final int pos) + { + int v = getInt(pos); + return v; + } + + public long readUnsignedInt() + { + return readInt() & 0xFFFFFFFFL; + } + + public long readLong() + { + long v = getLong(readerIndex); + readerIndex += 8; + return v; + } + + public void readBytes(final byte[] dst, final int dstIndex, final int length) + { + getBytes(readerIndex, dst, dstIndex, length); + readerIndex += length; + } + + public void readBytes(final byte[] dst) + { + readBytes(dst, 0, dst.length); + } + + public void readBytes(final HornetQBuffer dst) + { + readBytes(dst, dst.writableBytes()); + } + + public void readBytes(final HornetQBuffer dst, final int length) + { + if (length > dst.writableBytes()) + { + throw new IndexOutOfBoundsException(); + } + readBytes(dst, dst.writerIndex(), length); + dst.writerIndex(dst.writerIndex() + length); + } + + public void readBytes(final HornetQBuffer dst, final int dstIndex, final int length) + { + getBytes(readerIndex, dst, dstIndex, length); + readerIndex += length; + } + + public void readBytes(final ByteBuffer dst) + { + int length = dst.remaining(); + getBytes(readerIndex, dst); + readerIndex += length; + } + + public int readBytes(final GatheringByteChannel out, final int length) throws IOException + { + int readBytes = getBytes((int)readerIndex, out, length); + readerIndex += readBytes; + return readBytes; + } + + public void readBytes(final OutputStream out, final int length) throws IOException + { + getBytes(readerIndex, out, length); + readerIndex += length; + } + + public void skipBytes(final int length) + { + + long newReaderIndex = readerIndex + length; + checkForPacket(newReaderIndex); + readerIndex = newReaderIndex; + } + + public void writeByte(final byte value) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public void writeShort(final short value) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public void writeMedium(final int value) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public void writeInt(final int value) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public void writeLong(final long value) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public void writeBytes(final byte[] src, final int srcIndex, final int length) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public void writeBytes(final byte[] src) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public void writeBytes(final HornetQBuffer src) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public void writeBytes(final HornetQBuffer src, final int length) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public void writeBytes(final ByteBuffer src) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public int writeBytes(final InputStream in, final int length) throws IOException + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public int writeBytes(final ScatteringByteChannel in, final int length) throws IOException + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public void writeZero(final int length) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public ByteBuffer toByteBuffer() + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public ByteBuffer[] toByteBuffers() + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public ByteBuffer[] toByteBuffers(final int index, final int length) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public String toString(final String charsetName) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public Object getUnderlyingBuffer() + { + return this; + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#readBoolean() + */ + public boolean readBoolean() + { + return readByte() != 0; + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#readChar() + */ + public char readChar() + { + return (char)readShort(); + } + + public char getChar(final int index) + { + return (char)getShort(index); + } + + public double getDouble(final int index) + { + return Double.longBitsToDouble(getLong(index)); + } + + public float getFloat(final int index) + { + return Float.intBitsToFloat(getInt(index)); + } + + public HornetQBuffer readBytes(final int length) + { + byte bytesToGet[] = new byte[length]; + getBytes(readerIndex, bytesToGet); + readerIndex += length; + return HornetQBuffers.wrappedBuffer(bytesToGet); + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#readDouble() + */ + public double readDouble() + { + return Double.longBitsToDouble(readLong()); + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#readFloat() + */ + public float readFloat() + { + return Float.intBitsToFloat(readInt()); + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#readNullableSimpleString() + */ + public SimpleString readNullableSimpleString() + { + int b = readByte(); + if (b == DataConstants.NULL) + { + return null; + } + else + { + return readSimpleString(); + } + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#readNullableString() + */ + public String readNullableString() + { + int b = readByte(); + if (b == DataConstants.NULL) + { + return null; + } + else + { + return readString(); + } + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#readSimpleString() + */ + public SimpleString readSimpleString() + { + int len = readInt(); + byte[] data = new byte[len]; + readBytes(data); + return new SimpleString(data); + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#readString() + */ + public String readString() + { + int len = readInt(); + + if (len < 9) + { + char[] chars = new char[len]; + for (int i = 0; i < len; i++) + { + chars[i] = (char)readShort(); + } + return new String(chars); + } + else if (len < 0xfff) + { + return readUTF(); + } + else + { + return readSimpleString().toString(); + } + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#readUTF() + */ + public String readUTF() + { + return UTF8Util.readUTF(this); + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeBoolean(boolean) + */ + public void writeBoolean(final boolean val) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeChar(char) + */ + public void writeChar(final char val) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeDouble(double) + */ + public void writeDouble(final double val) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeFloat(float) + */ + public void writeFloat(final float val) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeNullableSimpleString(org.hornetq.util.SimpleString) + */ + public void writeNullableSimpleString(final SimpleString val) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeNullableString(java.lang.String) + */ + public void writeNullableString(final String val) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeSimpleString(org.hornetq.util.SimpleString) + */ + public void writeSimpleString(final SimpleString val) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeString(java.lang.String) + */ + public void writeString(final String val) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeUTF(java.lang.String) + */ + public void writeUTF(final String utf) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#compareTo(org.hornetq.api.core.buffers.ChannelBuffer) + */ + public int compareTo(final HornetQBuffer buffer) + { + return -1; + } + + public HornetQBuffer copy() + { + throw new UnsupportedOperationException(); + } + + public HornetQBuffer slice(final int index, final int length) + { + throw new UnsupportedOperationException(); + } + + // Package protected --------------------------------------------- + + // Protected ----------------------------------------------------- + + // Private ------------------------------------------------------- + + /** + * @param output + * @param packet + * @throws HornetQException + */ + private void sendPacketToOutput(final OutputStream output, final SessionReceiveContinuationMessage packet) throws HornetQException + { + try + { + output.write(packet.getBody()); + if (!packet.isContinues()) + { + streamEnded = true; + output.close(); + } + } + catch (IOException e) + { + throw new HornetQException(HornetQException.LARGE_MESSAGE_ERROR_BODY, "Error writing body of message", e); + } + } + + private void popPacket() + { + try + { + + if (streamEnded) + { + // no more packets, we are over the last one already + throw new IndexOutOfBoundsException(); + } + + int sizeToAdd = currentPacket != null ? currentPacket.getBody().length : 1; + currentPacket = packets.poll(readTimeout, TimeUnit.SECONDS); + if (currentPacket == null) + { + throw new IndexOutOfBoundsException(); + } + + if (currentPacket.getBody() == null) // Empty packet as a signal to interruption + { + currentPacket = null; + streamEnded = true; + throw new IndexOutOfBoundsException(); + } + + consumerInternal.flowControl(currentPacket.getPacketSize(), !currentPacket.isContinues()); + + packetPosition += sizeToAdd; + + packetLastPosition = packetPosition + currentPacket.getBody().length; + } + catch (IndexOutOfBoundsException e) + { + throw e; + } + catch (Exception e) + { + throw new RuntimeException(e); + } + } + + private void checkForPacket(final long index) + { + if (outStream != null) + { + throw new IllegalAccessError("Can't read the messageBody after setting outputStream"); + } + + if (index >= totalSize) + { + throw new IndexOutOfBoundsException(); + } + + if (streamClosed) + { + throw new IllegalAccessError("The consumer associated with this large message was closed before the body was read"); + } + + if (fileCache == null) + { + if (index < lastIndex) + { + throw new IllegalAccessError("LargeMessage have read-only and one-way buffers"); + } + lastIndex = index; + } + + while (index >= packetLastPosition && !streamEnded) + { + popPacket(); + } + } + + /** + * @param body + */ + // Inner classes ------------------------------------------------- + private class FileCache + { + + public FileCache(final File cachedFile) + { + this.cachedFile = cachedFile; + } + + ByteBuffer readCache; + + long readCachePositionStart = Integer.MAX_VALUE; + + long readCachePositionEnd = -1; + + private final File cachedFile; + + private volatile RandomAccessFile cachedRAFile; + + private volatile FileChannel cachedChannel; + + private synchronized void readCache(final long position) + { + + try + { + if (position < readCachePositionStart || position > readCachePositionEnd) + { + + checkOpen(); + + if (position > cachedChannel.size()) + { + throw new ArrayIndexOutOfBoundsException("position > " + cachedChannel.size()); + } + + readCachePositionStart = position / bufferSize * bufferSize; + + cachedChannel.position(readCachePositionStart); + + if (readCache == null) + { + readCache = ByteBuffer.allocate(bufferSize); + } + + readCache.clear(); + + readCachePositionEnd = readCachePositionStart + cachedChannel.read(readCache) - 1; + } + } + catch (Exception e) + { + LargeMessageControllerImpl.log.warn(e.getMessage(), e); + throw new RuntimeException(e.getMessage(), e); + } + finally + { + close(); + } + } + + public synchronized byte getByteFromCache(final long position) + { + readCache(position); + + return readCache.get((int)(position - readCachePositionStart)); + + } + + public void cachePackage(final byte[] body) throws Exception + { + checkOpen(); + + cachedChannel.position(cachedChannel.size()); + cachedChannel.write(ByteBuffer.wrap(body)); + + close(); + } + + /** + * @throws FileNotFoundException + */ + public void checkOpen() throws FileNotFoundException + { + if (cachedFile != null || !cachedChannel.isOpen()) + { + cachedRAFile = new RandomAccessFile(cachedFile, "rw"); + + cachedChannel = cachedRAFile.getChannel(); + } + } + + public void close() + { + if (cachedChannel != null && cachedChannel.isOpen()) + { + try + { + cachedChannel.close(); + } + catch (Exception e) + { + LargeMessageControllerImpl.log.warn(e.getMessage(), e); + } + cachedChannel = null; + } + + if (cachedRAFile != null) + { + try + { + cachedRAFile.close(); + } + catch (Exception e) + { + LargeMessageControllerImpl.log.warn(e.getMessage(), e); + } + cachedRAFile = null; + } + + } + + @Override + protected void finalize() + { + close(); + if (cachedFile != null && cachedFile.exists()) + { + try + { + cachedFile.delete(); + } + catch (Exception e) + { + LargeMessageControllerImpl.log.warn("Exception during finalization for LargeMessage file cache", e); + } + } + } + + } + + public ChannelBuffer channelBuffer() + { + return null; + } + + public HornetQBuffer copy(final int index, final int length) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public HornetQBuffer duplicate() + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public HornetQBuffer readSlice(final int length) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public void setChar(final int index, final char value) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public void setDouble(final int index, final double value) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public void setFloat(final int index, final float value) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public HornetQBuffer slice() + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public void writeBytes(final HornetQBuffer src, final int srcIndex, final int length) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + +} Modified: trunk/src/main/org/hornetq/core/message/impl/MessageImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/message/impl/MessageImpl.java 2010-12-07 18:05:17 UTC (rev 10007) +++ trunk/src/main/org/hornetq/core/message/impl/MessageImpl.java 2010-12-07 22:22:35 UTC (rev 10008) @@ -24,7 +24,7 @@ import org.hornetq.api.core.PropertyConversionException; import org.hornetq.api.core.SimpleString; import org.hornetq.core.buffers.impl.ResetLimitWrappedHornetQBuffer; -import org.hornetq.core.client.impl.LargeMessageBufferInternal; +import org.hornetq.core.client.impl.LargeMessageController; import org.hornetq.core.logging.Logger; import org.hornetq.core.message.BodyEncoder; import org.hornetq.core.protocol.core.impl.PacketImpl; @@ -245,11 +245,24 @@ properties.decode(buffer); } + public void copyHeadersAndProperties(final MessageInternal msg) + { + messageID = msg.getMessageID(); + address = msg.getAddress(); + userID = msg.getUserID(); + type = msg.getType(); + durable = msg.isDurable(); + expiration = msg.getExpiration(); + timestamp = msg.getTimestamp(); + priority = msg.getPriority(); + properties = msg.getTypedProperties(); + } + public HornetQBuffer getBodyBuffer() { if (bodyBuffer == null) { - if (buffer instanceof LargeMessageBufferInternal == false) + if (buffer instanceof LargeMessageController == false) { bodyBuffer = new ResetLimitWrappedHornetQBuffer(BODY_OFFSET, buffer, this); } @@ -845,6 +858,11 @@ { return new DecodingContext(); } + + public TypedProperties getTypedProperties() + { + return this.properties; + } // Public -------------------------------------------------------- Modified: trunk/src/main/org/hornetq/core/message/impl/MessageInternal.java =================================================================== --- trunk/src/main/org/hornetq/core/message/impl/MessageInternal.java 2010-12-07 18:05:17 UTC (rev 10007) +++ trunk/src/main/org/hornetq/core/message/impl/MessageInternal.java 2010-12-07 22:22:35 UTC (rev 10008) @@ -20,6 +20,7 @@ import org.hornetq.api.core.Message; import org.hornetq.api.core.SimpleString; import org.hornetq.core.message.BodyEncoder; +import org.hornetq.utils.TypedProperties; /** * A MessageInternal @@ -61,4 +62,6 @@ InputStream getBodyInputStream(); void setAddressTransient(SimpleString address); + + TypedProperties getTypedProperties(); } Modified: trunk/src/main/org/hornetq/core/persistence/StorageManager.java =================================================================== --- trunk/src/main/org/hornetq/core/persistence/StorageManager.java 2010-12-07 18:05:17 UTC (rev 10007) +++ trunk/src/main/org/hornetq/core/persistence/StorageManager.java 2010-12-07 22:22:35 UTC (rev 10008) @@ -23,6 +23,7 @@ import org.hornetq.api.core.SimpleString; import org.hornetq.core.journal.IOAsyncTask; import org.hornetq.core.journal.JournalLoadInformation; +import org.hornetq.core.message.impl.MessageInternal; import org.hornetq.core.paging.PageTransactionInfo; import org.hornetq.core.paging.PagedMessage; import org.hornetq.core.paging.PagingManager; @@ -127,7 +128,14 @@ LargeServerMessage createLargeMessage(); - LargeServerMessage createLargeMessage(long id, byte[] header); + /** + * + * @param id + * @param message This is a temporary message that holds the parsed properties. + * The remoting layer can't create a ServerMessage directly, then this will be replaced. + * @return + */ + LargeServerMessage createLargeMessage(long id, MessageInternal message); void prepare(long txID, Xid xid) throws Exception; Modified: trunk/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java =================================================================== --- trunk/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java 2010-12-07 18:05:17 UTC (rev 10007) +++ trunk/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java 2010-12-07 22:22:35 UTC (rev 10008) @@ -47,6 +47,7 @@ import org.hornetq.core.journal.impl.JournalImpl; import org.hornetq.core.journal.impl.NIOSequentialFileFactory; import org.hornetq.core.logging.Logger; +import org.hornetq.core.message.impl.MessageInternal; import org.hornetq.core.paging.PageTransactionInfo; import org.hornetq.core.paging.PagedMessage; import org.hornetq.core.paging.PagingManager; @@ -427,7 +428,7 @@ } } - public LargeServerMessage createLargeMessage(final long id, final byte[] header) + public LargeServerMessage createLargeMessage(final long id, final MessageInternal message) { if (isReplicated()) { @@ -435,11 +436,9 @@ } LargeServerMessageImpl largeMessage = (LargeServerMessageImpl)createLargeMessage(); + + largeMessage.copyHeadersAndProperties(message); - HornetQBuffer headerBuffer = HornetQBuffers.wrappedBuffer(header); - - largeMessage.decodeHeadersAndProperties(headerBuffer); - largeMessage.setMessageID(id); return largeMessage; Modified: trunk/src/main/org/hornetq/core/persistence/impl/nullpm/NullStorageManager.java =================================================================== --- trunk/src/main/org/hornetq/core/persistence/impl/nullpm/NullStorageManager.java 2010-12-07 18:05:17 UTC (rev 10007) +++ trunk/src/main/org/hornetq/core/persistence/impl/nullpm/NullStorageManager.java 2010-12-07 22:22:35 UTC (rev 10008) @@ -27,6 +27,7 @@ import org.hornetq.api.core.SimpleString; import org.hornetq.core.journal.IOAsyncTask; import org.hornetq.core.journal.JournalLoadInformation; +import org.hornetq.core.message.impl.MessageInternal; import org.hornetq.core.paging.PageTransactionInfo; import org.hornetq.core.paging.PagedMessage; import org.hornetq.core.paging.PagingManager; @@ -187,14 +188,12 @@ return new NullStorageLargeServerMessage(); } - public LargeServerMessage createLargeMessage(final long id, final byte[] header) + public LargeServerMessage createLargeMessage(final long id, final MessageInternal message) { NullStorageLargeServerMessage largeMessage = new NullStorageLargeServerMessage(); + + largeMessage.copyHeadersAndProperties(message); - HornetQBuffer headerBuffer = HornetQBuffers.wrappedBuffer(header); - - largeMessage.decodeHeadersAndProperties(headerBuffer); - largeMessage.setMessageID(id); return largeMessage; Modified: trunk/src/main/org/hornetq/core/protocol/core/ServerSessionPacketHandler.java =================================================================== --- trunk/src/main/org/hornetq/core/protocol/core/ServerSessionPacketHandler.java 2010-12-07 18:05:17 UTC (rev 10007) +++ trunk/src/main/org/hornetq/core/protocol/core/ServerSessionPacketHandler.java 2010-12-07 22:22:35 UTC (rev 10008) @@ -451,7 +451,7 @@ case SESS_SEND_LARGE: { SessionSendLargeMessage message = (SessionSendLargeMessage)packet; - session.sendLarge(message.getLargeMessageHeader()); + session.sendLarge(message.getLargeMessage()); break; } case SESS_SEND_CONTINUATION: Modified: trunk/src/main/org/hornetq/core/protocol/core/impl/CoreSessionCallback.java =================================================================== --- trunk/src/main/org/hornetq/core/protocol/core/impl/CoreSessionCallback.java 2010-12-07 18:05:17 UTC (rev 10007) +++ trunk/src/main/org/hornetq/core/protocol/core/impl/CoreSessionCallback.java 2010-12-07 22:22:35 UTC (rev 10008) @@ -49,9 +49,9 @@ this.channel = channel; } - public int sendLargeMessage(long consumerID, byte[] headerBuffer, long bodySize, int deliveryCount) + public int sendLargeMessage(ServerMessage message, long consumerID, long bodySize, int deliveryCount) { - Packet packet = new SessionReceiveLargeMessage(consumerID, headerBuffer, bodySize, deliveryCount); + Packet packet = new SessionReceiveLargeMessage(consumerID, message, bodySize, deliveryCount); channel.send(packet); Modified: trunk/src/main/org/hornetq/core/protocol/core/impl/PacketDecoder.java =================================================================== --- trunk/src/main/org/hornetq/core/protocol/core/impl/PacketDecoder.java 2010-12-07 18:05:17 UTC (rev 10007) +++ trunk/src/main/org/hornetq/core/protocol/core/impl/PacketDecoder.java 2010-12-07 22:22:35 UTC (rev 10008) @@ -84,7 +84,9 @@ import static org.hornetq.core.protocol.core.impl.PacketImpl.SESS_ADD_METADATA; import org.hornetq.api.core.HornetQBuffer; +import org.hornetq.core.client.impl.ClientMessageImpl; import org.hornetq.core.logging.Logger; +import org.hornetq.core.persistence.StorageManager; import org.hornetq.core.protocol.core.Packet; import org.hornetq.core.protocol.core.impl.wireformat.ClusterTopologyChangeMessage; import org.hornetq.core.protocol.core.impl.wireformat.CreateQueueMessage; @@ -161,7 +163,7 @@ public class PacketDecoder { private static final Logger log = Logger.getLogger(PacketDecoder.class); - + public Packet decode(final HornetQBuffer in) { final byte packetType = in.readByte(); @@ -372,6 +374,8 @@ } case SESS_SEND_LARGE: { + // Using a ClientMessage, but that will be replaced later.. + // This is just to avoid reading a byte[] to read the message packet = new SessionSendLargeMessage(); break; } Modified: trunk/src/main/org/hornetq/core/protocol/core/impl/wireformat/SessionReceiveLargeMessage.java =================================================================== --- trunk/src/main/org/hornetq/core/protocol/core/impl/wireformat/SessionReceiveLargeMessage.java 2010-12-07 18:05:17 UTC (rev 10007) +++ trunk/src/main/org/hornetq/core/protocol/core/impl/wireformat/SessionReceiveLargeMessage.java 2010-12-07 22:22:35 UTC (rev 10008) @@ -14,7 +14,9 @@ package org.hornetq.core.protocol.core.impl.wireformat; import org.hornetq.api.core.HornetQBuffer; +import org.hornetq.core.client.impl.ClientMessageImpl; import org.hornetq.core.logging.Logger; +import org.hornetq.core.message.impl.MessageInternal; import org.hornetq.core.protocol.core.impl.PacketImpl; /** @@ -28,7 +30,7 @@ { private static final Logger log = Logger.getLogger(SessionReceiveLargeMessage.class); - private byte[] largeMessageHeader; + private MessageInternal message; /** Since we receive the message before the entire message was received, */ private long largeMessageSize; @@ -37,13 +39,15 @@ private int deliveryCount; + // To be used on decoding at the client while receiving a large message public SessionReceiveLargeMessage() { super(PacketImpl.SESS_RECEIVE_LARGE_MSG); + this.message = new ClientMessageImpl(); } public SessionReceiveLargeMessage(final long consumerID, - final byte[] largeMessageHeader, + final MessageInternal message, final long largeMessageSize, final int deliveryCount) { @@ -51,16 +55,16 @@ this.consumerID = consumerID; - this.largeMessageHeader = largeMessageHeader; + this.message = message; this.deliveryCount = deliveryCount; this.largeMessageSize = largeMessageSize; } - public byte[] getLargeMessageHeader() + public MessageInternal getLargeMessage() { - return largeMessageHeader; + return message; } public long getConsumerID() @@ -87,8 +91,7 @@ buffer.writeLong(consumerID); buffer.writeInt(deliveryCount); buffer.writeLong(largeMessageSize); - buffer.writeInt(largeMessageHeader.length); - buffer.writeBytes(largeMessageHeader); + message.encodeHeadersAndProperties(buffer); } @Override @@ -97,9 +100,7 @@ consumerID = buffer.readLong(); deliveryCount = buffer.readInt(); largeMessageSize = buffer.readLong(); - int size = buffer.readInt(); - largeMessageHeader = new byte[size]; - buffer.readBytes(largeMessageHeader); + message.decodeHeadersAndProperties(buffer); } } Modified: trunk/src/main/org/hornetq/core/protocol/core/impl/wireformat/SessionSendLargeMessage.java =================================================================== --- trunk/src/main/org/hornetq/core/protocol/core/impl/wireformat/SessionSendLargeMessage.java 2010-12-07 18:05:17 UTC (rev 10007) +++ trunk/src/main/org/hornetq/core/protocol/core/impl/wireformat/SessionSendLargeMessage.java 2010-12-07 22:22:35 UTC (rev 10008) @@ -14,7 +14,9 @@ package org.hornetq.core.protocol.core.impl.wireformat; import org.hornetq.api.core.HornetQBuffer; +import org.hornetq.core.message.impl.MessageInternal; import org.hornetq.core.protocol.core.impl.PacketImpl; +import org.hornetq.core.server.impl.ServerMessageImpl; /** * @author <a href="mailto:tim.fox@jboss.com">Tim Fox</a> @@ -30,46 +32,42 @@ // Attributes ---------------------------------------------------- /** Used only if largeMessage */ - private byte[] largeMessageHeader; + private MessageInternal largeMessage; // Static -------------------------------------------------------- // Constructors -------------------------------------------------- - - public SessionSendLargeMessage(final byte[] largeMessageHeader) + + // To be used by the PacketDecoder + public SessionSendLargeMessage() { - super(PacketImpl.SESS_SEND_LARGE); - - this.largeMessageHeader = largeMessageHeader; + this(new ServerMessageImpl()); } - public SessionSendLargeMessage() + public SessionSendLargeMessage(final MessageInternal largeMessage) { super(PacketImpl.SESS_SEND_LARGE); + + this.largeMessage = largeMessage; } // Public -------------------------------------------------------- - public byte[] getLargeMessageHeader() + public MessageInternal getLargeMessage() { - return largeMessageHeader; + return largeMessage; } @Override public void encodeRest(final HornetQBuffer buffer) { - buffer.writeInt(largeMessageHeader.length); - buffer.writeBytes(largeMessageHeader); + largeMessage.encodeHeadersAndProperties(buffer); } @Override public void decodeRest(final HornetQBuffer buffer) { - int largeMessageLength = buffer.readInt(); - - largeMessageHeader = new byte[largeMessageLength]; - - buffer.readBytes(largeMessageHeader); + largeMessage.decodeHeadersAndProperties(buffer); } // Package protected --------------------------------------------- Modified: trunk/src/main/org/hornetq/core/protocol/core/impl/wireformat/SessionSendMessage.java =================================================================== --- trunk/src/main/org/hornetq/core/protocol/core/impl/wireformat/SessionSendMessage.java 2010-12-07 18:05:17 UTC (rev 10007) +++ trunk/src/main/org/hornetq/core/protocol/core/impl/wireformat/SessionSendMessage.java 2010-12-07 22:22:35 UTC (rev 10008) @@ -14,7 +14,6 @@ package org.hornetq.core.protocol.core.impl.wireformat; import org.hornetq.api.core.HornetQBuffer; -import org.hornetq.api.core.Message; import org.hornetq.core.logging.Logger; import org.hornetq.core.message.impl.MessageInternal; import org.hornetq.core.protocol.core.impl.PacketImpl; Modified: trunk/src/main/org/hornetq/core/protocol/stomp/StompSession.java =================================================================== --- trunk/src/main/org/hornetq/core/protocol/stomp/StompSession.java 2010-12-07 18:05:17 UTC (rev 10007) +++ trunk/src/main/org/hornetq/core/protocol/stomp/StompSession.java 2010-12-07 22:22:35 UTC (rev 10008) @@ -149,7 +149,7 @@ return 0; } - public int sendLargeMessage(long consumerID, byte[] headerBuffer, long bodySize, int deliveryCount) + public int sendLargeMessage(ServerMessage msg, long consumerID, long bodySize, int deliveryCount) { return 0; } Modified: trunk/src/main/org/hornetq/core/server/ServerSession.java =================================================================== --- trunk/src/main/org/hornetq/core/server/ServerSession.java 2010-12-07 18:05:17 UTC (rev 10007) +++ trunk/src/main/org/hornetq/core/server/ServerSession.java 2010-12-07 22:22:35 UTC (rev 10008) @@ -19,6 +19,7 @@ import javax.transaction.xa.Xid; import org.hornetq.api.core.SimpleString; +import org.hornetq.core.message.impl.MessageInternal; /** * @@ -102,7 +103,7 @@ void send(ServerMessage message, boolean direct) throws Exception; - void sendLarge(byte[] largeMessageHeader) throws Exception; + void sendLarge(MessageInternal msg) throws Exception; void forceConsumerDelivery(long consumerID, long sequence) throws Exception; Modified: trunk/src/main/org/hornetq/core/server/impl/ServerConsumerImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/ServerConsumerImpl.java 2010-12-07 18:05:17 UTC (rev 10007) +++ trunk/src/main/org/hornetq/core/server/impl/ServerConsumerImpl.java 2010-12-07 22:22:35 UTC (rev 10008) @@ -768,10 +768,6 @@ if (!sentInitialPacket) { - HornetQBuffer headerBuffer = HornetQBuffers.fixedBuffer(largeMessage.getHeadersAndPropertiesEncodeSize()); - - largeMessage.encodeHeadersAndProperties(headerBuffer); - context = largeMessage.getBodyEncoder(); sizePendingLargeMessage = context.getLargeBodySize(); @@ -780,8 +776,8 @@ sentInitialPacket = true; - int packetSize = callback.sendLargeMessage(id, - headerBuffer.toByteBuffer().array(), + int packetSize = callback.sendLargeMessage(largeMessage, + id, context.getLargeBodySize(), ref.getDeliveryCount()); Modified: trunk/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java 2010-12-07 18:05:17 UTC (rev 10007) +++ trunk/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java 2010-12-07 22:22:35 UTC (rev 10008) @@ -37,6 +37,7 @@ import org.hornetq.core.filter.impl.FilterImpl; import org.hornetq.core.journal.IOAsyncTask; import org.hornetq.core.logging.Logger; +import org.hornetq.core.message.impl.MessageInternal; import org.hornetq.core.paging.PagingStore; import org.hornetq.core.persistence.StorageManager; import org.hornetq.core.postoffice.Binding; @@ -965,19 +966,19 @@ consumer.receiveCredits(credits); } - public void sendLarge(final byte[] largeMessageHeader) throws Exception + public void sendLarge(final MessageInternal message) throws Exception { // need to create the LargeMessage before continue long id = storageManager.generateUniqueID(); - - LargeServerMessage msg = storageManager.createLargeMessage(id, largeMessageHeader); - + + LargeServerMessage largeMsg = storageManager.createLargeMessage(id, message); + if (currentLargeMessage != null) { ServerSessionImpl.log.warn("Replacing incomplete LargeMessage with ID=" + currentLargeMessage.getMessageID()); } - currentLargeMessage = msg; + currentLargeMessage = largeMsg; } public void send(final ServerMessage message, final boolean direct) throws Exception Modified: trunk/src/main/org/hornetq/spi/core/protocol/SessionCallback.java =================================================================== --- trunk/src/main/org/hornetq/spi/core/protocol/SessionCallback.java 2010-12-07 18:05:17 UTC (rev 10007) +++ trunk/src/main/org/hornetq/spi/core/protocol/SessionCallback.java 2010-12-07 22:22:35 UTC (rev 10008) @@ -30,7 +30,7 @@ int sendMessage(ServerMessage message, long consumerID, int deliveryCount); - int sendLargeMessage(long consumerID, byte[] headerBuffer, long bodySize, int deliveryCount); + int sendLargeMessage(ServerMessage message, long consumerID, long bodySize, int deliveryCount); int sendLargeMessageContinuation(long consumerID, byte[] body, boolean continues, boolean requiresResponse); Deleted: trunk/src/main/org/hornetq/utils/DecompressedLargeMessageBuffer.java =================================================================== --- trunk/src/main/org/hornetq/utils/DecompressedLargeMessageBuffer.java 2010-12-07 18:05:17 UTC (rev 10007) +++ trunk/src/main/org/hornetq/utils/DecompressedLargeMessageBuffer.java 2010-12-07 22:22:35 UTC (rev 10008) @@ -1,1091 +0,0 @@ -/* - * Copyright 2010 Red Hat, Inc. - * Red Hat 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.hornetq.utils; - -import java.io.DataInputStream; -import java.io.IOException; -import java.io.InputStream; -import java.io.OutputStream; -import java.nio.ByteBuffer; -import java.nio.channels.GatheringByteChannel; -import java.nio.channels.ScatteringByteChannel; -import java.util.concurrent.Executor; - -import org.hornetq.api.core.HornetQBuffer; -import org.hornetq.api.core.HornetQBuffers; -import org.hornetq.api.core.HornetQException; -import org.hornetq.api.core.SimpleString; -import org.hornetq.core.client.impl.LargeMessageBufferInternal; -import org.hornetq.core.protocol.core.impl.wireformat.SessionReceiveContinuationMessage; -import org.jboss.netty.buffer.ChannelBuffer; - -/** - * A DecompressedHornetQBuffer - * - * @author <a href="mailto:clebert.suconic@jboss.org">Clebert Suconic</a> - * - * - */ -public class DecompressedLargeMessageBuffer implements LargeMessageBufferInternal -{ - - // Constants ----------------------------------------------------- - - private static final String OPERATION_NOT_SUPPORTED = "Operation not supported"; - - private static final String READ_ONLY_ERROR_MESSAGE = "This is a read-only buffer, setOperations are not supported"; - - // Attributes ---------------------------------------------------- - - final LargeMessageBufferInternal bufferDelegate; - - // Static -------------------------------------------------------- - - // Constructors -------------------------------------------------- - - public DecompressedLargeMessageBuffer(final LargeMessageBufferInternal bufferDelegate) - { - this.bufferDelegate = bufferDelegate; - } - - - // Public -------------------------------------------------------- - - /** - * - */ - public void discardUnusedPackets() - { - bufferDelegate.discardUnusedPackets(); - } - - /** - * Add a buff to the List, or save it to the OutputStream if set - * @param packet - */ - public void addPacket(final SessionReceiveContinuationMessage packet) - { - bufferDelegate.addPacket(packet); - } - - public synchronized void cancel() - { - bufferDelegate.cancel(); - } - - public synchronized void close() - { - bufferDelegate.cancel(); - } - - public void setOutputStream(final OutputStream output) throws HornetQException - { - bufferDelegate.setOutputStream(new InflaterWriter(output)); - } - - public synchronized void saveBuffer(final OutputStream output) throws HornetQException - { - setOutputStream(output); - waitCompletion(0); - } - - /** - * - * @param timeWait Milliseconds to Wait. 0 means forever - * @throws Exception - */ - public synchronized boolean waitCompletion(final long timeWait) throws HornetQException - { - return bufferDelegate.waitCompletion(timeWait); - } - - // Channel Buffer Implementation --------------------------------- - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#array() - */ - public byte[] array() - { - throw new IllegalAccessError("array not supported on LargeMessageBufferImpl"); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#capacity() - */ - public int capacity() - { - return -1; - } - - DataInputStream dataInput = null; - - private DataInputStream getStream() - { - if (dataInput == null) - { - try - { - InputStream input = new HornetQBufferInputStream(bufferDelegate); - - dataInput = new DataInputStream(new InflaterReader(input)); - } - catch (Exception e) - { - throw new RuntimeException (e.getMessage(), e); - } - - } - return dataInput; - } - - private void positioningNotSupported() - { - throw new IllegalStateException("Position not supported over compressed large messages"); - } - - public byte readByte() - { - try - { - return getStream().readByte(); - } - catch (Exception e) - { - throw new RuntimeException (e.getMessage(), e); - } - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#getByte(int) - */ - public byte getByte(final int index) - { - positioningNotSupported(); - return 0; - } - - private byte getByte(final long index) - { - positioningNotSupported(); - return 0; - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, org.hornetq.api.core.buffers.ChannelBuffer, int, int) - */ - public void getBytes(final int index, final HornetQBuffer dst, final int dstIndex, final int length) - { - positioningNotSupported(); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, org.hornetq.api.core.buffers.ChannelBuffer, int, int) - */ - public void getBytes(final long index, final HornetQBuffer dst, final int dstIndex, final int length) - { - positioningNotSupported(); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, byte[], int, int) - */ - public void getBytes(final int index, final byte[] dst, final int dstIndex, final int length) - { - positioningNotSupported(); - } - - public void getBytes(final long index, final byte[] dst, final int dstIndex, final int length) - { - positioningNotSupported(); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, java.nio.ByteBuffer) - */ - public void getBytes(final int index, final ByteBuffer dst) - { - positioningNotSupported(); - } - - public void getBytes(final long index, final ByteBuffer dst) - { - positioningNotSupported(); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, java.io.OutputStream, int) - */ - public void getBytes(final int index, final OutputStream out, final int length) throws IOException - { - positioningNotSupported(); - } - - public void getBytes(final long index, final OutputStream out, final int length) throws IOException - { - positioningNotSupported(); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, java.nio.channels.GatheringByteChannel, int) - */ - public int getBytes(final int index, final GatheringByteChannel out, final int length) throws IOException - { - positioningNotSupported(); - return 0; - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#getInt(int) - */ - public int getInt(final int index) - { - positioningNotSupported(); - return 0; - } - - public int getInt(final long index) - { - positioningNotSupported(); - return 0; - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#getLong(int) - */ - public long getLong(final int index) - { - positioningNotSupported(); - return 0; - } - - public long getLong(final long index) - { - positioningNotSupported(); - return 0; - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#getShort(int) - */ - public short getShort(final int index) - { - positioningNotSupported(); - return 0; - } - - public short getShort(final long index) - { - return (short)(getByte(index) << 8 | getByte(index + 1) & 0xFF); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#getUnsignedMedium(int) - */ - public int getUnsignedMedium(final int index) - { - positioningNotSupported(); - return 0; - } - - - - public int getUnsignedMedium(final long index) - { - positioningNotSupported(); - return 0; - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#setByte(int, byte) - */ - public void setByte(final int index, final byte value) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#setBytes(int, org.hornetq.api.core.buffers.ChannelBuffer, int, int) - */ - public void setBytes(final int index, final HornetQBuffer src, final int srcIndex, final int length) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#setBytes(int, byte[], int, int) - */ - public void setBytes(final int index, final byte[] src, final int srcIndex, final int length) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#setBytes(int, java.nio.ByteBuffer) - */ - public void setBytes(final int index, final ByteBuffer src) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#setBytes(int, java.io.InputStream, int) - */ - public int setBytes(final int index, final InputStream in, final int length) throws IOException - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#setBytes(int, java.nio.channels.ScatteringByteChannel, int) - */ - public int setBytes(final int index, final ScatteringByteChannel in, final int length) throws IOException - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#setInt(int, int) - */ - public void setInt(final int index, final int value) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#setLong(int, long) - */ - public void setLong(final int index, final long value) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#setMedium(int, int) - */ - public void setMedium(final int index, final int value) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#setShort(int, short) - */ - public void setShort(final int index, final short value) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#toByteBuffer(int, int) - */ - public ByteBuffer toByteBuffer(final int index, final int length) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#toString(int, int, java.lang.String) - */ - public String toString(final int index, final int length, final String charsetName) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public int readerIndex() - { - return 0; - } - - public void readerIndex(final int readerIndex) - { - // TODO - } - - public int writerIndex() - { - // TODO - return 0; - } - - public long getSize() - { - // TODO - return 0; - } - - public void writerIndex(final int writerIndex) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void setIndex(final int readerIndex, final int writerIndex) - { - positioningNotSupported(); - } - - public void clear() - { - } - - public boolean readable() - { - return true; - } - - public boolean writable() - { - return false; - } - - public int readableBytes() - { - return 1; - } - - public int writableBytes() - { - return 0; - } - - public void markReaderIndex() - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void resetReaderIndex() - { - // TODO: reset positioning if possible - } - - public void markWriterIndex() - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void resetWriterIndex() - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void discardReadBytes() - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public short getUnsignedByte(final int index) - { - return (short)(getByte(index) & 0xFF); - } - - public int getUnsignedShort(final int index) - { - return getShort(index) & 0xFFFF; - } - - public int getMedium(final int index) - { - int value = getUnsignedMedium(index); - if ((value & 0x800000) != 0) - { - value |= 0xff000000; - } - return value; - } - - public long getUnsignedInt(final int index) - { - return getInt(index) & 0xFFFFFFFFL; - } - - public void getBytes(int index, final byte[] dst) - { - // TODO: optimize this by using System.arraycopy - for (int i = 0; i < dst.length; i++) - { - dst[i] = getByte(index++); - } - } - - public void getBytes(long index, final byte[] dst) - { - // TODO: optimize this by using System.arraycopy - for (int i = 0; i < dst.length; i++) - { - dst[i] = getByte(index++); - } - } - - public void getBytes(final int index, final HornetQBuffer dst) - { - getBytes(index, dst, dst.writableBytes()); - } - - public void getBytes(final int index, final HornetQBuffer dst, final int length) - { - if (length > dst.writableBytes()) - { - throw new IndexOutOfBoundsException(); - } - getBytes(index, dst, dst.writerIndex(), length); - dst.writerIndex(dst.writerIndex() + length); - } - - public void setBytes(final int index, final byte[] src) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void setBytes(final int index, final HornetQBuffer src) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void setBytes(final int index, final HornetQBuffer src, final int length) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void setZero(final int index, final int length) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public short readUnsignedByte() - { - try - { - return (short)getStream().readUnsignedByte(); - } - catch (Exception e) - { - throw new IllegalStateException (e.getMessage(), e); - } - } - - public short readShort() - { - try - { - return (short)getStream().readShort(); - } - catch (Exception e) - { - throw new IllegalStateException (e.getMessage(), e); - } - } - - public int readUnsignedShort() - { - try - { - return (int)getStream().readUnsignedShort(); - } - catch (Exception e) - { - throw new IllegalStateException (e.getMessage(), e); - } - } - - public int readMedium() - { - int value = readUnsignedMedium(); - if ((value & 0x800000) != 0) - { - value |= 0xff000000; - } - return value; - } - - - public int readUnsignedMedium() - { - return (readByte() & 0xff) << 16 | (readByte() & 0xff) << 8 | (readByte() & 0xff) << 0; - } - - public int readInt() - { - try - { - return getStream().readInt(); - } - catch (Exception e) - { - throw new IllegalStateException(e.getMessage(), e); - } - } - - public int readInt(final int pos) - { - positioningNotSupported(); - return 0; - } - - public long readUnsignedInt() - { - return readInt() & 0xFFFFFFFFL; - } - - public long readLong() - { - try - { - return getStream().readLong(); - } - catch (Exception e) - { - throw new IllegalStateException(e.getMessage(), e); - } - } - - public void readBytes(final byte[] dst, final int dstIndex, final int length) - { - try - { - getStream().read(dst, dstIndex, length); - } - catch (Exception e) - { - throw new IllegalStateException(e.getMessage(), e); - } - } - - public void readBytes(final byte[] dst) - { - readBytes(dst, 0, dst.length); - } - - public void readBytes(final HornetQBuffer dst) - { - readBytes(dst, dst.writableBytes()); - } - - public void readBytes(final HornetQBuffer dst, final int length) - { - if (length > dst.writableBytes()) - { - throw new IndexOutOfBoundsException(); - } - readBytes(dst, dst.writerIndex(), length); - dst.writerIndex(dst.writerIndex() + length); - } - - public void readBytes(final HornetQBuffer dst, final int dstIndex, final int length) - { - byte[] destBytes = new byte[length]; - readBytes(destBytes); - dst.setBytes(dstIndex, destBytes); - } - - public void readBytes(final ByteBuffer dst) - { - byte bytesToGet[] = new byte[dst.remaining()]; - readBytes(bytesToGet); - dst.put(bytesToGet); - } - - public int readBytes(final GatheringByteChannel out, final int length) throws IOException - { - throw new IllegalStateException("Not implemented!"); - } - - public void readBytes(final OutputStream out, final int length) throws IOException - { - throw new IllegalStateException("Not implemented!"); - } - - public void skipBytes(final int length) - { - - try - { - for (int i = 0 ; i < length; i++) - { - getStream().read(); - } - } - catch (Exception e) - { - throw new IllegalStateException(e.getMessage(), e); - } - } - - public void writeByte(final byte value) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void writeShort(final short value) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void writeMedium(final int value) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void writeInt(final int value) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void writeLong(final long value) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void writeBytes(final byte[] src, final int srcIndex, final int length) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void writeBytes(final byte[] src) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void writeBytes(final HornetQBuffer src) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void writeBytes(final HornetQBuffer src, final int length) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void writeBytes(final ByteBuffer src) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public int writeBytes(final InputStream in, final int length) throws IOException - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public int writeBytes(final ScatteringByteChannel in, final int length) throws IOException - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void writeZero(final int length) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public ByteBuffer toByteBuffer() - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public ByteBuffer[] toByteBuffers() - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public ByteBuffer[] toByteBuffers(final int index, final int length) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public String toString(final String charsetName) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public Object getUnderlyingBuffer() - { - return this; - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#readBoolean() - */ - public boolean readBoolean() - { - return readByte() != 0; - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#readChar() - */ - public char readChar() - { - return (char)readShort(); - } - - public char getChar(final int index) - { - return (char)getShort(index); - } - - public double getDouble(final int index) - { - return Double.longBitsToDouble(getLong(index)); - } - - public float getFloat(final int index) - { - return Float.intBitsToFloat(getInt(index)); - } - - public HornetQBuffer readBytes(final int length) - { - byte bytesToGet[] = new byte[length]; - readBytes(bytesToGet); - return HornetQBuffers.wrappedBuffer(bytesToGet); - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#readDouble() - */ - public double readDouble() - { - return Double.longBitsToDouble(readLong()); - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#readFloat() - */ - public float readFloat() - { - return Float.intBitsToFloat(readInt()); - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#readNullableSimpleString() - */ - public SimpleString readNullableSimpleString() - { - int b = readByte(); - if (b == DataConstants.NULL) - { - return null; - } - else - { - return readSimpleString(); - } - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#readNullableString() - */ - public String readNullableString() - { - int b = readByte(); - if (b == DataConstants.NULL) - { - return null; - } - else - { - return readString(); - } - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#readSimpleString() - */ - public SimpleString readSimpleString() - { - int len = readInt(); - byte[] data = new byte[len]; - readBytes(data); - return new SimpleString(data); - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#readString() - */ - public String readString() - { - int len = readInt(); - - if (len < 9) - { - char[] chars = new char[len]; - for (int i = 0; i < len; i++) - { - chars[i] = (char)readShort(); - } - return new String(chars); - } - else if (len < 0xfff) - { - return readUTF(); - } - else - { - return readSimpleString().toString(); - } - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#readUTF() - */ - public String readUTF() - { - return UTF8Util.readUTF(this); - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeBoolean(boolean) - */ - public void writeBoolean(final boolean val) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeChar(char) - */ - public void writeChar(final char val) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeDouble(double) - */ - public void writeDouble(final double val) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeFloat(float) - */ - public void writeFloat(final float val) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeNullableSimpleString(org.hornetq.util.SimpleString) - */ - public void writeNullableSimpleString(final SimpleString val) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeNullableString(java.lang.String) - */ - public void writeNullableString(final String val) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeSimpleString(org.hornetq.util.SimpleString) - */ - public void writeSimpleString(final SimpleString val) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeString(java.lang.String) - */ - public void writeString(final String val) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeUTF(java.lang.String) - */ - public void writeUTF(final String utf) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#compareTo(org.hornetq.api.core.buffers.ChannelBuffer) - */ - public int compareTo(final HornetQBuffer buffer) - { - return -1; - } - - public HornetQBuffer copy() - { - throw new UnsupportedOperationException(); - } - - public HornetQBuffer slice(final int index, final int length) - { - throw new UnsupportedOperationException(); - } - - // Package protected --------------------------------------------- - - // Protected ----------------------------------------------------- - - // Private ------------------------------------------------------- - - /** - * @param body - */ - // Inner classes ------------------------------------------------- - - public ChannelBuffer channelBuffer() - { - return null; - } - - public HornetQBuffer copy(final int index, final int length) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public HornetQBuffer duplicate() - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public HornetQBuffer readSlice(final int length) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void setChar(final int index, final char value) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void setDouble(final int index, final double value) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void setFloat(final int index, final float value) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public HornetQBuffer slice() - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void writeBytes(final HornetQBuffer src, final int srcIndex, final int length) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } -} Modified: trunk/tests/src/org/hornetq/tests/unit/core/client/impl/LargeMessageBufferTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/unit/core/client/impl/LargeMessageBufferTest.java 2010-12-07 18:05:17 UTC (rev 10007) +++ trunk/tests/src/org/hornetq/tests/unit/core/client/impl/LargeMessageBufferTest.java 2010-12-07 22:22:35 UTC (rev 10008) @@ -37,7 +37,7 @@ import org.hornetq.core.client.impl.ClientConsumerInternal; import org.hornetq.core.client.impl.ClientMessageInternal; import org.hornetq.core.client.impl.ClientSessionInternal; -import org.hornetq.core.client.impl.LargeMessageBufferImpl; +import org.hornetq.core.client.impl.LargeMessageControllerImpl; import org.hornetq.core.protocol.core.impl.wireformat.SessionQueueQueryResponseMessage; import org.hornetq.core.protocol.core.impl.wireformat.SessionReceiveContinuationMessage; import org.hornetq.core.protocol.core.impl.wireformat.SessionReceiveLargeMessage; @@ -85,7 +85,7 @@ // Test Simple getBytes public void testGetBytes() throws Exception { - LargeMessageBufferImpl buffer = create15BytesSample(); + LargeMessageControllerImpl buffer = create15BytesSample(); for (int i = 1; i <= 15; i++) { @@ -112,7 +112,7 @@ // Test for void getBytes(final int index, final byte[] dst) public void testGetBytesIByteArray() throws Exception { - LargeMessageBufferImpl buffer = create15BytesSample(); + LargeMessageControllerImpl buffer = create15BytesSample(); byte[] bytes = new byte[15]; buffer.getBytes(0, bytes); @@ -135,7 +135,7 @@ // testing void getBytes(int index, ChannelBuffer dst, int dstIndex, int length) public void testGetBytesILChannelBufferII() throws Exception { - LargeMessageBufferImpl buffer = create15BytesSample(); + LargeMessageControllerImpl buffer = create15BytesSample(); HornetQBuffer dstBuffer = HornetQBuffers.fixedBuffer(20); @@ -152,7 +152,7 @@ // testing void getBytes(int index, ChannelBuffer dst, int dstIndex, int length) public void testReadIntegers() throws Exception { - LargeMessageBufferImpl buffer = createBufferWithIntegers(3, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15); + LargeMessageControllerImpl buffer = createBufferWithIntegers(3, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15); for (int i = 1; i <= 15; i++) { @@ -171,7 +171,7 @@ public void testReadIntegersOverStream() throws Exception { - LargeMessageBufferImpl buffer = createBufferWithIntegers(3, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15); + LargeMessageControllerImpl buffer = createBufferWithIntegers(3, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15); HornetQBufferInputStream is = new HornetQBufferInputStream(buffer); DataInputStream dataInput = new DataInputStream(is); @@ -186,7 +186,7 @@ // testing void getBytes(int index, ChannelBuffer dst, int dstIndex, int length) public void testReadLongs() throws Exception { - LargeMessageBufferImpl buffer = createBufferWithLongs(3, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15); + LargeMessageControllerImpl buffer = createBufferWithLongs(3, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15); for (int i = 1; i <= 15; i++) { @@ -205,7 +205,7 @@ public void testReadLongsOverStream() throws Exception { - LargeMessageBufferImpl buffer = createBufferWithLongs(3, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15); + LargeMessageControllerImpl buffer = createBufferWithLongs(3, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15); HornetQBufferInputStream is = new HornetQBufferInputStream(buffer); DataInputStream dataInput = new DataInputStream(is); @@ -231,7 +231,7 @@ dynamic.writeDouble(d1); dynamic.writeFloat(f1); - LargeMessageBufferImpl readBuffer = splitBuffer(3, dynamic.toByteBuffer().array()); + LargeMessageControllerImpl readBuffer = splitBuffer(3, dynamic.toByteBuffer().array()); Assert.assertEquals(str1, readBuffer.readUTF()); Assert.assertEquals(str2, readBuffer.readString()); @@ -260,7 +260,7 @@ dynamic.writeDouble(d1); dynamic.writeFloat(f1); - LargeMessageBufferImpl readBuffer = splitBuffer(3, dynamic.toByteBuffer().array(), getTestFile()); + LargeMessageControllerImpl readBuffer = splitBuffer(3, dynamic.toByteBuffer().array(), getTestFile()); Assert.assertEquals(str1, readBuffer.readUTF()); Assert.assertEquals(str2, readBuffer.readString()); @@ -280,7 +280,7 @@ public void testReadPartialData() throws Exception { - final LargeMessageBufferImpl buffer = new LargeMessageBufferImpl(new FakeConsumerInternal(), 10, 10); + final LargeMessageControllerImpl buffer = new LargeMessageControllerImpl(new FakeConsumerInternal(), 10, 10); buffer.addPacket(new FakePacket(-1, new byte[] { 0, 1, 2, 3, 4 }, true, true)); @@ -336,7 +336,7 @@ public void testInterruptData() throws Exception { - LargeMessageBufferImpl readBuffer = splitBuffer(3, new byte[] { 0, 1, 2, 3, 4 }); + LargeMessageControllerImpl readBuffer = splitBuffer(3, new byte[] { 0, 1, 2, 3, 4 }); byte bytes[] = new byte[30]; readBuffer.readBytes(bytes, 0, 5); @@ -349,7 +349,7 @@ public void testSplitBufferOnFile() throws Exception { - LargeMessageBufferImpl outBuffer = new LargeMessageBufferImpl(new FakeConsumerInternal(), + LargeMessageControllerImpl outBuffer = new LargeMessageControllerImpl(new FakeConsumerInternal(), 1024 * 1024, 1, getTestFile(), @@ -391,7 +391,7 @@ public void testStreamData() throws Exception { - final LargeMessageBufferImpl outBuffer = new LargeMessageBufferImpl(new FakeConsumerInternal(), + final LargeMessageControllerImpl outBuffer = new LargeMessageControllerImpl(new FakeConsumerInternal(), 1024 * 11 + 123, 1); @@ -509,7 +509,7 @@ public void testStreamDataWaitCompletionOnCompleteBuffer() throws Exception { - final LargeMessageBufferImpl outBuffer = create15BytesSample(); + final LargeMessageControllerImpl outBuffer = create15BytesSample(); outBuffer.saveBuffer(new OutputStream() { @@ -524,7 +524,7 @@ public void testErrorOnSetStreaming() throws Exception { long start = System.currentTimeMillis(); - final LargeMessageBufferImpl outBuffer = new LargeMessageBufferImpl(new FakeConsumerInternal(), 5, 30); + final LargeMessageControllerImpl outBuffer = new LargeMessageControllerImpl(new FakeConsumerInternal(), 5, 30); outBuffer.addPacket(new FakePacket(-1, new byte[] { 0, 1, 2, 3, 4 }, true, false)); @@ -589,12 +589,12 @@ /** * @return */ - private LargeMessageBufferImpl create15BytesSample() throws Exception + private LargeMessageControllerImpl create15BytesSample() throws Exception { return splitBuffer(5, new byte[] { 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15 }); } - private LargeMessageBufferImpl createBufferWithIntegers(final int splitFactor, final int... values) throws Exception + private LargeMessageControllerImpl createBufferWithIntegers(final int splitFactor, final int... values) throws Exception { ByteArrayOutputStream byteOut = new ByteArrayOutputStream(values.length * 4); DataOutputStream dataOut = new DataOutputStream(byteOut); @@ -607,7 +607,7 @@ return splitBuffer(splitFactor, byteOut.toByteArray()); } - private LargeMessageBufferImpl createBufferWithLongs(final int splitFactor, final long... values) throws Exception + private LargeMessageControllerImpl createBufferWithLongs(final int splitFactor, final long... values) throws Exception { ByteArrayOutputStream byteOut = new ByteArrayOutputStream(values.length * 8); DataOutputStream dataOut = new DataOutputStream(byteOut); @@ -620,14 +620,14 @@ return splitBuffer(splitFactor, byteOut.toByteArray()); } - private LargeMessageBufferImpl splitBuffer(final int splitFactor, final byte[] bytes) throws Exception + private LargeMessageControllerImpl splitBuffer(final int splitFactor, final byte[] bytes) throws Exception { return splitBuffer(splitFactor, bytes, null); } - private LargeMessageBufferImpl splitBuffer(final int splitFactor, final byte[] bytes, final File file) throws Exception + private LargeMessageControllerImpl splitBuffer(final int splitFactor, final byte[] bytes, final File file) throws Exception { - LargeMessageBufferImpl outBuffer = new LargeMessageBufferImpl(new FakeConsumerInternal(), bytes.length, 5, file); + LargeMessageControllerImpl outBuffer = new LargeMessageControllerImpl(new FakeConsumerInternal(), bytes.length, 5, file); ByteArrayInputStream input = new ByteArrayInputStream(bytes); Modified: trunk/tests/src/org/hornetq/tests/unit/core/paging/impl/PagingStoreImplTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/unit/core/paging/impl/PagingStoreImplTest.java 2010-12-07 18:05:17 UTC (rev 10007) +++ trunk/tests/src/org/hornetq/tests/unit/core/paging/impl/PagingStoreImplTest.java 2010-12-07 22:22:35 UTC (rev 10008) @@ -40,6 +40,7 @@ import org.hornetq.core.journal.SequentialFile; import org.hornetq.core.journal.SequentialFileFactory; import org.hornetq.core.journal.impl.NIOSequentialFileFactory; +import org.hornetq.core.message.impl.MessageInternal; import org.hornetq.core.paging.Page; import org.hornetq.core.paging.PageTransactionInfo; import org.hornetq.core.paging.PagedMessage; @@ -75,7 +76,6 @@ import org.hornetq.tests.util.RandomUtil; import org.hornetq.tests.util.UnitTestCase; import org.hornetq.utils.ExecutorFactory; -import org.hornetq.utils.UUID; /** * @@ -1387,7 +1387,7 @@ /* (non-Javadoc) * @see org.hornetq.core.persistence.StorageManager#createLargeMessage(byte[]) */ - public LargeServerMessage createLargeMessage(final long messageId, final byte[] header) + public LargeServerMessage createLargeMessage(final long messageId, final MessageInternal msg) { return null; 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-12-07 18:05:17 UTC (rev 10007) +++ trunk/tests/src/org/hornetq/tests/unit/core/postoffice/impl/BindingsImplTest.java 2010-12-07 22:22:35 UTC (rev 10008) @@ -983,6 +983,15 @@ return false; } + /* (non-Javadoc) + * @see org.hornetq.core.message.impl.MessageInternal#getTypedProperties() + */ + public TypedProperties getTypedProperties() + { + // TODO Auto-generated method stub + return null; + } + } class FakeFilter implements Filter

15 years, 5 months

1
0
0 / 0

JBoss hornetq SVN: r10007 - trunk/src/main/org/hornetq/core/client/impl.

by do-not-reply＠jboss.org

Author: clebert.suconic(a)jboss.com Date: 2010-12-07 13:05:17 -0500 (Tue, 07 Dec 2010) New Revision: 10007 Removed: trunk/src/main/org/hornetq/core/client/impl/FailoverManagerImpl.java trunk/src/main/org/hornetq/core/client/impl/FailoverManager_Old.java Log: Removing unused classes Deleted: trunk/src/main/org/hornetq/core/client/impl/FailoverManagerImpl.java =================================================================== Deleted: trunk/src/main/org/hornetq/core/client/impl/FailoverManager_Old.java =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/FailoverManager_Old.java 2010-12-07 12:35:50 UTC (rev 10006) +++ trunk/src/main/org/hornetq/core/client/impl/FailoverManager_Old.java 2010-12-07 18:05:17 UTC (rev 10007) @@ -1,66 +0,0 @@ -/* - * Copyright 2009 Red Hat, Inc. - * Red Hat 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.hornetq.core.client.impl; - -import org.hornetq.api.core.HornetQException; -import org.hornetq.api.core.client.ClientSession; -import org.hornetq.api.core.client.SessionFailureListener; -import org.hornetq.core.protocol.core.CoreRemotingConnection; - -/** - * A ConnectionManager - * - * @author <a href="mailto:tim.fox@jboss.com">Tim Fox</a> - * - * Created 27 Nov 2008 18:45:46 - * - * - */ -public interface FailoverManager_Old -{ - ClientSession createSession(final String username, - final String password, - final boolean xa, - final boolean autoCommitSends, - final boolean autoCommitAcks, - final boolean preAcknowledge, - final int ackBatchSize, - final boolean cacheLargeMessageClient, - final int minLargeMessageSize, - final boolean blockOnAcknowledge, - final boolean autoGroup, - final int confirmationWindowSize, - final int producerWindowSize, - final int consumerWindowSize, - final int producerMaxRate, - final int consumerMaxRate, - final boolean blockOnNonDurableSend, - final boolean blockOnDurableSend, - final int initialMessagePacketSize, - final String groupID) throws HornetQException; - - void removeSession(final ClientSessionInternal session); - - public CoreRemotingConnection getConnection(); - - int numConnections(); - - int numSessions(); - - void addFailureListener(SessionFailureListener listener); - - boolean removeFailureListener(SessionFailureListener listener); - - void causeExit(); -}

15 years, 5 months

1
0
0 / 0

JBoss hornetq SVN: r10006 - in trunk/src/main/org/hornetq/core: server/impl and 1 other directory.

by do-not-reply＠jboss.org

Author: ataylor Date: 2010-12-07 07:35:50 -0500 (Tue, 07 Dec 2010) New Revision: 10006 Modified: trunk/src/main/org/hornetq/core/remoting/server/impl/RemotingServiceImpl.java trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java Log: renamed parameters Modified: trunk/src/main/org/hornetq/core/remoting/server/impl/RemotingServiceImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/remoting/server/impl/RemotingServiceImpl.java 2010-12-07 04:33:00 UTC (rev 10005) +++ trunk/src/main/org/hornetq/core/remoting/server/impl/RemotingServiceImpl.java 2010-12-07 12:35:50 UTC (rev 10006) @@ -257,7 +257,7 @@ stop(false); } - public void stop(boolean clientFailover) throws Exception + public void stop(boolean failoverOnServerShutdown) throws Exception { if (!started) { @@ -283,7 +283,7 @@ { RemotingConnection conn = entry.connection; - conn.disconnect(clientFailover); + conn.disconnect(failoverOnServerShutdown); } for (Acceptor acceptor : acceptors) Modified: trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2010-12-07 04:33:00 UTC (rev 10005) +++ trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2010-12-07 12:35:50 UTC (rev 10006) @@ -591,7 +591,7 @@ stop(false); } - public void stop(boolean permanently) throws Exception + public void stop(boolean failoverOnServerShutdown) throws Exception { System.out.println("*** stop called on server"); @@ -624,7 +624,7 @@ { System.out.println("HornetQServerImpl.stop"); } - remotingService.stop(permanently); + remotingService.stop(failoverOnServerShutdown); synchronized (this) { @@ -728,7 +728,7 @@ if (activation != null) { - activation.close(permanently); + activation.close(failoverOnServerShutdown); } if (backupActivationThread != null)

15 years, 5 months

1
0
0 / 0

JBoss hornetq SVN: r10005 - trunk/src/main/org/hornetq/utils.

by do-not-reply＠jboss.org

Author: clebert.suconic(a)jboss.com Date: 2010-12-06 23:33:00 -0500 (Mon, 06 Dec 2010) New Revision: 10005 Modified: trunk/src/main/org/hornetq/utils/InflaterWriter.java Log: just cleanup imports Modified: trunk/src/main/org/hornetq/utils/InflaterWriter.java =================================================================== --- trunk/src/main/org/hornetq/utils/InflaterWriter.java 2010-12-07 04:22:51 UTC (rev 10004) +++ trunk/src/main/org/hornetq/utils/InflaterWriter.java 2010-12-07 04:33:00 UTC (rev 10005) @@ -13,12 +13,9 @@ package org.hornetq.utils; -import java.io.ByteArrayInputStream; -import java.io.ByteArrayOutputStream; import java.io.IOException; import java.io.OutputStream; import java.util.zip.DataFormatException; -import java.util.zip.Deflater; import java.util.zip.Inflater; /** @@ -35,15 +32,17 @@ */ public class InflaterWriter extends OutputStream { - private Inflater inflater = new Inflater(); - private OutputStream output; - - private byte[] writeBuffer = new byte[1024]; + private final Inflater inflater = new Inflater(); + + private final OutputStream output; + + private final byte[] writeBuffer = new byte[1024]; + private int writePointer = 0; - - private byte[] outputBuffer = new byte[writeBuffer.length*2]; - - public InflaterWriter(OutputStream output) + + private final byte[] outputBuffer = new byte[writeBuffer.length * 2]; + + public InflaterWriter(final OutputStream output) { this.output = output; } @@ -52,11 +51,11 @@ * Write a compressed byte. */ @Override - public void write(int b) throws IOException + public void write(final int b) throws IOException { writeBuffer[writePointer] = (byte)(b & 0xFF); writePointer++; - + if (writePointer == writeBuffer.length) { writePointer = 0; @@ -70,7 +69,7 @@ } } } - + @Override public void close() throws IOException { @@ -93,12 +92,12 @@ } } } - + private void doWrite() throws DataFormatException, IOException { inflater.setInput(writeBuffer); int n = inflater.inflate(outputBuffer); - + while (n > 0) { output.write(outputBuffer, 0, n);

15 years, 5 months

1
0
0 / 0

JBoss hornetq SVN: r10004 - in trunk: src/main/org/hornetq/core/client/impl and 7 other directories.

by do-not-reply＠jboss.org

Author: clebert.suconic(a)jboss.com Date: 2010-12-06 23:22:51 -0500 (Mon, 06 Dec 2010) New Revision: 10004 Removed: trunk/src/main/org/hornetq/core/client/impl/FailoverManager.java Modified: trunk/src/main/org/hornetq/api/core/client/ClientSessionFactory.java trunk/src/main/org/hornetq/api/core/client/ServerLocator.java trunk/src/main/org/hornetq/core/client/impl/ClientMessageImpl.java trunk/src/main/org/hornetq/core/client/impl/ClientProducerImpl.java trunk/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java trunk/src/main/org/hornetq/core/client/impl/ServerLocatorImpl.java trunk/src/main/org/hornetq/core/message/impl/MessageInternal.java trunk/src/main/org/hornetq/core/server/impl/ServerMessageImpl.java trunk/src/main/org/hornetq/jms/client/HornetQConnectionFactory.java trunk/src/main/org/hornetq/jms/server/config/impl/ConnectionFactoryConfigurationImpl.java trunk/src/main/org/hornetq/jms/server/impl/JMSServerManagerImpl.java trunk/tests/src/org/hornetq/tests/integration/client/LargeMessageCompressTest.java trunk/tests/src/org/hornetq/tests/integration/client/LargeMessageTest.java trunk/tests/src/org/hornetq/tests/unit/core/postoffice/impl/BindingsImplTest.java Log: HORNETQ-448 moving a few properties towards ServerLocator Modified: trunk/src/main/org/hornetq/api/core/client/ClientSessionFactory.java =================================================================== --- trunk/src/main/org/hornetq/api/core/client/ClientSessionFactory.java 2010-12-06 16:57:19 UTC (rev 10003) +++ trunk/src/main/org/hornetq/api/core/client/ClientSessionFactory.java 2010-12-07 04:22:51 UTC (rev 10004) @@ -137,8 +137,4 @@ ServerLocator getServerLocator(); CoreRemotingConnection getConnection(); - - void setCompressLargeMessages(boolean compressLargeMessage); - - boolean isCompressLargeMessages(); } Modified: trunk/src/main/org/hornetq/api/core/client/ServerLocator.java =================================================================== --- trunk/src/main/org/hornetq/api/core/client/ServerLocator.java 2010-12-06 16:57:19 UTC (rev 10003) +++ trunk/src/main/org/hornetq/api/core/client/ServerLocator.java 2010-12-07 04:22:51 UTC (rev 10004) @@ -574,6 +574,10 @@ void close(); boolean isHA(); + + boolean isCompressLargeMessage(); + + void setCompressLargeMessage(boolean compress); void addClusterTopologyListener(ClusterTopologyListener listener); Modified: trunk/src/main/org/hornetq/core/client/impl/ClientMessageImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/ClientMessageImpl.java 2010-12-06 16:57:19 UTC (rev 10003) +++ trunk/src/main/org/hornetq/core/client/impl/ClientMessageImpl.java 2010-12-07 04:22:51 UTC (rev 10004) @@ -73,6 +73,11 @@ super(type, durable, expiration, timestamp, priority, initialMessageBufferSize); } + public boolean isServerMessage() + { + return false; + } + public void onReceipt(final ClientConsumerInternal consumer) { this.consumer = consumer; Modified: trunk/src/main/org/hornetq/core/client/impl/ClientProducerImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/ClientProducerImpl.java 2010-12-06 16:57:19 UTC (rev 10003) +++ trunk/src/main/org/hornetq/core/client/impl/ClientProducerImpl.java 2010-12-07 04:22:51 UTC (rev 10004) @@ -22,6 +22,7 @@ import org.hornetq.api.core.Message; import org.hornetq.api.core.SimpleString; import org.hornetq.core.logging.Logger; +import org.hornetq.core.message.BodyEncoder; import org.hornetq.core.message.impl.MessageInternal; import org.hornetq.core.protocol.core.Channel; import org.hornetq.core.protocol.core.impl.wireformat.SessionSendContinuationMessage; @@ -363,6 +364,12 @@ InputStream input = msgI.getBodyInputStream(); + + if (msgI.isServerMessage()) + { + largeMessageSendServer(sendBlocking, msgI, credits); + } + else if (input != null) { largeMessageSendStreamed(sendBlocking, msgI, input, credits); @@ -372,7 +379,72 @@ largeMessageSendBuffered(sendBlocking, msgI, credits); } } + + /** + * Used to send serverMessages through the bridges. + * No need to validate compression here since the message is only compressed at the client + * @param sendBlocking + * @param msgI + * @throws HornetQException + */ + private void largeMessageSendServer(final boolean sendBlocking, + final MessageInternal msgI, + final ClientProducerCredits credits) throws HornetQException + { + BodyEncoder context = msgI.getBodyEncoder(); + final long bodySize = context.getLargeBodySize(); + + context.open(); + try + { + + for (int pos = 0; pos < bodySize;) + { + final boolean lastChunk; + + final int chunkLength = Math.min((int)(bodySize - pos), minLargeMessageSize); + + final HornetQBuffer bodyBuffer = HornetQBuffers.fixedBuffer(chunkLength); + + context.encode(bodyBuffer, chunkLength); + + pos += chunkLength; + + lastChunk = pos >= bodySize; + + final SessionSendContinuationMessage chunk = new SessionSendContinuationMessage(bodyBuffer.toByteBuffer() + .array(), + !lastChunk, + lastChunk && sendBlocking); + + if (sendBlocking && lastChunk) + { + // When sending it blocking, only the last chunk will be blocking. + channel.sendBlocking(chunk); + } + else + { + channel.send(chunk); + } + + try + { + credits.acquireCredits(chunk.getPacketSize()); + } + catch (InterruptedException e) + { + } + } + } + finally + { + context.close(); + } + } + + + /** * @param sendBlocking * @param msgI Modified: trunk/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java 2010-12-06 16:57:19 UTC (rev 10003) +++ trunk/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java 2010-12-07 04:22:51 UTC (rev 10004) @@ -147,8 +147,6 @@ public final Exception e = new Exception(); private final Object waitLock = new Object(); - - private boolean compressLargeMessages; // Static // --------------------------------------------------------------------------------------- @@ -205,9 +203,7 @@ closeExecutor = orderedExecutorFactory.getExecutor(); this.interceptors = interceptors; - - compressLargeMessages = HornetQClient.DEFAULT_COMPRESS_LARGE_MESSAGES; - + } public void connect(int initialConnectAttempts, boolean failoverOnInitialConnection) throws HornetQException @@ -773,7 +769,7 @@ serverLocator.isBlockOnDurableSend(), serverLocator.isCacheLargeMessagesClient(), serverLocator.getMinLargeMessageSize(), - compressLargeMessages, + serverLocator.isCompressLargeMessage(), serverLocator.getInitialMessagePacketSize(), serverLocator.getGroupID(), connection, @@ -1364,14 +1360,4 @@ cancelled = true; } } - - public void setCompressLargeMessages(boolean compressLargeMessage) - { - this.compressLargeMessages = compressLargeMessage; - } - - public boolean isCompressLargeMessages() - { - return this.compressLargeMessages; - } } Deleted: trunk/src/main/org/hornetq/core/client/impl/FailoverManager.java =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/FailoverManager.java 2010-12-06 16:57:19 UTC (rev 10003) +++ trunk/src/main/org/hornetq/core/client/impl/FailoverManager.java 2010-12-07 04:22:51 UTC (rev 10004) @@ -1,68 +0,0 @@ -/* - * Copyright 2009 Red Hat, Inc. - * Red Hat 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.hornetq.core.client.impl; - -import org.hornetq.api.core.HornetQException; -import org.hornetq.api.core.client.ClientSession; -import org.hornetq.api.core.client.SessionFailureListener; -import org.hornetq.core.protocol.core.CoreRemotingConnection; - -/** - * A ConnectionManager - * - * @author <a href="mailto:tim.fox@jboss.com">Tim Fox</a> - * - * Created 27 Nov 2008 18:45:46 - * - * - */ -public interface FailoverManager -{ - ClientSession createSession(final String username, - final String password, - final boolean xa, - final boolean autoCommitSends, - final boolean autoCommitAcks, - final boolean preAcknowledge, - final int ackBatchSize, - final boolean cacheLargeMessageClient, - final int minLargeMessageSize, - final boolean compressLargeMessages, - final boolean blockOnAcknowledge, - final boolean autoGroup, - final int confirmationWindowSize, - final int producerWindowSize, - final int consumerWindowSize, - final int producerMaxRate, - final int consumerMaxRate, - final boolean blockOnNonDurableSend, - final boolean blockOnDurableSend, - final int initialMessagePacketSize, - final String groupID) throws HornetQException; - - void removeSession(final ClientSessionInternal session); - - public CoreRemotingConnection getConnection(); - - int numConnections(); - - int numSessions(); - - void addFailureListener(SessionFailureListener listener); - - boolean removeFailureListener(SessionFailureListener listener); - - void causeExit(); - -} Modified: trunk/src/main/org/hornetq/core/client/impl/ServerLocatorImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/ServerLocatorImpl.java 2010-12-06 16:57:19 UTC (rev 10003) +++ trunk/src/main/org/hornetq/core/client/impl/ServerLocatorImpl.java 2010-12-07 04:22:51 UTC (rev 10004) @@ -68,6 +68,8 @@ private Pair<TransportConfiguration, TransportConfiguration>[] topologyArray; private boolean receivedTopology; + + private boolean compressLargeMessage; private ExecutorService threadPool; @@ -359,6 +361,8 @@ initialMessagePacketSize = HornetQClient.DEFAULT_INITIAL_MESSAGE_PACKET_SIZE; cacheLargeMessagesClient = HornetQClient.DEFAULT_CACHE_LARGE_MESSAGE_CLIENT; + + compressLargeMessage = HornetQClient.DEFAULT_COMPRESS_LARGE_MESSAGES; clusterConnection = false; } @@ -917,6 +921,22 @@ return groupID; } + /* (non-Javadoc) + * @see org.hornetq.api.core.client.ServerLocator#isCompressLargeMessage() + */ + public boolean isCompressLargeMessage() + { + return compressLargeMessage; + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.client.ServerLocator#setCompressLargeMessage(boolean) + */ + public void setCompressLargeMessage(boolean compress) + { + this.compressLargeMessage = compress; + } + private void checkWrite() { if (readOnly) Modified: trunk/src/main/org/hornetq/core/message/impl/MessageInternal.java =================================================================== --- trunk/src/main/org/hornetq/core/message/impl/MessageInternal.java 2010-12-06 16:57:19 UTC (rev 10003) +++ trunk/src/main/org/hornetq/core/message/impl/MessageInternal.java 2010-12-07 04:22:51 UTC (rev 10004) @@ -43,6 +43,8 @@ void bodyChanged(); void resetCopied(); + + boolean isServerMessage(); HornetQBuffer getEncodedBuffer(); Modified: trunk/src/main/org/hornetq/core/server/impl/ServerMessageImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/ServerMessageImpl.java 2010-12-06 16:57:19 UTC (rev 10003) +++ trunk/src/main/org/hornetq/core/server/impl/ServerMessageImpl.java 2010-12-07 04:22:51 UTC (rev 10004) @@ -89,6 +89,11 @@ { super(other); } + + public boolean isServerMessage() + { + return true; + } public void setMessageID(final long id) { Modified: trunk/src/main/org/hornetq/jms/client/HornetQConnectionFactory.java =================================================================== --- trunk/src/main/org/hornetq/jms/client/HornetQConnectionFactory.java 2010-12-06 16:57:19 UTC (rev 10003) +++ trunk/src/main/org/hornetq/jms/client/HornetQConnectionFactory.java 2010-12-07 04:22:51 UTC (rev 10004) @@ -533,6 +533,16 @@ return serverLocator.getGroupID(); } + public boolean isCompressLargeMessage() + { + return serverLocator.isCompressLargeMessage(); + } + + public void setCompressLargeMessage(boolean compress) + { + serverLocator.setCompressLargeMessage(compress); + } + public void close() { serverLocator.close(); Modified: trunk/src/main/org/hornetq/jms/server/config/impl/ConnectionFactoryConfigurationImpl.java =================================================================== --- trunk/src/main/org/hornetq/jms/server/config/impl/ConnectionFactoryConfigurationImpl.java 2010-12-06 16:57:19 UTC (rev 10003) +++ trunk/src/main/org/hornetq/jms/server/config/impl/ConnectionFactoryConfigurationImpl.java 2010-12-07 04:22:51 UTC (rev 10004) @@ -17,17 +17,19 @@ import java.util.List; import org.hornetq.api.core.HornetQBuffer; -import org.hornetq.api.core.TransportConfiguration; +import org.hornetq.api.core.SimpleString; import org.hornetq.api.core.client.HornetQClient; import org.hornetq.jms.server.config.ConnectionFactoryConfiguration; import org.hornetq.jms.server.impl.JMSFactoryType; import org.hornetq.utils.BufferHelper; import org.hornetq.utils.DataConstants; -import org.hornetq.api.core.SimpleString;; - /** - * A ConnectionFactoryConfigurationImpl + * This class contains the configuration properties of a connection factory. + * + * It is also persisted on the journal at the time of management is used to created a connection factory and set to store. + * + * Every property on this class has to be also set through encoders through EncodingSupport implementation at this class. * * @author <a href="mailto:jmesnil@redhat.com">Jeff Mesnil</a> * @@ -570,6 +572,8 @@ reconnectAttempts = buffer.readInt(); failoverOnInitialConnection = buffer.readBoolean(); + + compressLargeMessage = buffer.readBoolean(); groupID = BufferHelper.readNullableSimpleStringAsString(buffer); @@ -654,6 +658,8 @@ buffer.writeInt(reconnectAttempts); buffer.writeBoolean(failoverOnInitialConnection); + + buffer.writeBoolean(compressLargeMessage); BufferHelper.writeAsNullableSimpleString(buffer, groupID); @@ -760,6 +766,9 @@ DataConstants.SIZE_BOOLEAN + // failoverOnInitialConnection + + DataConstants.SIZE_BOOLEAN + + // compress-large-message BufferHelper.sizeOfNullableSimpleString(groupID) + Modified: trunk/src/main/org/hornetq/jms/server/impl/JMSServerManagerImpl.java =================================================================== --- trunk/src/main/org/hornetq/jms/server/impl/JMSServerManagerImpl.java 2010-12-06 16:57:19 UTC (rev 10003) +++ trunk/src/main/org/hornetq/jms/server/impl/JMSServerManagerImpl.java 2010-12-07 04:22:51 UTC (rev 10004) @@ -1079,6 +1079,7 @@ cf.setMaxRetryInterval(cfConfig.getMaxRetryInterval()); cf.setReconnectAttempts(cfConfig.getReconnectAttempts()); cf.setFailoverOnInitialConnection(cfConfig.isFailoverOnInitialConnection()); + cf.setCompressLargeMessage(cfConfig.isCompressLargeMessages()); } connectionFactories.put(cfConfig.getName(), cf); Modified: trunk/tests/src/org/hornetq/tests/integration/client/LargeMessageCompressTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/client/LargeMessageCompressTest.java 2010-12-06 16:57:19 UTC (rev 10003) +++ trunk/tests/src/org/hornetq/tests/integration/client/LargeMessageCompressTest.java 2010-12-07 04:22:51 UTC (rev 10004) @@ -26,6 +26,7 @@ import org.hornetq.api.core.client.ClientSession; import org.hornetq.api.core.client.ClientSessionFactory; import org.hornetq.api.core.client.HornetQClient; +import org.hornetq.api.core.client.ServerLocator; /** * A LargeMessageCompressTest @@ -45,14 +46,13 @@ return false; } - protected ClientSessionFactory createSessionFactory() throws Exception + protected ServerLocator createFactory(final boolean isNetty) throws Exception { - ClientSessionFactory sf = locator.createSessionFactory(); - sf.setCompressLargeMessages(true); - return sf; + ServerLocator locator = super.createFactory(isNetty); + locator.setCompressLargeMessage(true); + return locator; } - public void testLargeMessageCompression() throws Exception { final int messageSize = (int)(3.5 * HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE); @@ -66,7 +66,6 @@ server.start(); ClientSessionFactory sf = locator.createSessionFactory(); - sf.setCompressLargeMessages(true); session = sf.createSession(false, false, false); @@ -134,7 +133,6 @@ server.start(); ClientSessionFactory sf = locator.createSessionFactory(); - sf.setCompressLargeMessages(true); session = sf.createSession(false, false, false); @@ -214,7 +212,6 @@ server.start(); ClientSessionFactory sf = locator.createSessionFactory(); - sf.setCompressLargeMessages(true); session = sf.createSession(false, false, false); @@ -280,7 +277,7 @@ } - // below are large message tests that are not applied to compressed messages + // TODO: Fix these tests on LargeMessageCompression public void testResendSmallStreamMessage() throws Exception { @@ -306,5 +303,8 @@ { } - + public void testSendServerMessage() throws Exception + { + // doesn't make sense as compressed + } } Modified: trunk/tests/src/org/hornetq/tests/integration/client/LargeMessageTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/client/LargeMessageTest.java 2010-12-06 16:57:19 UTC (rev 10003) +++ trunk/tests/src/org/hornetq/tests/integration/client/LargeMessageTest.java 2010-12-07 04:22:51 UTC (rev 10004) @@ -70,11 +70,6 @@ return false; } - protected ClientSessionFactory createSessionFactory() throws Exception - { - return locator.createSessionFactory(); - } - public void testCloseConsumer() throws Exception { final int messageSize = (int)(3.5 * HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE); @@ -87,7 +82,7 @@ server.start(); - ClientSessionFactory sf = createSessionFactory(); + ClientSessionFactory sf = locator.createSessionFactory(); session = sf.createSession(false, false, false); @@ -158,7 +153,7 @@ public void doTestLargeBuffer(boolean transacted) throws Exception { final int journalsize = 100 * 1024; - final int messageSize = 3 * journalsize + 5; + final int messageSize = 3 * journalsize; // final int messageSize = 5 * 1024; ClientSession session = null; @@ -175,9 +170,7 @@ server.start(); - ClientSessionFactory sf = createSessionFactory(); - - sf.setCompressLargeMessages(true); + ClientSessionFactory sf = locator.createSessionFactory(); session = sf.createSession(!transacted, !transacted, 0); @@ -261,7 +254,7 @@ server.start(); - ClientSessionFactory sf = createSessionFactory(); + ClientSessionFactory sf = locator.createSessionFactory(); session = sf.createSession(false, false, false); @@ -314,7 +307,7 @@ server.start(); - sf = createSessionFactory(); + sf = locator.createSessionFactory(); session = sf.createSession(false, false, false); @@ -388,7 +381,7 @@ server.start(); - ClientSessionFactory sf = createSessionFactory(); + ClientSessionFactory sf = locator.createSessionFactory(); session = sf.createSession(false, false, false); @@ -471,7 +464,7 @@ server.start(); - ClientSessionFactory sf = createSessionFactory(); + ClientSessionFactory sf = locator.createSessionFactory(); SimpleString ADDRESS_DLA = LargeMessageTest.ADDRESS.concat("-dla"); SimpleString ADDRESS_EXPIRY = LargeMessageTest.ADDRESS.concat("-expiry"); @@ -606,7 +599,7 @@ server.start(); - ClientSessionFactory sf = createSessionFactory(); + ClientSessionFactory sf = locator.createSessionFactory(); SimpleString ADDRESS_DLA = LargeMessageTest.ADDRESS.concat("-dla"); SimpleString ADDRESS_EXPIRY = LargeMessageTest.ADDRESS.concat("-expiry"); @@ -682,7 +675,7 @@ server.start(); - sf = createSessionFactory(); + sf = locator.createSessionFactory(); session = sf.createSession(false, false, false); @@ -749,7 +742,7 @@ server.getAddressSettingsRepository().addMatch("*", addressSettings); - ClientSessionFactory sf = createSessionFactory(); + ClientSessionFactory sf = locator.createSessionFactory(); session = sf.createSession(false, false, false); @@ -791,7 +784,7 @@ server.start(); - sf = createSessionFactory(); + sf = locator.createSessionFactory(); session = sf.createSession(false, false, false); @@ -856,7 +849,7 @@ server.start(); - ClientSessionFactory sf = createSessionFactory(); + ClientSessionFactory sf = locator.createSessionFactory(); session = sf.createSession(false, false, false); @@ -894,7 +887,6 @@ } catch (Throwable e) { - log.error("failed", e); failed = true; } @@ -963,7 +955,7 @@ locator.setCacheLargeMessagesClient(true); - ClientSessionFactory sf = createSessionFactory(); + ClientSessionFactory sf = locator.createSessionFactory(); session = sf.createSession(false, false, false); @@ -1895,7 +1887,7 @@ SimpleString queue[] = new SimpleString[] { new SimpleString("queue1"), new SimpleString("queue2") }; - ClientSessionFactory sf = createSessionFactory(); + ClientSessionFactory sf = locator.createSessionFactory(); ClientSession session = sf.createSession(null, null, false, true, true, false, 0); @@ -1978,7 +1970,7 @@ SimpleString queue[] = new SimpleString[] { new SimpleString("queue1"), new SimpleString("queue2") }; - ClientSessionFactory sf = createSessionFactory(); + ClientSessionFactory sf = locator.createSessionFactory(); ClientSession session = sf.createSession(null, null, false, true, true, false, 0); @@ -2006,7 +1998,7 @@ server.start(); - sf = createSessionFactory(); + sf = locator.createSessionFactory(); session = sf.createSession(null, null, false, true, true, false, 0); } @@ -2060,7 +2052,7 @@ server.start(); - ClientSessionFactory sf = createSessionFactory(); + ClientSessionFactory sf = locator.createSessionFactory(); session = sf.createSession(isXA, false, false); @@ -2090,7 +2082,7 @@ session.close(); server.stop(); server.start(); - sf = createSessionFactory(); + sf = locator.createSessionFactory(); session = sf.createSession(isXA, false, false); session.rollback(xid); @@ -2147,7 +2139,7 @@ server.start(); - ClientSessionFactory sf = createSessionFactory(); + ClientSessionFactory sf = locator.createSessionFactory(); ClientSession session = sf.createSession(isXA, false, false); @@ -2284,7 +2276,7 @@ locator.setMinLargeMessageSize(1024); locator.setConsumerWindowSize(1024 * 1024); - ClientSessionFactory sf = createSessionFactory(); + ClientSessionFactory sf = locator.createSessionFactory(); session = sf.createSession(null, null, false, false, false, false, 0); @@ -2388,7 +2380,7 @@ locator.setMinLargeMessageSize(1024); locator.setConsumerWindowSize(1024 * 1024); - ClientSessionFactory sf = createSessionFactory(); + ClientSessionFactory sf = locator.createSessionFactory(); session = sf.createSession(null, null, false, false, false, false, 0); @@ -2491,7 +2483,7 @@ locator.setMinLargeMessageSize(100 * 1024); - ClientSessionFactory sf = createSessionFactory(); + ClientSessionFactory sf = locator.createSessionFactory(); session = sf.createSession(null, null, false, true, true, false, 0); @@ -2565,7 +2557,7 @@ locator.setMinLargeMessageSize(1024); - ClientSessionFactory sf = createSessionFactory(); + ClientSessionFactory sf = locator.createSessionFactory(); session = sf.createSession(null, null, false, true, true, false, 0); @@ -2641,7 +2633,7 @@ server.start(); - ClientSessionFactory sf = createSessionFactory(); + ClientSessionFactory sf = locator.createSessionFactory(); ClientSession session = sf.createSession(false, false); @@ -2695,77 +2687,6 @@ } } - public void testLargeMessageCompression() throws Exception - { - final int messageSize = (int)(3.5 * HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE); - - ClientSession session = null; - - try - { - server = createServer(true, isNetty()); - - server.start(); - - ClientSessionFactory sf = createSessionFactory(); - sf.setCompressLargeMessages(true); - - session = sf.createSession(false, false, false); - - session.createTemporaryQueue(LargeMessageTest.ADDRESS, LargeMessageTest.ADDRESS); - - ClientProducer producer = session.createProducer(LargeMessageTest.ADDRESS); - - Message clientFile = createLargeClientMessage(session, messageSize, true); - - producer.send(clientFile); - - session.commit(); - - session.start(); - - ClientConsumer consumer = session.createConsumer(LargeMessageTest.ADDRESS); - ClientMessage msg1 = consumer.receive(1000); - Assert.assertNotNull(msg1); - - for (int i = 0 ; i < messageSize; i++) - { - //System.out.print(msg1.getBodyBuffer().readByte() + " "); - //if (i % 100 == 0) System.out.println(); - byte b = msg1.getBodyBuffer().readByte(); - //System.out.println("Byte read: " + (char)b + " i " + i); - assertEquals("position = " + i, getSamplebyte(i), b); - } - - msg1.acknowledge(); - session.commit(); - - consumer.close(); - - session.close(); - - validateNoFilesOnLargeDir(); - } - finally - { - try - { - server.stop(); - } - catch (Throwable ignored) - { - } - - try - { - session.close(); - } - catch (Throwable ignored) - { - } - } - } - // Package protected --------------------------------------------- // Protected ----------------------------------------------------- @@ -2810,7 +2731,7 @@ try { - ClientSessionFactory sf = createSessionFactory(); + ClientSessionFactory sf = locator.createSessionFactory(); if (sendBlocking) { @@ -2855,7 +2776,7 @@ server = createServer(true, config, PAGE_SIZE, PAGE_MAX, map); server.start(); - sf = createSessionFactory(); + sf = locator.createSessionFactory(); } session = sf.createSession(null, null, false, true, true, false, 0); 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-12-06 16:57:19 UTC (rev 10003) +++ trunk/tests/src/org/hornetq/tests/unit/core/postoffice/impl/BindingsImplTest.java 2010-12-07 04:22:51 UTC (rev 10004) @@ -974,6 +974,15 @@ } + /* (non-Javadoc) + * @see org.hornetq.core.message.impl.MessageInternal#isServerMessage() + */ + public boolean isServerMessage() + { + // TODO Auto-generated method stub + return false; + } + } class FakeFilter implements Filter

15 years, 5 months

1
0
0 / 0

JBoss hornetq SVN: r10002 - trunk/tests/src/org/hornetq/tests/integration/persistence.

by do-not-reply＠jboss.org

Author: ataylor Date: 2010-12-06 10:46:45 -0500 (Mon, 06 Dec 2010) New Revision: 10002 Modified: trunk/tests/src/org/hornetq/tests/integration/persistence/ExportFormatTest.java Log: fixed journal import test Modified: trunk/tests/src/org/hornetq/tests/integration/persistence/ExportFormatTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/persistence/ExportFormatTest.java 2010-12-06 11:04:17 UTC (rev 10001) +++ trunk/tests/src/org/hornetq/tests/integration/persistence/ExportFormatTest.java 2010-12-06 15:46:45 UTC (rev 10002) @@ -41,36 +41,36 @@ // Attributes ---------------------------------------------------- // Case the format was changed, and the change was agreed, use _testCreateFormat to recreate this field - String bindingsFile = "#File,JournalFileImpl: (hornetq-bindings-1.bindings id = 1, recordID = 1)\n" + "operation@AddRecord,id@2,userRecordType@24,length@8,isUpdate@false,data@AAAAAH____8=\n" - + "operation@AddRecord,id@0,userRecordType@23,length@16,isUpdate@false,data@jhbeVKTqEd-gYwAi-v8IyA==\n" - + "operation@AddRecord,id@4,userRecordType@21,length@17,isUpdate@false,data@AAAABEEAMQAAAAAEQQAxAAA=\n" - + "operation@AddRecord,id@109,userRecordType@24,length@8,isUpdate@false,data@AAAAAAAAAG0=\n" - + "#File,JournalFileImpl: (hornetq-bindings-2.bindings id = 2, recordID = 2)\n"; + String bindingsFile = "#File,JournalFileImpl: (hornetq-bindings-1.bindings id = 1, recordID = 1)\n" + + "operation@AddRecord,id@2,userRecordType@24,length@8,isUpdate@false,compactCount@0,data@AAAAAH____8=\n" + + "operation@AddRecord,id@2,userRecordType@21,length@17,isUpdate@false,compactCount@0,data@AAAABEEAMQAAAAAEQQAxAAA=\n" + + "operation@AddRecord,id@20,userRecordType@24,length@8,isUpdate@false,compactCount@0,data@AAAAAAAAABQ=\n" + + "#File,JournalFileImpl: (hornetq-bindings-2.bindings id = 2, recordID = 2)"; // Case the format was changed, and the change was agreed, use _testCreateFormat to recreate this field - String journalFile = "#File,JournalFileImpl: (hornetq-data-1.hq id = 1, recordID = 1)\n" - + "operation@AddRecordTX,txID@3,id@6,userRecordType@31,length@65,isUpdate@false,data@AAAAEQAAAE4AAAAAAAAABgEAAAAEQQAxAAAA_wAAAAAAAAAAAAABKl7rANMEAQAAAAEAAAAGawBlAHkABgAAAAA=\n" - + "operation@UpdateTX,txID@3,id@6,userRecordType@32,length@8,isUpdate@true,data@AAAAAAAAAAQ=\n" - + "operation@AddRecordTX,txID@3,id@7,userRecordType@31,length@65,isUpdate@false,data@AAAAEQAAAE4AAAAAAAAABwEAAAAEQQAxAAAA_wAAAAAAAAAAAAABKl7rANcEAQAAAAEAAAAGawBlAHkABgAAAAE=\n" - + "operation@UpdateTX,txID@3,id@7,userRecordType@32,length@8,isUpdate@true,data@AAAAAAAAAAQ=\n" - + "operation@AddRecordTX,txID@3,id@8,userRecordType@31,length@65,isUpdate@false,data@AAAAEQAAAE4AAAAAAAAACAEAAAAEQQAxAAAA_wAAAAAAAAAAAAABKl7rANcEAQAAAAEAAAAGawBlAHkABgAAAAI=\n" - + "operation@UpdateTX,txID@3,id@8,userRecordType@32,length@8,isUpdate@true,data@AAAAAAAAAAQ=\n" - + "operation@AddRecordTX,txID@3,id@9,userRecordType@31,length@65,isUpdate@false,data@AAAAEQAAAE4AAAAAAAAACQEAAAAEQQAxAAAA_wAAAAAAAAAAAAABKl7rANcEAQAAAAEAAAAGawBlAHkABgAAAAM=\n" - + "operation@UpdateTX,txID@3,id@9,userRecordType@32,length@8,isUpdate@true,data@AAAAAAAAAAQ=\n" - + "operation@AddRecordTX,txID@3,id@10,userRecordType@31,length@65,isUpdate@false,data@AAAAEQAAAE4AAAAAAAAACgEAAAAEQQAxAAAA_wAAAAAAAAAAAAABKl7rANcEAQAAAAEAAAAGawBlAHkABgAAAAQ=\n" - + "operation@UpdateTX,txID@3,id@10,userRecordType@32,length@8,isUpdate@true,data@AAAAAAAAAAQ=\n" - + "operation@Commit,txID@3,numberOfRecords@10\n" - + "operation@AddRecord,id@14,userRecordType@31,length@65,isUpdate@false,data@AAAAEQAAAE4AAAAAAAAADgEAAAAEQQAxAAAA_wAAAAAAAAAAAAABKl7rAOEEAQAAAAEAAAAGawBlAHkABgAAAAU=\n" - + "operation@Update,id@14,userRecordType@32,length@8,isUpdate@true,data@AAAAAAAAAAQ=\n" - + "operation@AddRecord,id@15,userRecordType@31,length@65,isUpdate@false,data@AAAAEQAAAE4AAAAAAAAADwEAAAAEQQAxAAAA_wAAAAAAAAAAAAABKl7rAOQEAQAAAAEAAAAGawBlAHkABgAAAAY=\n" - + "operation@Update,id@15,userRecordType@32,length@8,isUpdate@true,data@AAAAAAAAAAQ=\n" - + "operation@AddRecord,id@16,userRecordType@31,length@65,isUpdate@false,data@AAAAEQAAAE4AAAAAAAAAEAEAAAAEQQAxAAAA_wAAAAAAAAAAAAABKl7rAOUEAQAAAAEAAAAGawBlAHkABgAAAAc=\n" - + "operation@Update,id@16,userRecordType@32,length@8,isUpdate@true,data@AAAAAAAAAAQ=\n" - + "operation@AddRecord,id@17,userRecordType@31,length@65,isUpdate@false,data@AAAAEQAAAE4AAAAAAAAAEQEAAAAEQQAxAAAA_wAAAAAAAAAAAAABKl7rAOcEAQAAAAEAAAAGawBlAHkABgAAAAg=\n" - + "operation@Update,id@17,userRecordType@32,length@8,isUpdate@true,data@AAAAAAAAAAQ=\n" - + "operation@AddRecord,id@18,userRecordType@31,length@65,isUpdate@false,data@AAAAEQAAAE4AAAAAAAAAEgEAAAAEQQAxAAAA_wAAAAAAAAAAAAABKl7rAOgEAQAAAAEAAAAGawBlAHkABgAAAAk=\n" - + "operation@Update,id@18,userRecordType@32,length@8,isUpdate@true,data@AAAAAAAAAAQ=\n" - + "#File,JournalFileImpl: (hornetq-data-2.hq id = 2, recordID = 2)\n"; + String journalFile = "#File,JournalFileImpl: (hornetq-data-1.hq id = 1, recordID = 1)\n" + + "operation@AddRecordTX,txID@0,id@4,userRecordType@31,length@65,isUpdate@false,compactCount@0,data@AAAAEQAAAE4AAAAAAAAABAEAAAAEQQAxAAAA_wAAAAAAAAAAAAABLLxYP40EAQAAAAEAAAAGawBlAHkABgAAAAA=\n" + + "operation@UpdateTX,txID@0,id@4,userRecordType@32,length@8,isUpdate@true,compactCount@0,data@AAAAAAAAAAI=\n" + + "operation@AddRecordTX,txID@0,id@5,userRecordType@31,length@65,isUpdate@false,compactCount@0,data@AAAAEQAAAE4AAAAAAAAABQEAAAAEQQAxAAAA_wAAAAAAAAAAAAABLLxYP5EEAQAAAAEAAAAGawBlAHkABgAAAAE=\n" + + "operation@UpdateTX,txID@0,id@5,userRecordType@32,length@8,isUpdate@true,compactCount@0,data@AAAAAAAAAAI=\n" + + "operation@AddRecordTX,txID@0,id@6,userRecordType@31,length@65,isUpdate@false,compactCount@0,data@AAAAEQAAAE4AAAAAAAAABgEAAAAEQQAxAAAA_wAAAAAAAAAAAAABLLxYP5EEAQAAAAEAAAAGawBlAHkABgAAAAI=\n" + + "operation@UpdateTX,txID@0,id@6,userRecordType@32,length@8,isUpdate@true,compactCount@0,data@AAAAAAAAAAI=\n" + + "operation@AddRecordTX,txID@0,id@7,userRecordType@31,length@65,isUpdate@false,compactCount@0,data@AAAAEQAAAE4AAAAAAAAABwEAAAAEQQAxAAAA_wAAAAAAAAAAAAABLLxYP5EEAQAAAAEAAAAGawBlAHkABgAAAAM=\n" + + "operation@UpdateTX,txID@0,id@7,userRecordType@32,length@8,isUpdate@true,compactCount@0,data@AAAAAAAAAAI=\n" + + "operation@AddRecordTX,txID@0,id@8,userRecordType@31,length@65,isUpdate@false,compactCount@0,data@AAAAEQAAAE4AAAAAAAAACAEAAAAEQQAxAAAA_wAAAAAAAAAAAAABLLxYP5EEAQAAAAEAAAAGawBlAHkABgAAAAQ=\n" + + "operation@UpdateTX,txID@0,id@8,userRecordType@32,length@8,isUpdate@true,compactCount@0,data@AAAAAAAAAAI=\n" + + "operation@Commit,txID@0,numberOfRecords@10\n" + + "operation@AddRecord,id@12,userRecordType@31,length@65,isUpdate@false,compactCount@0,data@AAAAEQAAAE4AAAAAAAAADAEAAAAEQQAxAAAA_wAAAAAAAAAAAAABLLxYP6gEAQAAAAEAAAAGawBlAHkABgAAAAU=\n" + + "operation@Update,id@12,userRecordType@32,length@8,isUpdate@true,compactCount@0,data@AAAAAAAAAAI=\n" + + "operation@AddRecord,id@13,userRecordType@31,length@65,isUpdate@false,compactCount@0,data@AAAAEQAAAE4AAAAAAAAADQEAAAAEQQAxAAAA_wAAAAAAAAAAAAABLLxYP6oEAQAAAAEAAAAGawBlAHkABgAAAAY=\n" + + "operation@Update,id@13,userRecordType@32,length@8,isUpdate@true,compactCount@0,data@AAAAAAAAAAI=\n" + + "operation@AddRecord,id@14,userRecordType@31,length@65,isUpdate@false,compactCount@0,data@AAAAEQAAAE4AAAAAAAAADgEAAAAEQQAxAAAA_wAAAAAAAAAAAAABLLxYP6sEAQAAAAEAAAAGawBlAHkABgAAAAc=\n" + + "operation@Update,id@14,userRecordType@32,length@8,isUpdate@true,compactCount@0,data@AAAAAAAAAAI=\n" + + "operation@AddRecord,id@15,userRecordType@31,length@65,isUpdate@false,compactCount@0,data@AAAAEQAAAE4AAAAAAAAADwEAAAAEQQAxAAAA_wAAAAAAAAAAAAABLLxYP60EAQAAAAEAAAAGawBlAHkABgAAAAg=\n" + + "operation@Update,id@15,userRecordType@32,length@8,isUpdate@true,compactCount@0,data@AAAAAAAAAAI=\n" + + "operation@AddRecord,id@16,userRecordType@31,length@65,isUpdate@false,compactCount@0,data@AAAAEQAAAE4AAAAAAAAAEAEAAAAEQQAxAAAA_wAAAAAAAAAAAAABLLxYP64EAQAAAAEAAAAGawBlAHkABgAAAAk=\n" + + "operation@Update,id@16,userRecordType@32,length@8,isUpdate@true,compactCount@0,data@AAAAAAAAAAI=\n" + + "#File,JournalFileImpl: (hornetq-data-2.hq id = 2, recordID = 2)"; // Static --------------------------------------------------------

15 years, 5 months

1
0
0 / 0

JBoss hornetq SVN: r10001 - trunk/tests/src/org/hornetq/tests/integration/jms.

by do-not-reply＠jboss.org

Author: ataylor Date: 2010-12-06 06:04:17 -0500 (Mon, 06 Dec 2010) New Revision: 10001 Modified: trunk/tests/src/org/hornetq/tests/integration/jms/HornetQConnectionFactoryTest.java Log: fixed connection factory test Modified: trunk/tests/src/org/hornetq/tests/integration/jms/HornetQConnectionFactoryTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/jms/HornetQConnectionFactoryTest.java 2010-12-06 10:57:42 UTC (rev 10000) +++ trunk/tests/src/org/hornetq/tests/integration/jms/HornetQConnectionFactoryTest.java 2010-12-06 11:04:17 UTC (rev 10001) @@ -156,7 +156,7 @@ public void testDiscoveryConstructor() throws Exception { - DiscoveryGroupConfiguration groupConfiguration = new DiscoveryGroupConfiguration("test", "foo", "192.168.5.4", 3456, 5000, 1000); + DiscoveryGroupConfiguration groupConfiguration = new DiscoveryGroupConfiguration(groupAddress, groupPort); HornetQConnectionFactory cf = (HornetQConnectionFactory) HornetQJMSClient.createConnectionFactoryWithoutHA(groupConfiguration, JMSFactoryType.CF); assertFactoryParams(cf, null,

15 years, 5 months

1
0
0 / 0

2026

2025

2024

2023

2022

2021

2020

2019

2018

2017

2016

2015

2014

2013

2012

2011

2010

2009

hornetq-commits December 2010