[jboss-cvs] JBoss Messaging SVN: r4425 - in trunk: src/main/org/jboss/messaging/core/client/impl and 5 other directories.
jboss-cvs-commits at lists.jboss.org
jboss-cvs-commits at lists.jboss.org
Tue Jun 10 18:49:14 EDT 2008
Author: timfox
Date: 2008-06-10 18:49:14 -0400 (Tue, 10 Jun 2008)
New Revision: 4425
Added:
trunk/tests/src/org/jboss/messaging/tests/unit/core/client/impl/ClientSessionImplTest.java
Modified:
trunk/src/main/org/jboss/messaging/core/client/ClientConnection.java
trunk/src/main/org/jboss/messaging/core/client/ClientSession.java
trunk/src/main/org/jboss/messaging/core/client/impl/ClientConnectionImpl.java
trunk/src/main/org/jboss/messaging/core/client/impl/ClientConnectionInternal.java
trunk/src/main/org/jboss/messaging/core/client/impl/ClientConsumerImpl.java
trunk/src/main/org/jboss/messaging/core/client/impl/ClientConsumerInternal.java
trunk/src/main/org/jboss/messaging/core/client/impl/ClientConsumerPacketHandler.java
trunk/src/main/org/jboss/messaging/core/client/impl/ClientSessionImpl.java
trunk/src/main/org/jboss/messaging/core/remoting/PacketDispatcher.java
trunk/src/main/org/jboss/messaging/core/remoting/impl/PacketDispatcherImpl.java
trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/ConsumerFlowCreditMessage.java
trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionAddDestinationMessage.java
trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionBindingQueryMessage.java
trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionCreateQueueMessage.java
trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionDeleteQueueMessage.java
trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionQueueQueryMessage.java
trunk/tests/src/org/jboss/messaging/tests/unit/core/remoting/PacketDispatcherTest.java
Log:
ClientSessionImpl test
Modified: trunk/src/main/org/jboss/messaging/core/client/ClientConnection.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/client/ClientConnection.java 2008-06-10 15:23:36 UTC (rev 4424)
+++ trunk/src/main/org/jboss/messaging/core/client/ClientConnection.java 2008-06-10 22:49:14 UTC (rev 4425)
@@ -34,4 +34,6 @@
boolean isClosed();
Version getServerVersion();
+
+ ClientConnectionFactory getConnectionFactory();
}
Modified: trunk/src/main/org/jboss/messaging/core/client/ClientSession.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/client/ClientSession.java 2008-06-10 15:23:36 UTC (rev 4424)
+++ trunk/src/main/org/jboss/messaging/core/client/ClientSession.java 2008-06-10 22:49:14 UTC (rev 4425)
@@ -33,11 +33,15 @@
SessionQueueQueryResponseMessage queueQuery(SimpleString queueName) throws MessagingException;
SessionBindingQueryResponseMessage bindingQuery(SimpleString address) throws MessagingException;
+
+ ClientConsumer createConsumer(SimpleString queueName) throws MessagingException;
ClientConsumer createConsumer(SimpleString queueName, SimpleString filterString, boolean noLocal,
boolean autoDeleteQueue, boolean direct) throws MessagingException;
- ClientConsumer createConsumer(SimpleString queueName) throws MessagingException;
+ ClientConsumer createConsumer(SimpleString queueName, SimpleString filterString, boolean noLocal,
+ boolean autoDeleteQueue, boolean direct,
+ int windowSize, int maxRate) throws MessagingException;
ClientBrowser createBrowser(SimpleString queueName, SimpleString filterString) throws MessagingException;
@@ -72,4 +76,6 @@
int getLazyAckBatchSize();
boolean isXA();
+
+ ClientConnection getConnection();
}
Modified: trunk/src/main/org/jboss/messaging/core/client/impl/ClientConnectionImpl.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/client/impl/ClientConnectionImpl.java 2008-06-10 15:23:36 UTC (rev 4424)
+++ trunk/src/main/org/jboss/messaging/core/client/impl/ClientConnectionImpl.java 2008-06-10 22:49:14 UTC (rev 4425)
@@ -103,11 +103,7 @@
ClientSession session =
new ClientSessionImpl(this, response.getSessionID(), xa, ackBatchSize, cacheProducers,
- autoCommitSends, autoCommitAcks, blockOnAcknowledge, connectionFactory.isDefaultBlockOnNonPersistentSend(),
- connectionFactory.isDefaultBlockOnPersistentSend(),
- connectionFactory.getDefaultConsumerWindowSize(), connectionFactory.getDefaultConsumerMaxRate(),
- connectionFactory.getDefaultProducerWindowSize(),
- connectionFactory.getDefaultProducerMaxRate());
+ autoCommitSends, autoCommitAcks, blockOnAcknowledge);
sessions.add(session);
@@ -191,6 +187,11 @@
{
return serverVersion;
}
+
+ public ClientConnectionFactory getConnectionFactory()
+ {
+ return connectionFactory;
+ }
// Public ---------------------------------------------------------------------------------------
Modified: trunk/src/main/org/jboss/messaging/core/client/impl/ClientConnectionInternal.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/client/impl/ClientConnectionInternal.java 2008-06-10 15:23:36 UTC (rev 4424)
+++ trunk/src/main/org/jboss/messaging/core/client/impl/ClientConnectionInternal.java 2008-06-10 22:49:14 UTC (rev 4425)
@@ -9,6 +9,7 @@
import java.util.Set;
import org.jboss.messaging.core.client.ClientConnection;
+import org.jboss.messaging.core.client.ClientConnectionFactory;
import org.jboss.messaging.core.client.ClientSession;
import org.jboss.messaging.core.remoting.RemotingConnection;
Modified: trunk/src/main/org/jboss/messaging/core/client/impl/ClientConsumerImpl.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/client/impl/ClientConsumerImpl.java 2008-06-10 15:23:36 UTC (rev 4424)
+++ trunk/src/main/org/jboss/messaging/core/client/impl/ClientConsumerImpl.java 2008-06-10 22:49:14 UTC (rev 4425)
@@ -87,8 +87,7 @@
private volatile int creditsToSend;
-
-
+
// Constructors
// ---------------------------------------------------------------------------------
@@ -292,7 +291,7 @@
public long getClientTargetID()
{
- return targetID;
+ return this.clientTargetID;
}
public void handleMessage(final ClientMessage message) throws Exception
@@ -348,9 +347,7 @@
synchronized (this)
{
- buffer.addLast(message, message.getPriority());
-
- maxSize = Math.max(maxSize, buffer.size());
+ buffer.addLast(message, message.getPriority());
}
queueExecutor();
@@ -368,14 +365,17 @@
}
}
- int maxSize = 0;
-
public void recover(final long lastDeliveryID)
{
ignoreDeliveryMark = lastDeliveryID;
buffer.clear();
}
+
+ public int getClientWindowSize()
+ {
+ return clientWindowSize;
+ }
// Public
// ---------------------------------------------------------------------------------------
Modified: trunk/src/main/org/jboss/messaging/core/client/impl/ClientConsumerInternal.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/client/impl/ClientConsumerInternal.java 2008-06-10 15:23:36 UTC (rev 4424)
+++ trunk/src/main/org/jboss/messaging/core/client/impl/ClientConsumerInternal.java 2008-06-10 22:49:14 UTC (rev 4425)
@@ -24,4 +24,6 @@
void handleMessage(ClientMessage message) throws Exception;
void recover(long lastDeliveryID) throws MessagingException;
+
+ int getClientWindowSize();
}
Modified: trunk/src/main/org/jboss/messaging/core/client/impl/ClientConsumerPacketHandler.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/client/impl/ClientConsumerPacketHandler.java 2008-06-10 15:23:36 UTC (rev 4424)
+++ trunk/src/main/org/jboss/messaging/core/client/impl/ClientConsumerPacketHandler.java 2008-06-10 22:49:14 UTC (rev 4425)
@@ -9,6 +9,7 @@
/**
* @author <a href="mailto:jmesnil at redhat.com">Jeff Mesnil</a>
+ * @author <a href="mailto:tim.fox at jboss.com">Tim Fox</a>
*
* @version <tt>$Revision$</tt>
*
@@ -62,4 +63,16 @@
{
return "ClientConsumerPacketHandler[id=" + consumerID + "]";
}
+
+ public boolean equals(Object other)
+ {
+ if (other instanceof ClientConsumerPacketHandler == false)
+ {
+ return false;
+ }
+
+ ClientConsumerPacketHandler r = (ClientConsumerPacketHandler)other;
+
+ return r.consumerID == this.consumerID;
+ }
}
\ No newline at end of file
Modified: trunk/src/main/org/jboss/messaging/core/client/impl/ClientSessionImpl.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/client/impl/ClientSessionImpl.java 2008-06-10 15:23:36 UTC (rev 4424)
+++ trunk/src/main/org/jboss/messaging/core/client/impl/ClientSessionImpl.java 2008-06-10 22:49:14 UTC (rev 4425)
@@ -103,16 +103,18 @@
private final boolean cacheProducers;
- private final int defaultConsumerWindowSize;
+ private final ExecutorService executor;
- private final int defaultConsumerMaxRate;
+ private final RemotingConnection remotingConnection;
- private final int defaultProducerWindowSize;
+ private final Set<ClientBrowser> browsers = new HashSet<ClientBrowser>();
- private final int defaultProducerMaxRate;
-
- private final ExecutorService executor;
+ private final Set<ClientProducer> producers = new HashSet<ClientProducer>();
+ private final Map<Long, ClientConsumerInternal> consumers = new HashMap<Long, ClientConsumerInternal>();
+
+ private final Map<SimpleString, ClientProducerInternal> producerCache;
+
private volatile boolean closed;
private boolean acked = true;
@@ -126,20 +128,7 @@
private long deliverID;
private boolean deliveryExpired;
-
- private final RemotingConnection remotingConnection;
- private final Set<ClientBrowser> browsers = new HashSet<ClientBrowser>();
-
- private final Set<ClientProducer> producers = new HashSet<ClientProducer>();
-
- private final Map<Long, ClientConsumerInternal> consumers = new HashMap<Long, ClientConsumerInternal>();
-
- private final Map<SimpleString, ClientProducerInternal> producerCache;
-
- //For testing only
- private boolean forceNotSameRM;
-
private long lastCommittedID = -1;
private final boolean autoCommitAcks;
@@ -148,9 +137,9 @@
private final boolean blockOnAcknowledge;
- private final boolean sendNonPersistentMessagesBlocking;
+ //For testing only
+ private boolean forceNotSameRM;
- private final boolean sendPersistentMessagesBlocking;
// Constructors ---------------------------------------------------------------------------------
@@ -158,13 +147,7 @@
final boolean xa,
final int lazyAckBatchSize, final boolean cacheProducers,
final boolean autoCommitSends, final boolean autoCommitAcks,
- final boolean blockOnAcknowledge,
- final boolean sendNonPersistentMessagesBlocking,
- final boolean sendPersistentMessagesBlocking,
- final int defaultConsumerWindowSize,
- final int defaultConsumerMaxRate,
- final int defaultProducerWindowSize,
- final int defaultProducerMaxRate) throws MessagingException
+ final boolean blockOnAcknowledge) throws MessagingException
{
if (lazyAckBatchSize < -1 || lazyAckBatchSize == 0)
{
@@ -179,14 +162,7 @@
this.cacheProducers = cacheProducers;
- this.defaultConsumerWindowSize = defaultConsumerWindowSize;
-
- this.defaultConsumerMaxRate = defaultConsumerMaxRate;
-
- this.defaultProducerWindowSize = defaultProducerWindowSize;
-
- this.defaultProducerMaxRate = defaultProducerMaxRate;
-
+ //TODO - we should use OrderedExecutorFactory and a pool here
executor = Executors.newSingleThreadExecutor();
this.xa = xa;
@@ -207,10 +183,6 @@
this.autoCommitSends = autoCommitSends;
this.blockOnAcknowledge = blockOnAcknowledge;
-
- this.sendNonPersistentMessagesBlocking = sendNonPersistentMessagesBlocking;
-
- this.sendPersistentMessagesBlocking = sendPersistentMessagesBlocking;
}
// ClientSession implementation -----------------------------------------------------------------
@@ -281,37 +253,48 @@
public ClientConsumer createConsumer(final SimpleString queueName, final SimpleString filterString, final boolean noLocal,
final boolean autoDeleteQueue, final boolean direct) throws MessagingException
{
+ return createConsumer(queueName, filterString, noLocal, autoDeleteQueue, direct,
+ connection.getConnectionFactory().getDefaultConsumerWindowSize(),
+ connection.getConnectionFactory().getDefaultConsumerMaxRate());
+ }
+
+ public ClientConsumer createConsumer(final SimpleString queueName, final SimpleString filterString, final boolean noLocal,
+ final boolean autoDeleteQueue, final boolean direct,
+ final int windowSize, final int maxRate) throws MessagingException
+ {
checkClosed();
long clientTargetID = remotingConnection.getPacketDispatcher().generateID();
SessionCreateConsumerMessage request =
new SessionCreateConsumerMessage(clientTargetID, queueName, filterString, noLocal, autoDeleteQueue,
- defaultConsumerWindowSize, defaultConsumerMaxRate);
-
+ windowSize, maxRate);
+
SessionCreateConsumerResponseMessage response = (SessionCreateConsumerResponseMessage)remotingConnection.sendBlocking(serverTargetID, serverTargetID, request);
- int windowSize = response.getWindowSize();
+ //The actual windows size that gets used is determined by the user since could be overridden on the queue settings
+ //The value we send is just a hint
+ int actualWindowSize = response.getWindowSize();
int clientWindowSize;
- if (windowSize == -1)
+ if (actualWindowSize == -1)
{
//No flow control - buffer can increase without bound! Only use with caution for very fast consumers
clientWindowSize = 0;
}
- else if (windowSize == 1)
+ else if (actualWindowSize == 1)
{
//Slow consumer - no buffering
clientWindowSize = 1;
}
- else if (windowSize > 1)
+ else if (actualWindowSize > 1)
{
//Client window size is half server window size
- clientWindowSize = windowSize >> 1;
+ clientWindowSize = actualWindowSize >> 1;
}
else
{
- throw new IllegalArgumentException("Invalid window size " + windowSize);
+ throw new IllegalArgumentException("Invalid window size " + actualWindowSize);
}
ClientConsumerInternal consumer =
@@ -319,13 +302,13 @@
consumers.put(response.getConsumerTargetID(), consumer);
-
remotingConnection.getPacketDispatcher().register(new ClientConsumerPacketHandler(consumer, clientTargetID));
//Now we send window size credits to start the consumption
//We even send it if windowSize == -1, since we need to start the consumer
-
- remotingConnection.sendOneWay(response.getConsumerTargetID(), serverTargetID, new ConsumerFlowCreditMessage(response.getWindowSize()));
+
+ remotingConnection.sendOneWay(response.getConsumerTargetID(), serverTargetID,
+ new ConsumerFlowCreditMessage(response.getWindowSize()));
return consumer;
}
@@ -352,7 +335,8 @@
public ClientProducer createProducer(final SimpleString address) throws MessagingException
{
- return createProducer(address, defaultProducerWindowSize, defaultProducerMaxRate);
+ return createProducer(address, connection.getConnectionFactory().getDefaultProducerWindowSize(),
+ connection.getConnectionFactory().getDefaultProducerMaxRate());
}
public ClientProducer createProducer(final SimpleString address, final int windowSize, final int maxRate) throws MessagingException
@@ -380,8 +364,8 @@
producer = new ClientProducerImpl(this, response.getProducerTargetID(), clientTargetID, address,
remotingConnection,
response.getMaxRate(),
- sendNonPersistentMessagesBlocking,
- autoCommitSends && sendPersistentMessagesBlocking,
+ connection.getConnectionFactory().isDefaultBlockOnNonPersistentSend(),
+ autoCommitSends && connection.getConnectionFactory().isDefaultBlockOnPersistentSend(),
response.getInitialCredits());
remotingConnection.getPacketDispatcher().register(new ClientProducerPacketHandler(producer, clientTargetID));
@@ -595,8 +579,6 @@
browsers.remove(browser);
}
-
-
// XAResource implementation --------------------------------------------------------------------
public void commit(final Xid xid, final boolean onePhase) throws XAException
Modified: trunk/src/main/org/jboss/messaging/core/remoting/PacketDispatcher.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/remoting/PacketDispatcher.java 2008-06-10 15:23:36 UTC (rev 4424)
+++ trunk/src/main/org/jboss/messaging/core/remoting/PacketDispatcher.java 2008-06-10 22:49:14 UTC (rev 4425)
@@ -6,8 +6,6 @@
*/
package org.jboss.messaging.core.remoting;
-
-
/**
* @author <a href="mailto:jmesnil at redhat.com">Jeff Mesnil</a>
*
Modified: trunk/src/main/org/jboss/messaging/core/remoting/impl/PacketDispatcherImpl.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/remoting/impl/PacketDispatcherImpl.java 2008-06-10 15:23:36 UTC (rev 4424)
+++ trunk/src/main/org/jboss/messaging/core/remoting/impl/PacketDispatcherImpl.java 2008-06-10 22:49:14 UTC (rev 4425)
@@ -97,8 +97,7 @@
if (handler == null)
{
- log.warn("no handler defined for " + handlerID);
- dump();
+ throw new IllegalArgumentException("Failed to unregister handler " + handlerID);
}
if (trace)
{
Modified: trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/ConsumerFlowCreditMessage.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/ConsumerFlowCreditMessage.java 2008-06-10 15:23:36 UTC (rev 4424)
+++ trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/ConsumerFlowCreditMessage.java 2008-06-10 22:49:14 UTC (rev 4425)
@@ -60,6 +60,18 @@
{
return getParentString() + ", credits=" + credits + "]";
}
+
+ public boolean equals(Object other)
+ {
+ if (other instanceof ConsumerFlowCreditMessage == false)
+ {
+ return false;
+ }
+
+ ConsumerFlowCreditMessage r = (ConsumerFlowCreditMessage)other;
+
+ return this.credits == r.credits;
+ }
// Package protected ---------------------------------------------
// Protected -----------------------------------------------------
Modified: trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionAddDestinationMessage.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionAddDestinationMessage.java 2008-06-10 15:23:36 UTC (rev 4424)
+++ trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionAddDestinationMessage.java 2008-06-10 22:49:14 UTC (rev 4425)
@@ -75,6 +75,19 @@
return getParentString() + ", address=" + address + ", temp=" + temporary +"]";
}
+ public boolean equals(Object other)
+ {
+ if (other instanceof SessionAddDestinationMessage == false)
+ {
+ return false;
+ }
+
+ SessionAddDestinationMessage r = (SessionAddDestinationMessage)other;
+
+ return this.address.equals(r.address) &&
+ this.temporary == r.temporary;
+ }
+
// Package protected ---------------------------------------------
// Protected -----------------------------------------------------
Modified: trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionBindingQueryMessage.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionBindingQueryMessage.java 2008-06-10 15:23:36 UTC (rev 4424)
+++ trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionBindingQueryMessage.java 2008-06-10 22:49:14 UTC (rev 4425)
@@ -48,4 +48,16 @@
address = buffer.getSimpleString();
}
+ public boolean equals(Object other)
+ {
+ if (other instanceof SessionBindingQueryMessage == false)
+ {
+ return false;
+ }
+
+ SessionBindingQueryMessage r = (SessionBindingQueryMessage)other;
+
+ return this.address.equals(r.address);
+ }
+
}
Modified: trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionCreateQueueMessage.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionCreateQueueMessage.java 2008-06-10 15:23:36 UTC (rev 4424)
+++ trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionCreateQueueMessage.java 2008-06-10 22:49:14 UTC (rev 4425)
@@ -9,7 +9,6 @@
import org.jboss.messaging.util.MessagingBuffer;
import org.jboss.messaging.util.SimpleString;
-
/**
* @author <a href="mailto:tim.fox at jboss.com">Tim Fox</a>
@@ -105,6 +104,22 @@
durable = buffer.getBoolean();
temporary = buffer.getBoolean();
}
+
+ public boolean equals(Object other)
+ {
+ if (other instanceof SessionCreateQueueMessage == false)
+ {
+ return false;
+ }
+
+ SessionCreateQueueMessage r = (SessionCreateQueueMessage)other;
+
+ return r.address.equals(this.address) &&
+ r.queueName.equals(this.queueName) &&
+ (r.filterString == null ? this.filterString == null : r.filterString.equals(this.filterString)) &&
+ r.durable == this.durable &&
+ r.temporary == this.temporary;
+ }
// Package protected ---------------------------------------------
Modified: trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionDeleteQueueMessage.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionDeleteQueueMessage.java 2008-06-10 15:23:36 UTC (rev 4424)
+++ trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionDeleteQueueMessage.java 2008-06-10 22:49:14 UTC (rev 4425)
@@ -9,7 +9,6 @@
import org.jboss.messaging.util.MessagingBuffer;
import org.jboss.messaging.util.SimpleString;
-
/**
* @author <a href="mailto:tim.fox at jboss.com">Tim Fox</a>
@@ -64,6 +63,18 @@
{
queueName = buffer.getSimpleString();
}
+
+ public boolean equals(Object other)
+ {
+ if (other instanceof SessionDeleteQueueMessage == false)
+ {
+ return false;
+ }
+
+ SessionDeleteQueueMessage r = (SessionDeleteQueueMessage)other;
+
+ return r.queueName.equals(this.queueName);
+ }
// Package protected ---------------------------------------------
Modified: trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionQueueQueryMessage.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionQueueQueryMessage.java 2008-06-10 15:23:36 UTC (rev 4424)
+++ trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionQueueQueryMessage.java 2008-06-10 22:49:14 UTC (rev 4425)
@@ -48,4 +48,16 @@
queueName = buffer.getSimpleString();
}
+ public boolean equals(Object other)
+ {
+ if (other instanceof SessionQueueQueryMessage == false)
+ {
+ return false;
+ }
+
+ SessionQueueQueryMessage r = (SessionQueueQueryMessage)other;
+
+ return r.queueName.equals(this.queueName);
+ }
+
}
Added: trunk/tests/src/org/jboss/messaging/tests/unit/core/client/impl/ClientSessionImplTest.java
===================================================================
--- trunk/tests/src/org/jboss/messaging/tests/unit/core/client/impl/ClientSessionImplTest.java (rev 0)
+++ trunk/tests/src/org/jboss/messaging/tests/unit/core/client/impl/ClientSessionImplTest.java 2008-06-10 22:49:14 UTC (rev 4425)
@@ -0,0 +1,555 @@
+/*
+ * JBoss, Home of Professional Open Source
+ * Copyright 2005, JBoss Inc., and individual contributors as indicated
+ * by the @authors tag. See the copyright.txt in the distribution for a
+ * full listing of individual contributors.
+ *
+ * This is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU Lesser General Public License as
+ * published by the Free Software Foundation; either version 2.1 of
+ * the License, or (at your option) any later version.
+ *
+ * This software is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this software; if not, write to the Free
+ * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
+ * 02110-1301 USA, or see the FSF site: http://www.fsf.org.
+ */
+package org.jboss.messaging.tests.unit.core.client.impl;
+
+import org.easymock.EasyMock;
+import org.jboss.messaging.core.client.ClientConnectionFactory;
+import org.jboss.messaging.core.client.ClientSession;
+import org.jboss.messaging.core.client.impl.ClientConnectionFactoryImpl;
+import org.jboss.messaging.core.client.impl.ClientConnectionInternal;
+import org.jboss.messaging.core.client.impl.ClientConsumerInternal;
+import org.jboss.messaging.core.client.impl.ClientConsumerPacketHandler;
+import org.jboss.messaging.core.client.impl.ClientSessionImpl;
+import org.jboss.messaging.core.logging.Logger;
+import org.jboss.messaging.core.remoting.PacketDispatcher;
+import org.jboss.messaging.core.remoting.RemotingConnection;
+import org.jboss.messaging.core.remoting.impl.wireformat.ConsumerFlowCreditMessage;
+import org.jboss.messaging.core.remoting.impl.wireformat.SessionAddDestinationMessage;
+import org.jboss.messaging.core.remoting.impl.wireformat.SessionBindingQueryMessage;
+import org.jboss.messaging.core.remoting.impl.wireformat.SessionBindingQueryResponseMessage;
+import org.jboss.messaging.core.remoting.impl.wireformat.SessionCreateConsumerMessage;
+import org.jboss.messaging.core.remoting.impl.wireformat.SessionCreateConsumerResponseMessage;
+import org.jboss.messaging.core.remoting.impl.wireformat.SessionCreateQueueMessage;
+import org.jboss.messaging.core.remoting.impl.wireformat.SessionDeleteQueueMessage;
+import org.jboss.messaging.core.remoting.impl.wireformat.SessionQueueQueryMessage;
+import org.jboss.messaging.core.remoting.impl.wireformat.SessionQueueQueryResponseMessage;
+import org.jboss.messaging.core.remoting.impl.wireformat.SessionRemoveDestinationMessage;
+import org.jboss.messaging.tests.util.UnitTestCase;
+import org.jboss.messaging.util.SimpleString;
+
+/**
+ *
+ * A ClientSessionImplTest
+ *
+ * @author <a href="mailto:tim.fox at jboss.com">Tim Fox</a>
+ *
+ */
+public class ClientSessionImplTest extends UnitTestCase
+{
+ private static final Logger log = Logger.getLogger(ClientSessionImplTest.class);
+
+ // Private -----------------------------------------------------------------------------------------------------------
+
+ public void testConstructor() throws Exception
+ {
+ testConstructor(132, true, 10, true, true, true, true);
+ testConstructor(132, false, 10, false, false, false, false);
+ }
+
+ public void testConstructorInvalidArgs() throws Exception
+ {
+ ClientConnectionInternal conn = EasyMock.createStrictMock(ClientConnectionInternal.class);
+
+ try
+ {
+ new ClientSessionImpl(conn, 1, false, -2, false, false, false, false);
+ fail("Should throw exception");
+ }
+ catch (IllegalArgumentException e)
+ {
+ //Ok
+ }
+
+ try
+ {
+ new ClientSessionImpl(conn, 1, false, -10, false, false, false, false);
+ fail("Should throw exception");
+ }
+ catch (IllegalArgumentException e)
+ {
+ //Ok
+ }
+
+ try
+ {
+ new ClientSessionImpl(conn, 1, false, 0, false, false, false, false);
+ fail("Should throw exception");
+ }
+ catch (IllegalArgumentException e)
+ {
+ //Ok
+ }
+ }
+
+ public void testCreateQueue() throws Exception
+ {
+ ClientConnectionInternal conn = EasyMock.createStrictMock(ClientConnectionInternal.class);
+
+ RemotingConnection rc = EasyMock.createStrictMock(RemotingConnection.class);
+
+ EasyMock.expect(conn.getRemotingConnection()).andReturn(rc);
+
+ SessionCreateQueueMessage request = new SessionCreateQueueMessage(new SimpleString("blah"), new SimpleString("hagshg"),
+ new SimpleString("jhjhs"), false, false);
+
+ final int targetID = 121;
+
+ EasyMock.expect(rc.sendBlocking(targetID, targetID, request)).andReturn(null);
+
+ EasyMock.replay(conn);
+ EasyMock.replay(rc);
+
+ ClientSession session = new ClientSessionImpl(conn, targetID, false, -1, false, false, false, false);
+
+ session.createQueue(request.getAddress(), request.getQueueName(), request.getFilterString(), request.isDurable(), request.isTemporary());
+
+ EasyMock.verify(conn);
+ EasyMock.verify(rc);
+ }
+
+ public void testDeleteQueue() throws Exception
+ {
+ ClientConnectionInternal conn = EasyMock.createStrictMock(ClientConnectionInternal.class);
+
+ RemotingConnection rc = EasyMock.createStrictMock(RemotingConnection.class);
+
+ EasyMock.expect(conn.getRemotingConnection()).andReturn(rc);
+
+ SessionDeleteQueueMessage request = new SessionDeleteQueueMessage(new SimpleString("blah"));
+
+ final int targetID = 121;
+
+ EasyMock.expect(rc.sendBlocking(targetID, targetID, request)).andReturn(null);
+
+ EasyMock.replay(conn);
+ EasyMock.replay(rc);
+
+ ClientSession session = new ClientSessionImpl(conn, targetID, false, -1, false, false, false, false);
+
+ session.deleteQueue(request.getQueueName());
+
+ EasyMock.verify(conn);
+ EasyMock.verify(rc);
+ }
+
+ public void testQueueQuery() throws Exception
+ {
+ ClientConnectionInternal conn = EasyMock.createStrictMock(ClientConnectionInternal.class);
+
+ RemotingConnection rc = EasyMock.createStrictMock(RemotingConnection.class);
+
+ EasyMock.expect(conn.getRemotingConnection()).andReturn(rc);
+
+ SessionQueueQueryMessage request = new SessionQueueQueryMessage(new SimpleString("blah"));
+
+ SessionQueueQueryResponseMessage resp = new SessionQueueQueryResponseMessage();
+
+ final int targetID = 121;
+
+ EasyMock.expect(rc.sendBlocking(targetID, targetID, request)).andReturn(resp);
+
+ EasyMock.replay(conn);
+ EasyMock.replay(rc);
+
+ ClientSession session = new ClientSessionImpl(conn, targetID, false, -1, false, false, false, false);
+
+ SessionQueueQueryResponseMessage resp2 = session.queueQuery(request.getQueueName());
+
+ EasyMock.verify(conn);
+ EasyMock.verify(rc);
+
+ assertTrue(resp == resp2);
+ }
+
+ public void testBindingQuery() throws Exception
+ {
+ ClientConnectionInternal conn = EasyMock.createStrictMock(ClientConnectionInternal.class);
+
+ RemotingConnection rc = EasyMock.createStrictMock(RemotingConnection.class);
+
+ EasyMock.expect(conn.getRemotingConnection()).andReturn(rc);
+
+ SessionBindingQueryMessage request = new SessionBindingQueryMessage(new SimpleString("blah"));
+
+ SessionBindingQueryResponseMessage resp = new SessionBindingQueryResponseMessage();
+
+ final int targetID = 121;
+
+ EasyMock.expect(rc.sendBlocking(targetID, targetID, request)).andReturn(resp);
+
+ EasyMock.replay(conn);
+ EasyMock.replay(rc);
+
+ ClientSession session = new ClientSessionImpl(conn, targetID, false, -1, false, false, false, false);
+
+ SessionBindingQueryResponseMessage resp2 = session.bindingQuery(request.getAddress());
+
+ EasyMock.verify(conn);
+ EasyMock.verify(rc);
+
+ assertTrue(resp == resp2);
+ }
+
+ public void testAddDestination() throws Exception
+ {
+ ClientConnectionInternal conn = EasyMock.createStrictMock(ClientConnectionInternal.class);
+
+ RemotingConnection rc = EasyMock.createStrictMock(RemotingConnection.class);
+
+ EasyMock.expect(conn.getRemotingConnection()).andReturn(rc);
+
+ SessionAddDestinationMessage request = new SessionAddDestinationMessage(new SimpleString("blah"), true);
+
+ final int targetID = 121;
+
+ EasyMock.expect(rc.sendBlocking(targetID, targetID, request)).andReturn(null);
+
+ EasyMock.replay(conn);
+ EasyMock.replay(rc);
+
+ ClientSession session = new ClientSessionImpl(conn, targetID, false, -1, false, false, false, false);
+
+ session.addDestination(request.getAddress(), request.isTemporary());
+
+ EasyMock.verify(conn);
+ EasyMock.verify(rc);
+ }
+
+ public void testRemoveDestination() throws Exception
+ {
+ ClientConnectionInternal conn = EasyMock.createStrictMock(ClientConnectionInternal.class);
+
+ RemotingConnection rc = EasyMock.createStrictMock(RemotingConnection.class);
+
+ EasyMock.expect(conn.getRemotingConnection()).andReturn(rc);
+
+ SessionRemoveDestinationMessage request = new SessionRemoveDestinationMessage(new SimpleString("blah"), true);
+
+ final int targetID = 121;
+
+ EasyMock.expect(rc.sendBlocking(targetID, targetID, request)).andReturn(null);
+
+ EasyMock.replay(conn);
+ EasyMock.replay(rc);
+
+ ClientSession session = new ClientSessionImpl(conn, targetID, false, -1, false, false, false, false);
+
+ session.removeDestination(request.getAddress(), true);
+
+ EasyMock.verify(conn);
+ EasyMock.verify(rc);
+ }
+
+ public void testCreateConsumer() throws Exception
+ {
+ //First test with the wide createConsumer method
+
+ testCreateConsumerWideMethod(new SimpleString("usahduiahs"), new SimpleString("ygyggyg"),
+ false, false, false, 121455, 76556, 121455);
+ testCreateConsumerWideMethod(new SimpleString("usahduiahs"), null,
+ false, false, false, 121455, 76556, 121455);
+
+ //test where server window size overrides client window size
+ testCreateConsumerWideMethod(new SimpleString("usahduiahs"), null,
+ false, false, false, 121455, 76556, 675675765);
+ testCreateConsumerWideMethod(new SimpleString("usahduiahs"), null,
+ false, false, false, 121455, 76556, 1);
+ testCreateConsumerWideMethod(new SimpleString("usahduiahs"), null,
+ false, false, false, 121455, 76556, -1);
+
+ //And with the method that takes defaults from the cf
+
+ testCreateConsumerDefaultsMethod(new SimpleString("usahduiahs"), new SimpleString("ygyggyg"),
+ false, false, false, 121455, 76556, 121455);
+ testCreateConsumerDefaultsMethod(new SimpleString("usahduiahs"), null,
+ false, false, false, 121455, 76556, 121455);
+
+ //test where server window size overrides client window size
+ testCreateConsumerDefaultsMethod(new SimpleString("usahduiahs"), null,
+ false, false, false, 121455, 76556, 675675765);
+ testCreateConsumerDefaultsMethod(new SimpleString("usahduiahs"), null,
+ false, false, false, 121455, 76556, 1);
+ testCreateConsumerDefaultsMethod(new SimpleString("usahduiahs"), null,
+ false, false, false, 121455, 76556, -1);
+
+ // And with the basic createConsumer method:
+
+ testCreateConsumerBasicMethod(new SimpleString("usahduiahs"), 121455, 76556, 121455);
+ testCreateConsumerBasicMethod(new SimpleString("usahduiahs"), 121455, 76556, 121455);
+
+ //test where server window size overrides client window size
+ testCreateConsumerBasicMethod(new SimpleString("usahduiahs"), 121455, 76556, 675675765);
+ testCreateConsumerBasicMethod(new SimpleString("usahduiahs"), 121455, 76556, 1);
+ testCreateConsumerBasicMethod(new SimpleString("usahduiahs"), 121455, 76556, -1);
+ }
+
+ // Private -------------------------------------------------------------------------------------------
+
+ private void testCreateConsumerDefaultsMethod(final SimpleString queueName, final SimpleString filterString, final boolean noLocal,
+ final boolean autoDeleteQueue, final boolean direct,
+ final int windowSize, final int maxRate, final int serverWindowSize) throws Exception
+ {
+ ClientConnectionFactory cf = EasyMock.createStrictMock(ClientConnectionFactory.class);
+
+ ClientConnectionInternal conn = EasyMock.createStrictMock(ClientConnectionInternal.class);
+
+ RemotingConnection rc = EasyMock.createStrictMock(RemotingConnection.class);
+
+ PacketDispatcher pd = EasyMock.createStrictMock(PacketDispatcher.class);
+
+ EasyMock.expect(conn.getRemotingConnection()).andReturn(rc);
+
+ EasyMock.expect(conn.getConnectionFactory()).andReturn(cf);
+
+ EasyMock.expect(cf.getDefaultConsumerWindowSize()).andReturn(windowSize);
+
+ EasyMock.expect(conn.getConnectionFactory()).andReturn(cf);
+
+ EasyMock.expect(cf.getDefaultConsumerMaxRate()).andReturn(maxRate);
+
+ EasyMock.expect(rc.getPacketDispatcher()).andReturn(pd);
+
+ final long clientTargetID = 87126716;
+
+ EasyMock.expect(pd.generateID()).andReturn(clientTargetID);
+
+ final long sessionTargetID = 9121892;
+
+ SessionCreateConsumerMessage request =
+ new SessionCreateConsumerMessage(clientTargetID, queueName, filterString, noLocal, autoDeleteQueue,
+ windowSize, maxRate);
+
+ SessionCreateConsumerResponseMessage resp =
+ new SessionCreateConsumerResponseMessage(656652, serverWindowSize);
+
+ EasyMock.expect(rc.sendBlocking(sessionTargetID, sessionTargetID, request)).andReturn(resp);
+
+ EasyMock.expect(rc.getPacketDispatcher()).andReturn(pd);
+
+ pd.register(new ClientConsumerPacketHandler(null, clientTargetID));
+
+ rc.sendOneWay(resp.getConsumerTargetID(), sessionTargetID,
+ new ConsumerFlowCreditMessage(resp.getWindowSize()));
+
+ EasyMock.replay(cf);
+ EasyMock.replay(conn);
+ EasyMock.replay(rc);
+ EasyMock.replay(pd);
+
+ ClientSession session = new ClientSessionImpl(conn, sessionTargetID, false, -1, false, false, false, false);
+
+ ClientConsumerInternal consumer = (ClientConsumerInternal)session.createConsumer(queueName, filterString, noLocal, autoDeleteQueue,
+ direct);
+ EasyMock.verify(cf);
+ EasyMock.verify(conn);
+ EasyMock.verify(rc);
+ EasyMock.verify(pd);
+
+ assertEquals(clientTargetID, consumer.getClientTargetID());
+
+ if (serverWindowSize == -1)
+ {
+ assertEquals(0, consumer.getClientWindowSize());
+ }
+ else if (serverWindowSize == 1)
+ {
+ assertEquals(1, consumer.getClientWindowSize());
+ }
+ else if (serverWindowSize > 1)
+ {
+ assertEquals(serverWindowSize >> 1, consumer.getClientWindowSize());
+ }
+ }
+
+ private void testCreateConsumerWideMethod(final SimpleString queueName, final SimpleString filterString, final boolean noLocal,
+ final boolean autoDeleteQueue, final boolean direct,
+ final int windowSize, final int maxRate, final int serverWindowSize) throws Exception
+ {
+ ClientConnectionFactory cf = EasyMock.createStrictMock(ClientConnectionFactory.class);
+
+ ClientConnectionInternal conn = EasyMock.createStrictMock(ClientConnectionInternal.class);
+
+ RemotingConnection rc = EasyMock.createStrictMock(RemotingConnection.class);
+
+ PacketDispatcher pd = EasyMock.createStrictMock(PacketDispatcher.class);
+
+ EasyMock.expect(conn.getRemotingConnection()).andReturn(rc);
+
+ EasyMock.expect(rc.getPacketDispatcher()).andReturn(pd);
+
+ final long clientTargetID = 87126716;
+
+ EasyMock.expect(pd.generateID()).andReturn(clientTargetID);
+
+ final long sessionTargetID = 9121892;
+
+ SessionCreateConsumerMessage request =
+ new SessionCreateConsumerMessage(clientTargetID, queueName, filterString, noLocal, autoDeleteQueue,
+ windowSize, maxRate);
+
+ SessionCreateConsumerResponseMessage resp =
+ new SessionCreateConsumerResponseMessage(656652, serverWindowSize);
+
+ EasyMock.expect(rc.sendBlocking(sessionTargetID, sessionTargetID, request)).andReturn(resp);
+
+ EasyMock.expect(rc.getPacketDispatcher()).andReturn(pd);
+
+ pd.register(new ClientConsumerPacketHandler(null, clientTargetID));
+
+ rc.sendOneWay(resp.getConsumerTargetID(), sessionTargetID,
+ new ConsumerFlowCreditMessage(resp.getWindowSize()));
+
+ EasyMock.replay(cf);
+ EasyMock.replay(conn);
+ EasyMock.replay(rc);
+ EasyMock.replay(pd);
+
+ ClientSession session = new ClientSessionImpl(conn, sessionTargetID, false, -1, false, false, false, false);
+
+ ClientConsumerInternal consumer = (ClientConsumerInternal)session.createConsumer(queueName, filterString, noLocal, autoDeleteQueue,
+ direct, windowSize, maxRate);
+ EasyMock.verify(cf);
+ EasyMock.verify(conn);
+ EasyMock.verify(rc);
+ EasyMock.verify(pd);
+
+ assertEquals(clientTargetID, consumer.getClientTargetID());
+
+ if (serverWindowSize == -1)
+ {
+ assertEquals(0, consumer.getClientWindowSize());
+ }
+ else if (serverWindowSize == 1)
+ {
+ assertEquals(1, consumer.getClientWindowSize());
+ }
+ else if (serverWindowSize > 1)
+ {
+ assertEquals(serverWindowSize >> 1, consumer.getClientWindowSize());
+ }
+ }
+
+ private void testCreateConsumerBasicMethod(final SimpleString queueName, final int windowSize,
+ final int maxRate, final int serverWindowSize) throws Exception
+ {
+ ClientConnectionFactory cf = EasyMock.createStrictMock(ClientConnectionFactory.class);
+
+ ClientConnectionInternal conn = EasyMock.createStrictMock(ClientConnectionInternal.class);
+
+ RemotingConnection rc = EasyMock.createStrictMock(RemotingConnection.class);
+
+ PacketDispatcher pd = EasyMock.createStrictMock(PacketDispatcher.class);
+
+ EasyMock.expect(conn.getRemotingConnection()).andReturn(rc);
+
+ EasyMock.expect(conn.getConnectionFactory()).andReturn(cf);
+
+ EasyMock.expect(cf.getDefaultConsumerWindowSize()).andReturn(windowSize);
+
+ EasyMock.expect(conn.getConnectionFactory()).andReturn(cf);
+
+ EasyMock.expect(cf.getDefaultConsumerMaxRate()).andReturn(maxRate);
+
+ EasyMock.expect(rc.getPacketDispatcher()).andReturn(pd);
+
+ final long clientTargetID = 87126716;
+
+ EasyMock.expect(pd.generateID()).andReturn(clientTargetID);
+
+ final long sessionTargetID = 9121892;
+
+ SessionCreateConsumerMessage request =
+ new SessionCreateConsumerMessage(clientTargetID, queueName, null, false, false,
+ windowSize, maxRate);
+
+ SessionCreateConsumerResponseMessage resp =
+ new SessionCreateConsumerResponseMessage(656652, serverWindowSize);
+
+ EasyMock.expect(rc.sendBlocking(sessionTargetID, sessionTargetID, request)).andReturn(resp);
+
+ EasyMock.expect(rc.getPacketDispatcher()).andReturn(pd);
+
+ pd.register(new ClientConsumerPacketHandler(null, clientTargetID));
+
+ rc.sendOneWay(resp.getConsumerTargetID(), sessionTargetID,
+ new ConsumerFlowCreditMessage(resp.getWindowSize()));
+
+ EasyMock.replay(cf);
+ EasyMock.replay(conn);
+ EasyMock.replay(rc);
+ EasyMock.replay(pd);
+
+ ClientSession session = new ClientSessionImpl(conn, sessionTargetID, false, -1, false, false, false, false);
+
+ ClientConsumerInternal consumer = (ClientConsumerInternal)session.createConsumer(queueName);
+ EasyMock.verify(cf);
+ EasyMock.verify(conn);
+ EasyMock.verify(rc);
+ EasyMock.verify(pd);
+
+ assertEquals(clientTargetID, consumer.getClientTargetID());
+
+ if (serverWindowSize == -1)
+ {
+ assertEquals(0, consumer.getClientWindowSize());
+ }
+ else if (serverWindowSize == 1)
+ {
+ assertEquals(1, consumer.getClientWindowSize());
+ }
+ else if (serverWindowSize > 1)
+ {
+ assertEquals(serverWindowSize >> 1, consumer.getClientWindowSize());
+ }
+ }
+
+ private void testConstructor(final long serverTargetID,
+ final boolean xa,
+ final int lazyAckBatchSize, final boolean cacheProducers,
+ final boolean autoCommitSends, final boolean autoCommitAcks,
+ final boolean blockOnAcknowledge) throws Exception
+ {
+ ClientConnectionInternal conn = EasyMock.createStrictMock(ClientConnectionInternal.class);
+
+ RemotingConnection rc = EasyMock.createStrictMock(RemotingConnection.class);
+
+ EasyMock.expect(conn.getRemotingConnection()).andReturn(rc);
+
+ EasyMock.replay(conn);
+ EasyMock.replay(rc);
+
+ ClientSession session = new ClientSessionImpl(conn, serverTargetID, xa,
+ lazyAckBatchSize, cacheProducers, autoCommitSends, autoCommitAcks, blockOnAcknowledge);
+
+ EasyMock.verify(conn);
+ EasyMock.verify(rc);
+
+ assertTrue(conn == session.getConnection());
+ assertEquals(xa, session.isXA());
+ assertEquals(lazyAckBatchSize, session.getLazyAckBatchSize());
+ assertEquals(cacheProducers, session.isCacheProducers());
+ assertEquals(autoCommitSends, session.isAutoCommitSends());
+ assertEquals(autoCommitAcks, session.isAutoCommitAcks());
+ assertEquals(blockOnAcknowledge, session.isBlockOnAcknowledge());
+ }
+}
+
Modified: trunk/tests/src/org/jboss/messaging/tests/unit/core/remoting/PacketDispatcherTest.java
===================================================================
--- trunk/tests/src/org/jboss/messaging/tests/unit/core/remoting/PacketDispatcherTest.java 2008-06-10 15:23:36 UTC (rev 4424)
+++ trunk/tests/src/org/jboss/messaging/tests/unit/core/remoting/PacketDispatcherTest.java 2008-06-10 22:49:14 UTC (rev 4425)
@@ -62,7 +62,15 @@
replay(handler);
- dispatcher.unregister(id);
+ try
+ {
+ dispatcher.unregister(id);
+ fail("Should throw Exception");
+ }
+ catch (IllegalArgumentException e)
+ {
+ //Ok
+ }
assertNull(dispatcher.getHandler(id));
verify(handler);
More information about the jboss-cvs-commits
mailing list