JBoss hornetq SVN: r8617 - trunk/src/main/org/hornetq/core/buffers.
by do-not-reply＠jboss.org
Author: jmesnil
Date: 2009-12-08 05:23:43 -0500 (Tue, 08 Dec 2009)
New Revision: 8617
Added:
trunk/src/main/org/hornetq/core/buffers/package-info.java
Removed:
trunk/src/main/org/hornetq/core/buffers/package.html
Modified:
trunk/src/main/org/hornetq/core/buffers/HornetQBuffer.java
trunk/src/main/org/hornetq/core/buffers/HornetQBuffers.java
Log:
HORNETQ-186: fill in Javadocs for core API
* documented buffers API
Modified: trunk/src/main/org/hornetq/core/buffers/HornetQBuffer.java
===================================================================
--- trunk/src/main/org/hornetq/core/buffers/HornetQBuffer.java 2009-12-08 10:23:19 UTC (rev 8616)
+++ trunk/src/main/org/hornetq/core/buffers/HornetQBuffer.java 2009-12-08 10:23:43 UTC (rev 8617)
@@ -20,15 +20,22 @@
/**
*
- * A HornetQBuffer
+ * A HornetQBuffer wraps a Netty's ChannelBuffer and is used throughout HornetQ code base.
*
* Much of it derived from Netty ChannelBuffer by Trustin Lee
*
* @author <a href="mailto:tim.fox@jboss.com">Tim Fox</a>
+ *
+ * @see HornetQBuffers
*
*/
public interface HornetQBuffer
{
+ /**
+ * Return the underlying Netty's ChannelBuffer
+ *
+ * @return the underlying Netty's ChannelBuffer
+ */
ChannelBuffer channelBuffer();
int capacity();
Modified: trunk/src/main/org/hornetq/core/buffers/HornetQBuffers.java
===================================================================
--- trunk/src/main/org/hornetq/core/buffers/HornetQBuffers.java 2009-12-08 10:23:19 UTC (rev 8616)
+++ trunk/src/main/org/hornetq/core/buffers/HornetQBuffers.java 2009-12-08 10:23:43 UTC (rev 8617)
@@ -20,20 +20,29 @@
import org.jboss.netty.buffer.ChannelBuffers;
/**
- *
- * A HornetQChannelBuffers
+ * Factory class to create HornetQBuffers
*
- * @author tim
- *
- *
+ * @author <a href="mailto:tim.fox@jboss.com">Tim Fox</a>
*/
public class HornetQBuffers
-{
- public static HornetQBuffer dynamicBuffer(final int estimatedLength)
+{
+ /**
+ * Create a <em>self-expanding</em> HornetQBuffer with the given initial size
+ *
+ * @param size the initial size of the created HornetQBuffer
+ * @return a self-expanding HornetQBuffer starting with the given size
+ */
+ public static HornetQBuffer dynamicBuffer(final int size)
{
- return new ChannelBufferWrapper(ChannelBuffers.dynamicBuffer(estimatedLength));
+ return new ChannelBufferWrapper(ChannelBuffers.dynamicBuffer(size));
}
+ /**
+ * Create a <em>self-expanding</em> HornetQBuffer filled with the given byte array
+ *
+ * @param bytes the created buffer will be initially filled with this byte array
+ * @return a self-expanding HornetQBuffer filled with the given byte array
+ */
public static HornetQBuffer dynamicBuffer(final byte[] bytes)
{
ChannelBuffer buff = ChannelBuffers.dynamicBuffer(bytes.length);
@@ -43,6 +52,14 @@
return new ChannelBufferWrapper(buff);
}
+ /**
+ * Create a HornetQBuffer wrapping the underlying NIO ByteBuffer
+ *
+ * The position on this buffer won't affect the position on the inner buffer
+ *
+ * @param underlying the underlying NIO ByteBuffer
+ * @return a HornetQBuffer wrapping the underlying NIO ByteBuffer
+ */
public static HornetQBuffer wrappedBuffer(final ByteBuffer underlying)
{
HornetQBuffer buff = new ChannelBufferWrapper(ChannelBuffers.wrappedBuffer(underlying));
@@ -52,11 +69,23 @@
return buff;
}
+ /**
+ * Create a HornetQBuffer wrapping the underlying byte array
+ *
+ * @param underlying the underlying byte array
+ * @return a HornetQBuffer wrapping the underlying byte array
+ */
public static HornetQBuffer wrappedBuffer(final byte[] underlying)
{
return new ChannelBufferWrapper(ChannelBuffers.wrappedBuffer(underlying));
}
+ /**
+ * Create a <em>fixed</em> HornetQBuffer of the given size
+ *
+ * @param size the size of the created HornetQBuffer
+ * @return a fixed HornetQBuffer with the given size
+ */
public static HornetQBuffer fixedBuffer(final int size)
{
return new ChannelBufferWrapper(ChannelBuffers.buffer(size));
Copied: trunk/src/main/org/hornetq/core/buffers/package-info.java (from rev 8601, trunk/src/main/org/hornetq/core/buffers/package.html)
===================================================================
--- trunk/src/main/org/hornetq/core/buffers/package-info.java (rev 0)
+++ trunk/src/main/org/hornetq/core/buffers/package-info.java 2009-12-08 10:23:43 UTC (rev 8617)
@@ -0,0 +1,32 @@
+/*
+ * 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.
+ */
+
+/**
+ * HornetQ Buffering.
+ *
+ * This package defines the buffers used by HornetQ. The underlying implementations uses
+ * Netty's ChannelBuffer and wraps it with methods required by HornetQ usage.
+ *
+ * ChannelBuffer differs from {@link java.nio.ByteBuffer} in two ways:
+ * <ol>
+ * <li>it is possible to interface almost directly with byte arrays, what is faster</li>
+ * <li>there are two positions, one for reading, and one for writing. Hence you will find methods to read from the buffer
+ * and methods to write to the buffer</li>
+ * </ol>
+ *
+ * <h2>Usage</h2>
+ *
+ * Always use the static methods declared at {@link org.hornetq.core.buffers.HornetQBuffers} to create the buffers.
+*/
+package org.hornetq.core.buffers;
+
Deleted: trunk/src/main/org/hornetq/core/buffers/package.html
===================================================================
--- trunk/src/main/org/hornetq/core/buffers/package.html 2009-12-08 10:23:19 UTC (rev 8616)
+++ trunk/src/main/org/hornetq/core/buffers/package.html 2009-12-08 10:23:43 UTC (rev 8617)
@@ -1,46 +0,0 @@
-<!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN">
-<html>
- <head>
- <!--
-
- JBoss: The OpenSource J2EE WebOS
-
- Distributable under LGPL license.
- See terms of license at gnu.org.
-
- -->
- </head>
- <body>
-
-<h2>HornetQ Buffering</h2>
-
-
-<p>This package is a stripped down version of org.jboss.netty.buffer, and we kept only what we needed for HornetQ buffers.</p>
-
-<p>ChannelBuffer differs from java.nio.ByteBuffer in two ways:</p>
-
-<p>- It's possible to interface almost directly with byte arrays, what is faster</p>
-
-<p>- There are two positions, one for reading, and one for writing. Hence you will find methods for read, and methods for writing</p>
-
-<h2>Usage</h2>
-
-<p> Always use the static methods declared at org.hornetq.core.buffers.ChannelBuffers to create the buffers you need: </p>
-
-
-<p> buffer(int capacity) - This method will create a fixed size MessagingBuffer </p>
-
-<p> dynamicBuffer(final int estimatedLength) - This method will create a SelfExpandable MessagingBuffer </p>
-
-<p> dynamicBuffer(final byte[] initialBuffer) - This method will create a SelfExpandable MessagingBuffer, but reusing the initialBuffer. Be careful though, the reference will be directly used on the createdBuffer. If your buffer will be used by other operations, you should instead create a new buffer and perform a write. </p>
-
-<p> buffer(final int capacity) - This method will create a fixed size MessagingBuffer </p>
-
-<p> wrappedBuffer(final byte[] array) - It will wrap a byte[] on a Buffer, with writePosition at the end, and readPosition at 0 </p>
-
-<p> wrappedBuffer(final ByteBuffer buffer) - It will wrap a Bytebuffer on a MessagingBuffer. The position on this buffer won't affect the position on the inner buffer </p>
-
-
-
- </body>
-</html>
\ No newline at end of file
15 years, 1 month
JBoss hornetq SVN: r8616 - trunk/src/main/org/hornetq/core/client.
by do-not-reply＠jboss.org
Author: jmesnil
Date: 2009-12-08 05:23:19 -0500 (Tue, 08 Dec 2009)
New Revision: 8616
Added:
trunk/src/main/org/hornetq/core/client/package-info.java
Modified:
trunk/src/main/org/hornetq/core/client/ClientSession.java
Log:
HORNETQ-186: fill in Javadocs for core API
* documented ClientSession interface
Modified: trunk/src/main/org/hornetq/core/client/ClientSession.java
===================================================================
--- trunk/src/main/org/hornetq/core/client/ClientSession.java 2009-12-08 10:22:50 UTC (rev 8615)
+++ trunk/src/main/org/hornetq/core/client/ClientSession.java 2009-12-08 10:23:19 UTC (rev 8616)
@@ -21,33 +21,85 @@
import org.hornetq.utils.SimpleString;
/**
+ * A ClientSession is a single-thread object required for producing and consuming messages.
+ *
* @author <a href="mailto:tim.fox@jboss.com">Tim Fox</a>
* @author <a href="mailto:clebert.suconic@jboss.org">Clebert Suconic</a>
* @author <a href="mailto:ataylor@redhat.com">Andy Taylor</a>
- * @author <a href="jmesnil(a)redhat.com">Jeff Mesnil</a>
+ * @author <a href="mailto:jmesnil@redhat.com">Jeff Mesnil</a>
*
*/
public interface ClientSession extends XAResource
{
+ /**
+ * Information returned by a binding query
+ *
+ * @see ClientSession#bindingQuery(SimpleString)
+ */
public interface BindingQuery
{
+ /**
+ * Return <code>true</code> if the binding exists, <code>false</code> else.
+ *
+ * @return <code>true</code> if the binding exists, <code>false</code> else
+ */
boolean isExists();
+ /**
+ * Return the names of the queues bound to the binding.
+ *
+ * @return the names of the queues bound to the binding
+ */
public List<SimpleString> getQueueNames();
}
+ /**
+ * Information returned by a queue query
+ *
+ * @see ClientSession#queueQuery(SimpleString)
+ */
public interface QueueQuery
{
+ /**
+ * Return <code>true</code> if the queue exists, <code>false</code> else.
+ *
+ * @return <code>true</code> if the queue exists, <code>false</code> else
+ */
boolean isExists();
+ /**
+ * Return <code>true</code> if the queue is durable, <code>false</code> else.
+ *
+ * @return <code>true</code> if the queue is durable, <code>false</code> else
+ */
boolean isDurable();
+ /**
+ * Return the number of consumers attached to the queue.
+ *
+ * @return the number of consumers attached to the queue
+ */
int getConsumerCount();
+ /**
+ * Return the number of messages in the queue.
+ *
+ * @return the number of messages in the queue
+ */
int getMessageCount();
+ /**
+ * Return the queue's filter string (or <code>null</code> if the queue has no filter).
+ *
+ * @return the queue's filter string (or <code>null</code> if the queue has no filter)
+ */
SimpleString getFilterString();
+ /**
+ * Return the address that the queue is bound to.
+ *
+ * @return the address that the queue is bound to
+ */
SimpleString getAddress();
}
@@ -108,7 +160,7 @@
// Queue Operations ----------------------------------------------
/**
- * Create a queue. The created queue is <em>not</em> temporary.
+ * Create a <em>non-temporary</em> queue.
*
* @param address the queue will be bound to this address
* @param queueName the name of the queue
@@ -118,7 +170,7 @@
void createQueue(SimpleString address, SimpleString queueName, boolean durable) throws HornetQException;
/**
- * Create a queue. The created queue is <em>not</em> temporary.
+ * Create a <em>non-temporary</em> queue.
*
* @param address the queue will be bound to this address
* @param queueName the name of the queue
@@ -128,7 +180,7 @@
void createQueue(String address, String queueName, boolean durable) throws HornetQException;
/**
- * Create a queue. The created queue is <em>not</em> temporary and <em>not</em> durable.
+ * Create a <em>non-temporary</em> queue <em>non-durable</em> queue.
*
* @param address the queue will be bound to this address
* @param queueName the name of the queue
@@ -137,7 +189,7 @@
void createQueue(String address, String queueName) throws HornetQException;
/**
- * Create a queue. The created queue is <em>not</em> temporary.
+ * Create a <em>non-temporary</em> queue.
*
* @param address the queue will be bound to this address
* @param queueName the name of the queue
@@ -148,7 +200,7 @@
void createQueue(SimpleString address, SimpleString queueName, SimpleString filter, boolean durable) throws HornetQException;
/**
- * Create a queue. The created queue is <em>not</em> temporary.
+ * Create a <em>non-temporary</em>queue.
*
* @param address the queue will be bound to this address
* @param queueName the name of the queue
@@ -420,8 +472,24 @@
// Query operations ----------------------------------------------
+ /**
+ * Query information on a queue.
+ *
+ * @param queueName the name of the queue to query
+ * @return a QueueQuery containing information on the given queue
+ *
+ * @throws HornetQException if an exception occurs while querying the queue
+ */
QueueQuery queueQuery(SimpleString queueName) throws HornetQException;
+ /**
+ * Query information on a binding.
+ *
+ * @param address the address of the biding to query
+ * @return a BindingQuery containing information on the binding attached to the given address
+ *
+ * @throws HornetQException if an exception occurs while querying the binding
+ */
BindingQuery bindingQuery(SimpleString address) throws HornetQException;
// Transaction operations ----------------------------------------
@@ -493,6 +561,11 @@
*/
boolean isBlockOnAcknowledge();
+ /**
+ * Set a <code>SendAcknowledgementHandler</code> for this session.
+ *
+ * @param handler a SendAcknowledgementHandler
+ */
void setSendAcknowledgementHandler(SendAcknowledgementHandler handler);
}
Added: trunk/src/main/org/hornetq/core/client/package-info.java
===================================================================
--- trunk/src/main/org/hornetq/core/client/package-info.java (rev 0)
+++ trunk/src/main/org/hornetq/core/client/package-info.java 2009-12-08 10:23:19 UTC (rev 8616)
@@ -0,0 +1,20 @@
+/*
+ * 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.
+ */
+
+/**
+ * Core API to produce and consume messages.
+ *
+ * This package defines the API used by HornetQ clients to produce and consume messages.
+ */
+package org.hornetq.core.client;
+
15 years, 1 month
JBoss hornetq SVN: r8615 - trunk.
by do-not-reply＠jboss.org
Author: jmesnil
Date: 2009-12-08 05:22:50 -0500 (Tue, 08 Dec 2009)
New Revision: 8615
Modified:
trunk/build-hornetq.xml
Log:
HORNETQ-186: fill in Javadocs for core API
* added missing API/SPI packages to javadoc
Modified: trunk/build-hornetq.xml
===================================================================
--- trunk/build-hornetq.xml 2009-12-08 10:08:32 UTC (rev 8614)
+++ trunk/build-hornetq.xml 2009-12-08 10:22:50 UTC (rev 8615)
@@ -1009,11 +1009,14 @@
windowtitle="HornetQ ${module.version}">
<packageset dir="${src.main.dir}" defaultexcludes="false">
+ <include name="org/hornetq/core/buffers"/>
<include name="org/hornetq/core/config"/>
<include name="org/hornetq/core/client"/>
<include name="org/hornetq/core/exception"/>
+ <include name="org/hornetq/core/management"/>
+ <include name="org/hornetq/core/message"/>
<include name="org/hornetq/core/remoting"/>
- <include name="org/hornetq/core/management"/>
+ <include name="org/hornetq/core/remoting/spi"/>
<include name="org/hornetq/jms"/>
<include name="org/hornetq/jms/client"/>
<include name="org/hornetq/jms/server/management"/>
@@ -1022,8 +1025,10 @@
<classpath refid="javadoc.classpath"/>
<doctitle><![CDATA[<h2>HornetQ ${module.version}</h2>]]></doctitle>
<bottom><![CDATA[<i>Copyright © 2009 Red Hat Inc. All Rights Reserved.</i>]]></bottom>
- <group title="HornetQ Core API" packages="org.hornetq.core.client, org.hornetq.core.config, org.hornetq.core.exception, org.hornetq.core.remoting, org.hornetq.utils"/>
+ <group title="HornetQ Core API" packages="org.hornetq.core.buffers, org.hornetq.core.client, org.hornetq.core.config, org.hornetq.core.exception, org.hornetq.core.message, org.hornetq.utils"/>
<group title="HornetQ Management API" packages="org.hornetq.core.management"/>
+ <group title="HornetQ Remoting API" packages="org.hornetq.core.remoting"/>
+ <group title="HornetQ Remoting SPI" packages="org.hornetq.core.remoting.spi"/>
<group title="JMS Facade" packages="org.hornetq.jms, org.hornetq.jms.client"/>
<group title="JMS Management API" packages="org.hornetq.jms.server.management"/>
<link href="http://java.sun.com/j2se/1.5.0/docs/api"/>
15 years, 1 month
JBoss hornetq SVN: r8614 - trunk.
by do-not-reply＠jboss.org
Author: timfox
Date: 2009-12-08 05:08:32 -0500 (Tue, 08 Dec 2009)
New Revision: 8614
Modified:
trunk/build-hornetq.xml
Log:
commented out stress test for now, so we can get a proper blue run
Modified: trunk/build-hornetq.xml
===================================================================
--- trunk/build-hornetq.xml 2009-12-08 02:28:02 UTC (rev 8613)
+++ trunk/build-hornetq.xml 2009-12-08 10:08:32 UTC (rev 8614)
@@ -1369,6 +1369,7 @@
<formatter type="plain" usefile="${junit.formatter.usefile}"/>
<fileset dir="${test.classes.dir}">
<include name="**/org/hornetq/tests/stress/**/*${test-mask}.class"/>
+ <exclude name="**/org/hornetq/tests/stress/failover/MultiThreadRandomReattachStressTest.class"/>
</fileset>
</batchtest>
</junit>
15 years, 1 month
JBoss hornetq SVN: r8613 - trunk/src/main/org/hornetq/core/remoting/impl.
by do-not-reply＠jboss.org
Author: clebert.suconic(a)jboss.com
Date: 2009-12-07 21:28:02 -0500 (Mon, 07 Dec 2009)
New Revision: 8613
Modified:
trunk/src/main/org/hornetq/core/remoting/impl/ChannelImpl.java
Log:
Reverting change for now
Modified: trunk/src/main/org/hornetq/core/remoting/impl/ChannelImpl.java
===================================================================
--- trunk/src/main/org/hornetq/core/remoting/impl/ChannelImpl.java 2009-12-08 02:09:52 UTC (rev 8612)
+++ trunk/src/main/org/hornetq/core/remoting/impl/ChannelImpl.java 2009-12-08 02:28:02 UTC (rev 8613)
@@ -64,6 +64,8 @@
private final Object sendLock = new Object();
+ private final Object sendBlockingLock = new Object();
+
private boolean failingOver;
private final int confWindowSize;
@@ -192,7 +194,7 @@
// Synchronized since can't be called concurrently by more than one thread and this can occur
// E.g. blocking acknowledge() from inside a message handler at some time as other operation on main thread
- synchronized (sendLock)
+ synchronized (sendBlockingLock)
{
packet.setChannelID(id);
15 years, 1 month
JBoss hornetq SVN: r8612 - trunk/src/main/org/hornetq/core/remoting/impl.
by do-not-reply＠jboss.org
Author: clebert.suconic(a)jboss.com
Date: 2009-12-07 21:09:52 -0500 (Mon, 07 Dec 2009)
New Revision: 8612
Modified:
trunk/src/main/org/hornetq/core/remoting/impl/ChannelImpl.java
Log:
Possible fix for ChannelImpl
Modified: trunk/src/main/org/hornetq/core/remoting/impl/ChannelImpl.java
===================================================================
--- trunk/src/main/org/hornetq/core/remoting/impl/ChannelImpl.java 2009-12-08 01:06:31 UTC (rev 8611)
+++ trunk/src/main/org/hornetq/core/remoting/impl/ChannelImpl.java 2009-12-08 02:09:52 UTC (rev 8612)
@@ -64,8 +64,6 @@
private final Object sendLock = new Object();
- private final Object sendBlockingLock = new Object();
-
private boolean failingOver;
private final int confWindowSize;
@@ -194,7 +192,7 @@
// Synchronized since can't be called concurrently by more than one thread and this can occur
// E.g. blocking acknowledge() from inside a message handler at some time as other operation on main thread
- synchronized (sendBlockingLock)
+ synchronized (sendLock)
{
packet.setChannelID(id);
15 years, 1 month
JBoss hornetq SVN: r8610 - trunk.
by do-not-reply＠jboss.org
Author: jmesnil
Date: 2009-12-07 11:38:47 -0500 (Mon, 07 Dec 2009)
New Revision: 8610
Modified:
trunk/build-hornetq.xml
Log:
HORNETQ-186: fill in Javadocs for core API
* added org.hornetq.core.exception to the API javadoc
Modified: trunk/build-hornetq.xml
===================================================================
--- trunk/build-hornetq.xml 2009-12-07 16:31:21 UTC (rev 8609)
+++ trunk/build-hornetq.xml 2009-12-07 16:38:47 UTC (rev 8610)
@@ -1011,6 +1011,7 @@
<packageset dir="${src.main.dir}" defaultexcludes="false">
<include name="org/hornetq/core/config"/>
<include name="org/hornetq/core/client"/>
+ <include name="org/hornetq/core/exception"/>
<include name="org/hornetq/core/remoting"/>
<include name="org/hornetq/core/management"/>
<include name="org/hornetq/jms"/>
@@ -1021,7 +1022,7 @@
<classpath refid="javadoc.classpath"/>
<doctitle><![CDATA[<h2>HornetQ ${module.version}</h2>]]></doctitle>
<bottom><![CDATA[<i>Copyright © 2009 Red Hat Inc. All Rights Reserved.</i>]]></bottom>
- <group title="HornetQ Core API" packages="org.hornetq.core.client, org.hornetq.core.config, org.hornetq.core.remoting, org.hornetq.utils"/>
+ <group title="HornetQ Core API" packages="org.hornetq.core.client, org.hornetq.core.config, org.hornetq.core.exception, org.hornetq.core.remoting, org.hornetq.utils"/>
<group title="HornetQ Management API" packages="org.hornetq.core.management"/>
<group title="JMS Facade" packages="org.hornetq.jms, org.hornetq.jms.client"/>
<group title="JMS Management API" packages="org.hornetq.jms.server.management"/>
15 years, 1 month
JBoss hornetq SVN: r8609 - in trunk: src/main/org/hornetq/core/client/impl and 4 other directories.
by do-not-reply＠jboss.org
Author: jmesnil
Date: 2009-12-07 11:31:21 -0500 (Mon, 07 Dec 2009)
New Revision: 8609
Modified:
trunk/src/main/org/hornetq/core/client/ClientSession.java
trunk/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java
trunk/src/main/org/hornetq/core/client/impl/DelegatingSession.java
trunk/src/main/org/hornetq/jms/client/HornetQSession.java
trunk/src/main/org/hornetq/ra/inflow/HornetQMessageHandler.java
trunk/tests/jms-tests/src/org/hornetq/jms/tests/message/MessageHeaderTest.java
trunk/tests/src/org/hornetq/tests/integration/client/SessionTest.java
Log:
HORNETQ-185: review core API
* removed dependency from ClientSession API to wireformat classes
by introducing interfaces ClientSession.BindingQuery & ClientSession.QueueQuery
Modified: trunk/src/main/org/hornetq/core/client/ClientSession.java
===================================================================
--- trunk/src/main/org/hornetq/core/client/ClientSession.java 2009-12-07 15:48:53 UTC (rev 8608)
+++ trunk/src/main/org/hornetq/core/client/ClientSession.java 2009-12-07 16:31:21 UTC (rev 8609)
@@ -13,11 +13,12 @@
package org.hornetq.core.client;
+
+import java.util.List;
+
import javax.transaction.xa.XAResource;
import org.hornetq.core.exception.HornetQException;
-import org.hornetq.core.remoting.impl.wireformat.SessionBindingQueryResponseMessage;
-import org.hornetq.core.remoting.impl.wireformat.SessionQueueQueryResponseMessage;
import org.hornetq.utils.SimpleString;
/**
@@ -29,6 +30,28 @@
*/
public interface ClientSession extends XAResource
{
+ public interface BindingQuery
+ {
+ boolean isExists();
+
+ public List<SimpleString> getQueueNames();
+ }
+
+ public interface QueueQuery
+ {
+ boolean isExists();
+
+ boolean isDurable();
+
+ int getConsumerCount();
+
+ int getMessageCount();
+
+ SimpleString getFilterString();
+
+ SimpleString getAddress();
+ }
+
// Lifecycle operations ------------------------------------------
/**
@@ -399,9 +422,9 @@
// Query operations ----------------------------------------------
- SessionQueueQueryResponseMessage queueQuery(SimpleString queueName) throws HornetQException;
+ QueueQuery queueQuery(SimpleString queueName) throws HornetQException;
- SessionBindingQueryResponseMessage bindingQuery(SimpleString address) throws HornetQException;
+ BindingQuery bindingQuery(SimpleString address) throws HornetQException;
// Transaction operations ----------------------------------------
Modified: trunk/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java
===================================================================
--- trunk/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java 2009-12-07 15:48:53 UTC (rev 8608)
+++ trunk/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java 2009-12-07 16:31:21 UTC (rev 8609)
@@ -15,6 +15,7 @@
import static org.hornetq.core.exception.HornetQException.TRANSACTION_ROLLED_BACK;
import static org.hornetq.utils.SimpleString.toSimpleString;
+import java.util.ArrayList;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
@@ -104,7 +105,7 @@
private static final Logger log = Logger.getLogger(ClientSessionImpl.class);
private final boolean trace = log.isTraceEnabled();
-
+
// Attributes ----------------------------------------------------------------------------
private final FailoverManager failoverManager;
@@ -323,7 +324,7 @@
deleteQueue(toSimpleString(queueName));
}
- public SessionQueueQueryResponseMessage queueQuery(final SimpleString queueName) throws HornetQException
+ public QueueQuery queueQuery(final SimpleString queueName) throws HornetQException
{
checkClosed();
@@ -331,10 +332,10 @@
SessionQueueQueryResponseMessage response = (SessionQueueQueryResponseMessage)channel.sendBlocking(request);
- return response;
- }
+ return new QueueQueryImpl(response.isDurable(), response.getConsumerCount(), response.getMessageCount(), response.getFilterString(), response.getAddress(), response.isExists());
+ }
- public SessionBindingQueryResponseMessage bindingQuery(final SimpleString address) throws HornetQException
+ public BindingQuery bindingQuery(final SimpleString address) throws HornetQException
{
checkClosed();
@@ -342,7 +343,7 @@
SessionBindingQueryResponseMessage response = (SessionBindingQueryResponseMessage)channel.sendBlocking(request);
- return response;
+ return new BindingQueryImpl(response.isExists(), response.getQueueNames());
}
public void forceDelivery(long consumerID, long sequence) throws HornetQException
@@ -1588,4 +1589,84 @@
consumer.flushAcks();
}
}
+
+ private static class BindingQueryImpl implements BindingQuery
+ {
+
+ private final boolean exists;
+ private final ArrayList<SimpleString> queueNames;
+
+ public BindingQueryImpl(final boolean exists, List<SimpleString> queueNames)
+ {
+ this.exists = exists;
+ this.queueNames = new ArrayList<SimpleString>(queueNames);
+ }
+
+ public List<SimpleString> getQueueNames()
+ {
+ return queueNames;
+ }
+
+ public boolean isExists()
+ {
+ return exists;
+ }
+ }
+
+ private static class QueueQueryImpl implements QueueQuery
+ {
+
+ private final boolean exists;
+ private final boolean durable;
+ private final int messageCount;
+ private final SimpleString filterString;
+ private final int consumerCount;
+ private final SimpleString address;
+
+ public QueueQueryImpl(final boolean durable,
+ final int consumerCount,
+ final int messageCount,
+ final SimpleString filterString,
+ final SimpleString address,
+ final boolean exists)
+ {
+
+ this.durable = durable;
+ this.consumerCount = consumerCount;
+ this.messageCount = messageCount;
+ this.filterString = filterString;
+ this.address = address;
+ this.exists = exists;
+ }
+ public SimpleString getAddress()
+ {
+ return address;
+ }
+
+ public int getConsumerCount()
+ {
+ return consumerCount;
+ }
+
+ public SimpleString getFilterString()
+ {
+ return filterString;
+ }
+
+ public int getMessageCount()
+ {
+ return messageCount;
+ }
+
+ public boolean isDurable()
+ {
+ return durable;
+ }
+
+ public boolean isExists()
+ {
+ return exists;
+ }
+
+ }
}
Modified: trunk/src/main/org/hornetq/core/client/impl/DelegatingSession.java
===================================================================
--- trunk/src/main/org/hornetq/core/client/impl/DelegatingSession.java 2009-12-07 15:48:53 UTC (rev 8608)
+++ trunk/src/main/org/hornetq/core/client/impl/DelegatingSession.java 2009-12-07 16:31:21 UTC (rev 8609)
@@ -27,8 +27,6 @@
import org.hornetq.core.exception.HornetQException;
import org.hornetq.core.logging.Logger;
import org.hornetq.core.remoting.RemotingConnection;
-import org.hornetq.core.remoting.impl.wireformat.SessionBindingQueryResponseMessage;
-import org.hornetq.core.remoting.impl.wireformat.SessionQueueQueryResponseMessage;
import org.hornetq.core.remoting.impl.wireformat.SessionReceiveContinuationMessage;
import org.hornetq.core.remoting.impl.wireformat.SessionReceiveLargeMessage;
import org.hornetq.core.remoting.impl.wireformat.SessionReceiveMessage;
@@ -115,7 +113,7 @@
session.addProducer(producer);
}
- public SessionBindingQueryResponseMessage bindingQuery(SimpleString address) throws HornetQException
+ public BindingQuery bindingQuery(SimpleString address) throws HornetQException
{
return session.bindingQuery(address);
}
@@ -403,7 +401,7 @@
return session.prepare(xid);
}
- public SessionQueueQueryResponseMessage queueQuery(SimpleString queueName) throws HornetQException
+ public QueueQuery queueQuery(SimpleString queueName) throws HornetQException
{
return session.queueQuery(queueName);
}
Modified: trunk/src/main/org/hornetq/jms/client/HornetQSession.java
===================================================================
--- trunk/src/main/org/hornetq/jms/client/HornetQSession.java 2009-12-07 15:48:53 UTC (rev 8608)
+++ trunk/src/main/org/hornetq/jms/client/HornetQSession.java 2009-12-07 16:31:21 UTC (rev 8609)
@@ -54,11 +54,11 @@
import org.hornetq.core.client.ClientConsumer;
import org.hornetq.core.client.ClientProducer;
import org.hornetq.core.client.ClientSession;
+import org.hornetq.core.client.ClientSession.BindingQuery;
+import org.hornetq.core.client.ClientSession.QueueQuery;
import org.hornetq.core.exception.HornetQException;
import org.hornetq.core.filter.impl.FilterImpl;
import org.hornetq.core.logging.Logger;
-import org.hornetq.core.remoting.impl.wireformat.SessionBindingQueryResponseMessage;
-import org.hornetq.core.remoting.impl.wireformat.SessionQueueQueryResponseMessage;
import org.hornetq.jms.HornetQDestination;
import org.hornetq.jms.HornetQQueue;
import org.hornetq.jms.HornetQTemporaryQueue;
@@ -331,7 +331,7 @@
{
if (jbd instanceof Queue)
{
- SessionQueueQueryResponseMessage response = session.queueQuery(jbd.getSimpleAddress());
+ QueueQuery response = session.queueQuery(jbd.getSimpleAddress());
if (!response.isExists())
{
@@ -340,7 +340,7 @@
}
else
{
- SessionBindingQueryResponseMessage response = session.bindingQuery(jbd.getSimpleAddress());
+ BindingQuery response = session.bindingQuery(jbd.getSimpleAddress());
if (!response.isExists())
{
@@ -408,7 +408,7 @@
try
{
- SessionQueueQueryResponseMessage response = session.queueQuery(queue.getSimpleAddress());
+ QueueQuery response = session.queueQuery(queue.getSimpleAddress());
if (!response.isExists())
{
@@ -437,9 +437,9 @@
try
{
- SessionBindingQueryResponseMessage response = session.bindingQuery(topic.getSimpleAddress());
+ BindingQuery query = session.bindingQuery(topic.getSimpleAddress());
- if (!response.isExists())
+ if (!query.isExists())
{
throw new JMSException("There is no topic with name " + topicName);
}
@@ -527,7 +527,7 @@
if (dest instanceof Queue)
{
- SessionQueueQueryResponseMessage response = session.queueQuery(dest.getSimpleAddress());
+ QueueQuery response = session.queueQuery(dest.getSimpleAddress());
if (!response.isExists())
{
@@ -538,7 +538,7 @@
}
else
{
- SessionBindingQueryResponseMessage response = session.bindingQuery(dest.getSimpleAddress());
+ BindingQuery response = session.bindingQuery(dest.getSimpleAddress());
if (!response.isExists())
{
@@ -576,7 +576,7 @@
queueName = new SimpleString(HornetQTopic.createQueueNameForDurableSubscription(connection.getClientID(),
subscriptionName));
- SessionQueueQueryResponseMessage subResponse = session.queueQuery(queueName);
+ QueueQuery subResponse = session.queueQuery(queueName);
if (!subResponse.isExists())
{
@@ -678,7 +678,7 @@
try
{
- SessionBindingQueryResponseMessage message = session.bindingQuery(new SimpleString(jbq.getAddress()));
+ BindingQuery message = session.bindingQuery(new SimpleString(jbq.getAddress()));
if (!message.isExists())
{
throw new InvalidDestinationException(jbq.getAddress() + " does not exist");
@@ -767,7 +767,7 @@
try
{
- SessionQueueQueryResponseMessage response = session.queueQuery(queueName);
+ QueueQuery response = session.queueQuery(queueName);
if (!response.isExists())
{
@@ -880,7 +880,7 @@
{
try
{
- SessionBindingQueryResponseMessage response = session.bindingQuery(tempTopic.getSimpleAddress());
+ BindingQuery response = session.bindingQuery(tempTopic.getSimpleAddress());
if (!response.isExists())
{
@@ -910,7 +910,7 @@
{
try
{
- SessionQueueQueryResponseMessage response = session.queueQuery(tempQueue.getSimpleAddress());
+ QueueQuery response = session.queueQuery(tempQueue.getSimpleAddress());
if (!response.isExists())
{
Modified: trunk/src/main/org/hornetq/ra/inflow/HornetQMessageHandler.java
===================================================================
--- trunk/src/main/org/hornetq/ra/inflow/HornetQMessageHandler.java 2009-12-07 15:48:53 UTC (rev 8608)
+++ trunk/src/main/org/hornetq/ra/inflow/HornetQMessageHandler.java 2009-12-07 16:31:21 UTC (rev 8609)
@@ -24,9 +24,9 @@
import org.hornetq.core.client.ClientMessage;
import org.hornetq.core.client.ClientSession;
import org.hornetq.core.client.MessageHandler;
+import org.hornetq.core.client.ClientSession.QueueQuery;
import org.hornetq.core.exception.HornetQException;
import org.hornetq.core.logging.Logger;
-import org.hornetq.core.remoting.impl.wireformat.SessionQueueQueryResponseMessage;
import org.hornetq.jms.HornetQTopic;
import org.hornetq.jms.client.HornetQMessage;
import org.hornetq.utils.SimpleString;
@@ -99,7 +99,7 @@
.getClientID(),
subscriptionName));
- SessionQueueQueryResponseMessage subResponse = session.queueQuery(queueName);
+ QueueQuery subResponse = session.queueQuery(queueName);
if (!subResponse.isExists())
{
Modified: trunk/tests/jms-tests/src/org/hornetq/jms/tests/message/MessageHeaderTest.java
===================================================================
--- trunk/tests/jms-tests/src/org/hornetq/jms/tests/message/MessageHeaderTest.java 2009-12-07 15:48:53 UTC (rev 8608)
+++ trunk/tests/jms-tests/src/org/hornetq/jms/tests/message/MessageHeaderTest.java 2009-12-07 16:31:21 UTC (rev 8609)
@@ -43,8 +43,6 @@
import org.hornetq.core.client.impl.ClientMessageImpl;
import org.hornetq.core.exception.HornetQException;
import org.hornetq.core.remoting.FailureListener;
-import org.hornetq.core.remoting.impl.wireformat.SessionBindingQueryResponseMessage;
-import org.hornetq.core.remoting.impl.wireformat.SessionQueueQueryResponseMessage;
import org.hornetq.jms.client.HornetQBytesMessage;
import org.hornetq.jms.client.HornetQMapMessage;
import org.hornetq.jms.client.HornetQMessage;
@@ -1003,12 +1001,12 @@
return null;
}
- public SessionQueueQueryResponseMessage queueQuery(SimpleString queueName) throws HornetQException
+ public QueueQuery queueQuery(SimpleString queueName) throws HornetQException
{
return null;
}
- public SessionBindingQueryResponseMessage bindingQuery(SimpleString address) throws HornetQException
+ public BindingQuery bindingQuery(SimpleString address) throws HornetQException
{
return null;
}
Modified: trunk/tests/src/org/hornetq/tests/integration/client/SessionTest.java
===================================================================
--- trunk/tests/src/org/hornetq/tests/integration/client/SessionTest.java 2009-12-07 15:48:53 UTC (rev 8608)
+++ trunk/tests/src/org/hornetq/tests/integration/client/SessionTest.java 2009-12-07 16:31:21 UTC (rev 8609)
@@ -22,11 +22,11 @@
import org.hornetq.core.client.ClientSession;
import org.hornetq.core.client.ClientSessionFactory;
import org.hornetq.core.client.SessionFailureListener;
+import org.hornetq.core.client.ClientSession.BindingQuery;
+import org.hornetq.core.client.ClientSession.QueueQuery;
import org.hornetq.core.client.impl.ClientSessionInternal;
import org.hornetq.core.exception.HornetQException;
import org.hornetq.core.remoting.RemotingConnection;
-import org.hornetq.core.remoting.impl.wireformat.SessionBindingQueryResponseMessage;
-import org.hornetq.core.remoting.impl.wireformat.SessionQueueQueryResponseMessage;
import org.hornetq.core.server.HornetQServer;
import org.hornetq.core.server.Queue;
import org.hornetq.tests.util.ServiceTestBase;
@@ -186,7 +186,7 @@
clientSession.createQueue("a2", "q3", false);
clientSession.createQueue("a2", "q4", false);
clientSession.createQueue("a2", "q5", false);
- SessionBindingQueryResponseMessage resp = clientSession.bindingQuery(new SimpleString("a"));
+ BindingQuery resp = clientSession.bindingQuery(new SimpleString("a"));
List<SimpleString> queues = resp.getQueueNames();
assertTrue(queues.isEmpty());
resp = clientSession.bindingQuery(new SimpleString("a1"));
@@ -225,7 +225,7 @@
ClientProducer cp = clientSession.createProducer("a1");
cp.send(clientSession.createClientMessage(false));
cp.send(clientSession.createClientMessage(false));
- SessionQueueQueryResponseMessage resp = clientSession.queueQuery(new SimpleString(queueName));
+ QueueQuery resp = clientSession.queueQuery(new SimpleString(queueName));
assertEquals(new SimpleString("a1"), resp.getAddress());
assertEquals(2, resp.getConsumerCount());
assertEquals(2, resp.getMessageCount());
@@ -252,7 +252,7 @@
clientSession.createQueue("a1", queueName, "foo=bar", false);
clientSession.createConsumer(queueName);
clientSession.createConsumer(queueName);
- SessionQueueQueryResponseMessage resp = clientSession.queueQuery(new SimpleString(queueName));
+ QueueQuery resp = clientSession.queueQuery(new SimpleString(queueName));
assertEquals(new SimpleString("a1"), resp.getAddress());
assertEquals(2, resp.getConsumerCount());
assertEquals(0, resp.getMessageCount());
@@ -276,7 +276,7 @@
server.start();
ClientSessionFactory cf = createInVMFactory();
ClientSession clientSession = cf.createSession(false, true, true);
- SessionQueueQueryResponseMessage resp = clientSession.queueQuery(new SimpleString(queueName));
+ QueueQuery resp = clientSession.queueQuery(new SimpleString(queueName));
assertFalse(resp.isExists());
assertEquals(null, resp.getAddress());
clientSession.close();
15 years, 1 month
JBoss hornetq SVN: r8608 - in trunk: tests/src/org/hornetq/tests/integration/journal and 1 other directory.
by do-not-reply＠jboss.org
Author: clebert.suconic(a)jboss.com
Date: 2009-12-07 10:48:53 -0500 (Mon, 07 Dec 2009)
New Revision: 8608
Modified:
trunk/src/main/org/hornetq/core/journal/impl/JournalImpl.java
trunk/tests/src/org/hornetq/tests/integration/journal/ValidateTransactionHealthTest.java
Log:
Tweaks to ValidateTransactionHealthTest. Changing System.exit to halt
Modified: trunk/src/main/org/hornetq/core/journal/impl/JournalImpl.java
===================================================================
--- trunk/src/main/org/hornetq/core/journal/impl/JournalImpl.java 2009-12-07 15:10:28 UTC (rev 8607)
+++ trunk/src/main/org/hornetq/core/journal/impl/JournalImpl.java 2009-12-07 15:48:53 UTC (rev 8608)
@@ -3255,7 +3255,7 @@
if (leftFiles.size() > 0)
{
- log.warn("Compacted files were left unnatended on journal directory, deleting invalid files now");
+ log.warn("Temporary files were left unnatended after a crash on journal directory, deleting invalid files now");
for (String fileToDelete : leftFiles)
{
Modified: trunk/tests/src/org/hornetq/tests/integration/journal/ValidateTransactionHealthTest.java
===================================================================
--- trunk/tests/src/org/hornetq/tests/integration/journal/ValidateTransactionHealthTest.java 2009-12-07 15:10:28 UTC (rev 8607)
+++ trunk/tests/src/org/hornetq/tests/integration/journal/ValidateTransactionHealthTest.java 2009-12-07 15:48:53 UTC (rev 8608)
@@ -313,7 +313,8 @@
System.exit(-1);
}
- System.exit(OK);
+ // Simulating a crash on the system right after the data was committed.
+ Runtime.getRuntime().halt(OK);
}
public static JournalImpl appendData(String journalType,
15 years, 1 month
JBoss hornetq SVN: r8607 - trunk/src/main/org/hornetq/core/client.
by do-not-reply＠jboss.org
Author: jmesnil
Date: 2009-12-07 10:10:28 -0500 (Mon, 07 Dec 2009)
New Revision: 8607
Modified:
trunk/src/main/org/hornetq/core/client/ClientRequestor.java
Log:
HORNETQ-186: fill in Javadocs for core API
* documented API
Modified: trunk/src/main/org/hornetq/core/client/ClientRequestor.java
===================================================================
--- trunk/src/main/org/hornetq/core/client/ClientRequestor.java 2009-12-07 14:53:52 UTC (rev 8606)
+++ trunk/src/main/org/hornetq/core/client/ClientRequestor.java 2009-12-07 15:10:28 UTC (rev 8607)
@@ -21,7 +21,14 @@
/**
- * a ClientRequestor.
+ * The ClientRequestor class helps making requests.
+ *
+ * The ClientRequestor constructor is given a ClientSession and a request address.
+ * It creates a temporary queue for the responses and provides a request method that sends the request message and waits for its reply.
+ *
+ * @apiviz.uses org.hornetq.core.client.ClientSession
+ * @apiviz.owns org.hornetq.core.client.ClientProducer
+ * @apiviz.owns org.hornetq.core.client.ClientConsumer
*
* @author <a href="jmesnil(a)redhat.com">Jeff Mesnil</a>
*/
@@ -35,6 +42,15 @@
private final SimpleString replyQueue;
+ /**
+ * Constructor for the ClientRequestor.
+ *
+ * The implementation expects a ClientSession with automatic commits of sends and acknowledgements
+ *
+ * @param session a ClientSession uses to handle requests and replies
+ * @param requestAddress the address to send request messages to
+ * @throws Exception
+ */
public ClientRequestor(final ClientSession session, final SimpleString requestAddress) throws Exception
{
queueSession = session;
@@ -45,16 +61,36 @@
replyConsumer = queueSession.createConsumer(replyQueue);
}
+ /**
+ * @see ClientRequestor#ClientRequestor(ClientSession, SimpleString)
+ */
public ClientRequestor(final ClientSession session, final String requestAddress) throws Exception
{
this(session, toSimpleString(requestAddress));
}
+ /**
+ * Send a message to the request address and wait indefinitely for a reply.
+ * The temporary queue is used for the REPLYTO_HEADER_NAME, and only one reply per request is expected
+ *
+ * @param request the message to send
+ * @return the reply message
+ * @throws Exception
+ */
public ClientMessage request(final ClientMessage request) throws Exception
{
return request(request, 0);
}
+ /**
+ * Send a message to the request address and wait for the given timeout for a reply.
+ * The temporary queue is used for the REPLYTO_HEADER_NAME, and only one reply per request is expected
+ *
+ * @param request the message to send
+ * @param timeout the timeout to wait for a reply (in milliseconds)
+ * @return the reply message or <code>null</code> if no message is replied before the timeout elapses
+ * @throws Exception
+ */
public ClientMessage request(final ClientMessage request, final long timeout) throws Exception
{
request.putStringProperty(ClientMessageImpl.REPLYTO_HEADER_NAME, replyQueue);
@@ -62,6 +98,11 @@
return replyConsumer.receive(timeout);
}
+ /**
+ * Close the ClientRequestor and its session.
+ *
+ * @throws Exception if an exception occurs while closing the ClientRequestor
+ */
public void close() throws Exception
{
replyConsumer.close();
15 years, 1 month