[jboss-cvs] JBoss Messaging SVN: r5138 - in trunk: src/main/org/jboss/messaging/core/client/impl and 8 other directories.
jboss-cvs-commits at lists.jboss.org
jboss-cvs-commits at lists.jboss.org
Fri Oct 17 14:08:39 EDT 2008
Author: ataylor
Date: 2008-10-17 14:08:38 -0400 (Fri, 17 Oct 2008)
New Revision: 5138
Added:
trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionCreateQueueCopyMessage.java
Removed:
trunk/src/main/org/jboss/messaging/core/client/ClientBrowser.java
trunk/src/main/org/jboss/messaging/core/client/impl/ClientBrowserImpl.java
trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionBrowseMessage.java
trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionBrowserCloseMessage.java
trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionBrowserHasNextMessageMessage.java
trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionBrowserHasNextMessageResponseMessage.java
trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionBrowserNextMessageMessage.java
trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionBrowserResetMessage.java
trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionCreateBrowserMessage.java
trunk/src/main/org/jboss/messaging/core/server/impl/ServerBrowserImpl.java
trunk/tests/src/org/jboss/messaging/tests/timing/core/client/ScheduledMessageTest.java
Modified:
trunk/src/main/org/jboss/messaging/core/client/ClientSession.java
trunk/src/main/org/jboss/messaging/core/client/impl/ClientSessionImpl.java
trunk/src/main/org/jboss/messaging/core/client/impl/ClientSessionInternal.java
trunk/src/main/org/jboss/messaging/core/remoting/impl/RemotingConnectionImpl.java
trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/PacketImpl.java
trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionCreateConsumerMessage.java
trunk/src/main/org/jboss/messaging/core/server/ServerSession.java
trunk/src/main/org/jboss/messaging/core/server/impl/ServerConsumerImpl.java
trunk/src/main/org/jboss/messaging/core/server/impl/ServerSessionImpl.java
trunk/src/main/org/jboss/messaging/core/server/impl/ServerSessionPacketHandler.java
trunk/src/main/org/jboss/messaging/jms/client/JBossQueueBrowser.java
trunk/src/main/org/jboss/messaging/jms/client/JBossSession.java
trunk/tests/jms-tests/src/org/jboss/test/messaging/jms/BrowserTest.java
trunk/tests/jms-tests/src/org/jboss/test/messaging/jms/SecurityTest.java
trunk/tests/src/org/jboss/messaging/tests/unit/jms/client/JBossQueueBrowserTest.java
trunk/tests/src/org/jboss/messaging/tests/unit/jms/client/JBossSessionTest.java
Log:
https://jira.jboss.org/jira/browse/JBMESSAGING-1426 - second check in
Deleted: trunk/src/main/org/jboss/messaging/core/client/ClientBrowser.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/client/ClientBrowser.java 2008-10-17 16:53:35 UTC (rev 5137)
+++ trunk/src/main/org/jboss/messaging/core/client/ClientBrowser.java 2008-10-17 18:08:38 UTC (rev 5138)
@@ -1,48 +0,0 @@
-/*
- * JBoss, Home of Professional Open Source
- * Copyright 2005-2008, Red Hat Middleware LLC, and individual contributors
- * 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.core.client;
-
-import org.jboss.messaging.core.exception.MessagingException;
-
-/**
- * @author <a href="mailto:tim.fox at jboss.com">Tim Fox</a>
- * @author <a href="mailto:ovidiu at feodorov.com">Ovidiu Feodorov</a>
- * @author <a href="mailto:clebert.suconic at jboss.org">Clebert Suconic</a>
- * @author <a href="mailto:ataylor at redhat.com">Andy Taylor</a>
- */
-public interface ClientBrowser
-{
- long getID();
-
- void reset() throws MessagingException;
-
- ClientMessage nextMessage() throws MessagingException;
-
- boolean hasNextMessage() throws MessagingException;
-
- void close() throws MessagingException;
-
- boolean isClosed();
-
- void cleanUp();
-}
Modified: trunk/src/main/org/jboss/messaging/core/client/ClientSession.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/client/ClientSession.java 2008-10-17 16:53:35 UTC (rev 5137)
+++ trunk/src/main/org/jboss/messaging/core/client/ClientSession.java 2008-10-17 18:08:38 UTC (rev 5138)
@@ -22,14 +22,14 @@
package org.jboss.messaging.core.client;
-import javax.transaction.xa.XAResource;
-
import org.jboss.messaging.core.exception.MessagingException;
import org.jboss.messaging.core.remoting.FailureListener;
import org.jboss.messaging.core.remoting.impl.wireformat.SessionBindingQueryResponseMessage;
import org.jboss.messaging.core.remoting.impl.wireformat.SessionQueueQueryResponseMessage;
import org.jboss.messaging.util.SimpleString;
+import javax.transaction.xa.XAResource;
+
/*
*
* @author <a href="mailto:tim.fox at jboss.com">Tim Fox</a>
@@ -60,16 +60,15 @@
ClientConsumer createConsumer(SimpleString queueName, SimpleString filterString, boolean direct) throws MessagingException;
+ ClientConsumer createConsumer(SimpleString queueName, SimpleString filterString, boolean direct, boolean browseOnly) throws MessagingException;
+
ClientConsumer createConsumer(SimpleString queueName,
SimpleString filterString,
boolean direct,
int windowSize,
- int maxRate) throws MessagingException;
+ int maxRate,
+ boolean browseOnly) throws MessagingException;
- ClientBrowser createBrowser(SimpleString queueName, SimpleString filterString) throws MessagingException;
-
- ClientBrowser createBrowser(SimpleString queueName) throws MessagingException;
-
ClientProducer createProducer(SimpleString address) throws MessagingException;
ClientProducer createProducer(SimpleString address,
@@ -121,4 +120,6 @@
boolean removeFailureListener(FailureListener listener);
int getVersion();
+
+ void createQueueCopy(SimpleString queueName, SimpleString queueCopyName, SimpleString coreSelector, boolean durable, boolean temporary) throws MessagingException;
}
Deleted: trunk/src/main/org/jboss/messaging/core/client/impl/ClientBrowserImpl.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/client/impl/ClientBrowserImpl.java 2008-10-17 16:53:35 UTC (rev 5137)
+++ trunk/src/main/org/jboss/messaging/core/client/impl/ClientBrowserImpl.java 2008-10-17 18:08:38 UTC (rev 5138)
@@ -1,140 +0,0 @@
-/*
- * JBoss, Home of Professional Open Source Copyright 2005-2008, Red Hat Middleware LLC, and individual contributors 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.core.client.impl;
-
-import org.jboss.messaging.core.client.ClientBrowser;
-import org.jboss.messaging.core.client.ClientMessage;
-import org.jboss.messaging.core.exception.MessagingException;
-import org.jboss.messaging.core.remoting.Channel;
-import org.jboss.messaging.core.remoting.impl.wireformat.SessionBrowseMessage;
-import org.jboss.messaging.core.remoting.impl.wireformat.SessionBrowserCloseMessage;
-import org.jboss.messaging.core.remoting.impl.wireformat.SessionBrowserHasNextMessageMessage;
-import org.jboss.messaging.core.remoting.impl.wireformat.SessionBrowserHasNextMessageResponseMessage;
-import org.jboss.messaging.core.remoting.impl.wireformat.SessionBrowserNextMessageMessage;
-import org.jboss.messaging.core.remoting.impl.wireformat.SessionBrowserResetMessage;
-
-/**
- * @author <a href="mailto:tim.fox at jboss.com">Tim Fox</a>
- * @author <a href="mailto:ovidiu at feodorov.com">Ovidiu Feodorov</a>
- * @author <a href="mailto:clebert.suconic at jboss.org">Clebert Suconic</a>
- * @author <a href="mailto:ataylor at redhat.com">Andy Taylor</a>
- * @version <tt>$Revision: 3602 $</tt> $Id: ClientBrowserImpl.java 3602 2008-01-21 17:48:32Z timfox $
- */
-public class ClientBrowserImpl implements ClientBrowser
-{
- // Constants ------------------------------------------------------------------------------------
-
- // Attributes -----------------------------------------------------------------------------------
-
- private final long id;
-
- private final ClientSessionInternal session;
-
- private final Channel channel;
-
- private volatile boolean closed;
-
- // Static ---------------------------------------------------------------------------------------
-
- // Constructors ---------------------------------------------------------------------------------
-
- public ClientBrowserImpl(final ClientSessionInternal session, final long id, final Channel channel)
- {
- this.id = id;
-
- this.session = session;
-
- this.channel = channel;
- }
-
- // ClientBrowser implementation -----------------------------------------------------------------
-
- public long getID()
- {
- return id;
- }
-
- public synchronized void close() throws MessagingException
- {
- if (closed)
- {
- return;
- }
-
- try
- {
- channel.sendBlocking(new SessionBrowserCloseMessage(id));
- }
- finally
- {
- session.removeBrowser(this);
-
- closed = true;
- }
- }
-
- public synchronized void cleanUp()
- {
- session.removeBrowser(this);
-
- closed = true;
- }
-
- public boolean isClosed()
- {
- return closed;
- }
-
- public void reset() throws MessagingException
- {
- checkClosed();
-
- channel.sendBlocking(new SessionBrowserResetMessage(id));
- }
-
- public boolean hasNextMessage() throws MessagingException
- {
- checkClosed();
-
- SessionBrowserHasNextMessageResponseMessage response = (SessionBrowserHasNextMessageResponseMessage)channel.sendBlocking(new SessionBrowserHasNextMessageMessage(id));
-
- return response.hasNext();
- }
-
- public ClientMessage nextMessage() throws MessagingException
- {
- checkClosed();
-
- SessionBrowseMessage response = (SessionBrowseMessage)channel.sendBlocking(new SessionBrowserNextMessageMessage(id));
-
- return response.getClientMessage();
- }
-
- // Public ---------------------------------------------------------------------------------------
-
- // Protected ------------------------------------------------------------------------------------
-
- // Package Private ------------------------------------------------------------------------------
-
- // Private --------------------------------------------------------------------------------------
-
- private void checkClosed() throws MessagingException
- {
- if (closed)
- {
- throw new MessagingException(MessagingException.OBJECT_CLOSED, "Browser is closed");
- }
- }
-
- // Inner Classes --------------------------------------------------------------------------------
-
-}
Modified: trunk/src/main/org/jboss/messaging/core/client/impl/ClientSessionImpl.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/client/impl/ClientSessionImpl.java 2008-10-17 16:53:35 UTC (rev 5137)
+++ trunk/src/main/org/jboss/messaging/core/client/impl/ClientSessionImpl.java 2008-10-17 18:08:38 UTC (rev 5138)
@@ -21,20 +21,6 @@
*/
package org.jboss.messaging.core.client.impl;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.Executor;
-import java.util.concurrent.Executors;
-
-import javax.transaction.xa.XAException;
-import javax.transaction.xa.XAResource;
-import javax.transaction.xa.Xid;
-
-import org.jboss.messaging.core.client.ClientBrowser;
import org.jboss.messaging.core.client.ClientConsumer;
import org.jboss.messaging.core.client.ClientMessage;
import org.jboss.messaging.core.client.ClientProducer;
@@ -56,11 +42,11 @@
import org.jboss.messaging.core.remoting.impl.wireformat.SessionBindingQueryResponseMessage;
import org.jboss.messaging.core.remoting.impl.wireformat.SessionCloseMessage;
import org.jboss.messaging.core.remoting.impl.wireformat.SessionConsumerFlowCreditMessage;
-import org.jboss.messaging.core.remoting.impl.wireformat.SessionCreateBrowserMessage;
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.SessionCreateProducerMessage;
import org.jboss.messaging.core.remoting.impl.wireformat.SessionCreateProducerResponseMessage;
+import org.jboss.messaging.core.remoting.impl.wireformat.SessionCreateQueueCopyMessage;
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.SessionFailoverCompleteMessage;
@@ -89,21 +75,33 @@
import org.jboss.messaging.util.SimpleString;
import org.jboss.messaging.util.TokenBucketLimiterImpl;
+import javax.transaction.xa.XAException;
+import javax.transaction.xa.XAResource;
+import javax.transaction.xa.Xid;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+
/*
* @author <a href="mailto:tim.fox at jboss.com">Tim Fox</a>
- *
+ *
* @author <a href="mailto:ovidiu at feodorov.com">Ovidiu Feodorov</a>
- *
+ *
* @author <a href="mailto:clebert.suconic at jboss.org">Clebert Suconic</a>
- *
+ *
* @author <a href="mailto:jmesnil at redhat.com">Jeff Mesnil</a>
- *
+ *
* @author <a href="mailto:ataylor at redhat.com">Andy Taylor</a>
- *
+ *
* @version <tt>$Revision: 3603 $</tt> $Id: ClientSessionImpl.java 3603 2008-01-21 18:49:20Z timfox $
- *
+ *
* $Id: ClientSessionImpl.java 3603 2008-01-21 18:49:20Z timfox $
- *
+ *
*/
public class ClientSessionImpl implements ClientSessionInternal, FailureListener
{
@@ -131,8 +129,6 @@
private volatile RemotingConnection remotingConnection;
- private final Map<Long, ClientBrowser> browsers = new ConcurrentHashMap<Long, ClientBrowser>();
-
private final Map<Long, ClientProducerInternal> producers = new ConcurrentHashMap<Long, ClientProducerInternal>();
private final Map<Long, ClientConsumerInternal> consumers = new ConcurrentHashMap<Long, ClientConsumerInternal>();
@@ -161,9 +157,9 @@
private boolean forceNotSameRM;
private final IDGenerator idGenerator = new SimpleIDGenerator(0);
-
+
private volatile boolean failedOver;
-
+
private volatile boolean started;
// Constructors ----------------------------------------------------------------------------
@@ -217,7 +213,7 @@
this.version = version;
connectionRegistry = ConnectionRegistryImpl.instance;
-
+
remotingConnection.addFailureListener(this);
}
@@ -231,7 +227,7 @@
final boolean temp) throws MessagingException
{
checkClosed();
-
+
SessionCreateQueueMessage request = new SessionCreateQueueMessage(address, queueName, filterString, durable, temp);
channel.sendBlocking(request);
@@ -301,21 +297,35 @@
filterString,
direct,
connectionFactory.getConsumerWindowSize(),
- connectionFactory.getConsumerMaxRate());
+ connectionFactory.getConsumerMaxRate(), false);
}
public ClientConsumer createConsumer(final SimpleString queueName,
final SimpleString filterString,
final boolean direct,
+ final boolean browseOnly) throws MessagingException
+ {
+ return createConsumer(queueName,
+ filterString,
+ direct,
+ connectionFactory.getConsumerWindowSize(),
+ connectionFactory.getConsumerMaxRate(), browseOnly);
+ }
+
+ public ClientConsumer createConsumer(final SimpleString queueName,
+ final SimpleString filterString,
+ final boolean direct,
final int windowSize,
- final int maxRate) throws MessagingException
+ final int maxRate,
+ final boolean browseOnly) throws MessagingException
{
checkClosed();
-
+
SessionCreateConsumerMessage request = new SessionCreateConsumerMessage(queueName,
filterString,
windowSize,
- maxRate);
+ maxRate,
+ browseOnly);
SessionCreateConsumerResponseMessage response = (SessionCreateConsumerResponseMessage)channel.sendBlocking(request);
@@ -366,26 +376,7 @@
return consumer;
}
- public ClientBrowser createBrowser(final SimpleString queueName) throws MessagingException
- {
- return createBrowser(queueName, null);
- }
- public ClientBrowser createBrowser(final SimpleString queueName, final SimpleString filterString) throws MessagingException
- {
- checkClosed();
-
- SessionCreateBrowserMessage request = new SessionCreateBrowserMessage(queueName, filterString);
-
- channel.sendBlocking(request);
-
- ClientBrowser browser = new ClientBrowserImpl(this, idGenerator.generateID(), channel);
-
- addBrowser(browser);
-
- return browser;
- }
-
public ClientProducer createProducer(final SimpleString address) throws MessagingException
{
checkClosed();
@@ -479,19 +470,19 @@
public void rollback() throws MessagingException
{
checkClosed();
-
+
//We do a "JMS style" rollback where the session is stopped, and the buffer is cancelled back
//first before rolling back
//This ensures messages are received in the same order after rollback w.r.t. to messages in the buffer
//For core we could just do a straight rollback, it really depends if we want JMS style semantics or not...
-
+
boolean wasStarted = started;
-
+
if (wasStarted)
{
stop();
}
-
+
// We need to make sure we don't get any inflight messages
for (ClientConsumerInternal consumer : consumers.values())
{
@@ -499,7 +490,7 @@
}
channel.sendBlocking(new PacketImpl(PacketImpl.SESS_ROLLBACK));
-
+
if (wasStarted)
{
start();
@@ -564,11 +555,11 @@
public void start() throws MessagingException
{
checkClosed();
-
+
if (!started)
- {
+ {
channel.send(new PacketImpl(PacketImpl.SESS_START));
-
+
started = true;
}
}
@@ -576,11 +567,11 @@
public void stop() throws MessagingException
{
checkClosed();
-
+
if (started)
{
channel.sendBlocking(new PacketImpl(PacketImpl.SESS_STOP));
-
+
started = false;
}
}
@@ -600,6 +591,15 @@
return version;
}
+ public void createQueueCopy(SimpleString queueName, SimpleString queueCopyName, SimpleString coreSelector, boolean durable, boolean temporary) throws MessagingException
+ {
+ checkClosed();
+
+ SessionCreateQueueCopyMessage sessionCreateQueueCopyMessage = new SessionCreateQueueCopyMessage(queueCopyName, queueName, coreSelector, durable, temporary);
+
+ channel.sendBlocking(sessionCreateQueueCopyMessage);
+ }
+
// ClientSessionInternal implementation
// ------------------------------------------------------------
@@ -614,7 +614,7 @@
checkClosed();
SessionAcknowledgeMessage message = new SessionAcknowledgeMessage(consumerID, messageID, blockOnAcknowledge);
-
+
if (blockOnAcknowledge)
{
channel.sendBlocking(message);
@@ -635,11 +635,6 @@
producers.put(producer.getID(), producer);
}
- public void addBrowser(final ClientBrowser browser)
- {
- browsers.put(browser.getID(), browser);
- }
-
public void removeConsumer(final ClientConsumerInternal consumer) throws MessagingException
{
consumers.remove(consumer.getID());
@@ -655,11 +650,6 @@
}
}
- public void removeBrowser(final ClientBrowser browser)
- {
- browsers.remove(browser.getID());
- }
-
public Set<ClientProducerInternal> getProducers()
{
return new HashSet<ClientProducerInternal>(producers.values());
@@ -670,11 +660,6 @@
return new HashSet<ClientConsumerInternal>(consumers.values());
}
- public Set<ClientBrowser> getBrowsers()
- {
- return new HashSet<ClientBrowser>(browsers.values());
- }
-
public Map<SimpleString, ClientProducerInternal> getProducerCache()
{
return new HashMap<SimpleString, ClientProducerInternal>(producerCache);
@@ -699,7 +684,7 @@
producer.receiveCredits(credits);
}
}
-
+
public void close() throws MessagingException
{
if (closed)
@@ -720,19 +705,19 @@
doCleanup();
}
-
+
public synchronized void cleanUp() throws Exception
{
if (closed)
{
return;
}
-
+
cleanUpChildren();
-
- doCleanup();
+
+ doCleanup();
}
-
+
//Needs to be synchronized to prevent issues with occurring concurrently with close()
public synchronized boolean handleFailover(final RemotingConnection backupConnection)
{
@@ -773,18 +758,18 @@
catch (Throwable t)
{
log.error("Failed to handle failover", t);
-
+
return false;
}
finally
{
channel.unlock();
}
-
+
channel.send(new SessionFailoverCompleteMessage(name));
-
+
failedOver = true;
-
+
return true;
}
@@ -1052,9 +1037,9 @@
throw new XAException(XAException.XAER_RMERR);
}
}
-
+
// FailureListener implementation --------------------------------------------
-
+
public void connectionFailed(final MessagingException me)
{
if (!sessionFactory.checkFailover(me))
@@ -1116,7 +1101,7 @@
throw new MessagingException(MessagingException.OBJECT_CLOSED, "Session is closed");
}
}
-
+
private void doCleanup()
{
if (cacheProducers)
@@ -1125,7 +1110,7 @@
}
channel.close();
-
+
remotingConnection.removeFailureListener(this);
synchronized (this)
@@ -1134,10 +1119,10 @@
connectionRegistry.returnConnection(remotingConnection.getID());
}
-
+
sessionFactory.removeSession(this);
}
-
+
private void cleanUpChildren() throws Exception
{
Set<ClientConsumerInternal> consumersClone = new HashSet<ClientConsumerInternal>(consumers.values());
@@ -1153,13 +1138,6 @@
{
producer.cleanUp();
}
-
- Set<ClientBrowser> browsersClone = new HashSet<ClientBrowser>(browsers.values());
-
- for (ClientBrowser browser : browsersClone)
- {
- browser.cleanUp();
- }
}
private void closeChildren() throws MessagingException
@@ -1177,13 +1155,6 @@
{
producer.close();
}
-
- Set<ClientBrowser> browsersClone = new HashSet<ClientBrowser>(browsers.values());
-
- for (ClientBrowser browser : browsersClone)
- {
- browser.close();
- }
}
}
Modified: trunk/src/main/org/jboss/messaging/core/client/impl/ClientSessionInternal.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/client/impl/ClientSessionInternal.java 2008-10-17 16:53:35 UTC (rev 5137)
+++ trunk/src/main/org/jboss/messaging/core/client/impl/ClientSessionInternal.java 2008-10-17 18:08:38 UTC (rev 5138)
@@ -12,16 +12,15 @@
package org.jboss.messaging.core.client.impl;
-import java.util.Map;
-import java.util.Set;
-
-import org.jboss.messaging.core.client.ClientBrowser;
import org.jboss.messaging.core.client.ClientMessage;
import org.jboss.messaging.core.client.ClientSession;
import org.jboss.messaging.core.exception.MessagingException;
import org.jboss.messaging.core.remoting.RemotingConnection;
import org.jboss.messaging.util.SimpleString;
+import java.util.Map;
+import java.util.Set;
+
/**
* A ClientSessionInternal
*
@@ -37,20 +36,14 @@
void addProducer(ClientProducerInternal producer);
- void addBrowser(ClientBrowser browser);
-
void removeConsumer(ClientConsumerInternal consumer) throws MessagingException;
void removeProducer(ClientProducerInternal producer);
- void removeBrowser(ClientBrowser browser);
-
Set<ClientProducerInternal> getProducers();
Set<ClientConsumerInternal> getConsumers();
- Set<ClientBrowser> getBrowsers();
-
Map<SimpleString, ClientProducerInternal> getProducerCache();
//void cleanUp() throws Exception;
Modified: trunk/src/main/org/jboss/messaging/core/remoting/impl/RemotingConnectionImpl.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/remoting/impl/RemotingConnectionImpl.java 2008-10-17 16:53:35 UTC (rev 5137)
+++ trunk/src/main/org/jboss/messaging/core/remoting/impl/RemotingConnectionImpl.java 2008-10-17 18:08:38 UTC (rev 5138)
@@ -12,6 +12,20 @@
package org.jboss.messaging.core.remoting.impl;
+import org.jboss.messaging.core.exception.MessagingException;
+import org.jboss.messaging.core.logging.Logger;
+import org.jboss.messaging.core.remoting.Channel;
+import org.jboss.messaging.core.remoting.ChannelHandler;
+import org.jboss.messaging.core.remoting.DelayedResult;
+import org.jboss.messaging.core.remoting.FailureListener;
+import org.jboss.messaging.core.remoting.Interceptor;
+import org.jboss.messaging.core.remoting.Packet;
+import org.jboss.messaging.core.remoting.RemotingConnection;
+import org.jboss.messaging.core.remoting.impl.wireformat.CreateSessionMessage;
+import org.jboss.messaging.core.remoting.impl.wireformat.CreateSessionResponseMessage;
+import org.jboss.messaging.core.remoting.impl.wireformat.MessagingExceptionMessage;
+import org.jboss.messaging.core.remoting.impl.wireformat.NullResponseMessage;
+import org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl;
import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.CREATESESSION;
import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.CREATESESSION_RESP;
import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.EXCEPTION;
@@ -26,20 +40,15 @@
import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.SESS_ADD_DESTINATION;
import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.SESS_BINDINGQUERY;
import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.SESS_BINDINGQUERY_RESP;
-import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.SESS_BROWSER_CLOSE;
-import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.SESS_BROWSER_HASNEXTMESSAGE;
-import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.SESS_BROWSER_HASNEXTMESSAGE_RESP;
-import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.SESS_BROWSER_NEXTMESSAGE;
-import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.SESS_BROWSER_RESET;
import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.SESS_CLOSE;
import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.SESS_COMMIT;
import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.SESS_CONSUMER_CLOSE;
-import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.SESS_CREATEBROWSER;
import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.SESS_CREATECONSUMER;
import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.SESS_CREATECONSUMER_RESP;
import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.SESS_CREATEPRODUCER;
import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.SESS_CREATEPRODUCER_RESP;
import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.SESS_CREATEQUEUE;
+import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.SESS_CREATEQUEUECOPY;
import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.SESS_DELETE_QUEUE;
import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.SESS_FAILOVER_COMPLETE;
import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.SESS_FLOWTOKEN;
@@ -72,37 +81,6 @@
import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.SESS_XA_SET_TIMEOUT_RESP;
import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.SESS_XA_START;
import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.SESS_XA_SUSPEND;
-
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Queue;
-import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentLinkedQueue;
-import java.util.concurrent.CopyOnWriteArrayList;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ScheduledFuture;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.locks.Condition;
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.ReadWriteLock;
-import java.util.concurrent.locks.ReentrantLock;
-
-import org.jboss.messaging.core.exception.MessagingException;
-import org.jboss.messaging.core.logging.Logger;
-import org.jboss.messaging.core.remoting.Channel;
-import org.jboss.messaging.core.remoting.ChannelHandler;
-import org.jboss.messaging.core.remoting.DelayedResult;
-import org.jboss.messaging.core.remoting.FailureListener;
-import org.jboss.messaging.core.remoting.Interceptor;
-import org.jboss.messaging.core.remoting.Packet;
-import org.jboss.messaging.core.remoting.RemotingConnection;
-import org.jboss.messaging.core.remoting.impl.wireformat.CreateSessionMessage;
-import org.jboss.messaging.core.remoting.impl.wireformat.CreateSessionResponseMessage;
-import org.jboss.messaging.core.remoting.impl.wireformat.MessagingExceptionMessage;
-import org.jboss.messaging.core.remoting.impl.wireformat.NullResponseMessage;
-import org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl;
import org.jboss.messaging.core.remoting.impl.wireformat.PacketsConfirmedMessage;
import org.jboss.messaging.core.remoting.impl.wireformat.Ping;
import org.jboss.messaging.core.remoting.impl.wireformat.Pong;
@@ -112,20 +90,14 @@
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.SessionBrowseMessage;
-import org.jboss.messaging.core.remoting.impl.wireformat.SessionBrowserCloseMessage;
-import org.jboss.messaging.core.remoting.impl.wireformat.SessionBrowserHasNextMessageMessage;
-import org.jboss.messaging.core.remoting.impl.wireformat.SessionBrowserHasNextMessageResponseMessage;
-import org.jboss.messaging.core.remoting.impl.wireformat.SessionBrowserNextMessageMessage;
-import org.jboss.messaging.core.remoting.impl.wireformat.SessionBrowserResetMessage;
import org.jboss.messaging.core.remoting.impl.wireformat.SessionCloseMessage;
import org.jboss.messaging.core.remoting.impl.wireformat.SessionConsumerCloseMessage;
import org.jboss.messaging.core.remoting.impl.wireformat.SessionConsumerFlowCreditMessage;
-import org.jboss.messaging.core.remoting.impl.wireformat.SessionCreateBrowserMessage;
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.SessionCreateProducerMessage;
import org.jboss.messaging.core.remoting.impl.wireformat.SessionCreateProducerResponseMessage;
+import org.jboss.messaging.core.remoting.impl.wireformat.SessionCreateQueueCopyMessage;
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.SessionFailoverCompleteMessage;
@@ -156,6 +128,22 @@
import org.jboss.messaging.core.remoting.spi.MessagingBuffer;
import org.jboss.messaging.util.SimpleIDGenerator;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Queue;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantLock;
+
/**
* @author <a href="tim.fox at jboss.com">Tim Fox</a>
* @author <a href="mailto:jmesnil at redhat.com">Jeff Mesnil</a>
@@ -218,17 +206,17 @@
private volatile SimpleIDGenerator idGenerator = new SimpleIDGenerator(10);
private boolean idGeneratorSynced = false;
-
+
private final ReadWriteLock readWriteLock;
-
+
private final Object transferLock = new Object();
-
+
// Constructors
// ---------------------------------------------------------------------------------
public RemotingConnectionImpl(final Connection transportConnection,
final long blockingCallTimeout,
- final long pingPeriod,
+ final long pingPeriod,
final ScheduledExecutorService pingExecutor,
final List<Interceptor> interceptors,
final RemotingConnection replicatingConnection,
@@ -249,7 +237,7 @@
this.pingPeriod = pingPeriod;
this.pingExecutor = pingExecutor;
-
+
this.readWriteLock = readWriteLock;
// Channel zero is reserved for pinging
@@ -284,7 +272,7 @@
return transportConnection.getID();
}
- public synchronized Channel getChannel(final long channelID,
+ public synchronized Channel getChannel(final long channelID,
final int packetConfirmationBatchSize,
final boolean interruptBlockOnFailure)
{
@@ -411,12 +399,12 @@
final Packet packet = decode(buffer);
final long channelID = packet.getChannelID();
-
+
//FIXME - need to redo global ordering since this won't work with multiple connections
//Instead use lastSeq technique
-
+
final boolean useLock = readWriteLock != null;
-
+
if (useLock)
{
if (packet.isRequiresGlobalOrdering() || packet.getType() == REPLICATION_RESPONSE)
@@ -428,7 +416,7 @@
readWriteLock.readLock().lock();
}
}
-
+
try
{
// This needs to be synchronized so plays nice with transfer connection
@@ -440,7 +428,7 @@
{
channel.handlePacket(packet);
}
- }
+ }
}
finally
{
@@ -611,11 +599,6 @@
packet = new SessionCreateProducerResponseMessage();
break;
}
- case SESS_CREATEBROWSER:
- {
- packet = new SessionCreateBrowserMessage();
- break;
- }
case SESS_ACKNOWLEDGE:
{
packet = new SessionAcknowledgeMessage();
@@ -646,6 +629,11 @@
packet = new SessionCreateQueueMessage();
break;
}
+ case SESS_CREATEQUEUECOPY:
+ {
+ packet = new SessionCreateQueueCopyMessage();
+ break;
+ }
case SESS_DELETE_QUEUE:
{
packet = new SessionDeleteQueueMessage();
@@ -671,31 +659,6 @@
packet = new SessionBindingQueryResponseMessage();
break;
}
- case PacketImpl.SESS_BROWSER_MESSAGE:
- {
- packet = new SessionBrowseMessage();
- break;
- }
- case SESS_BROWSER_RESET:
- {
- packet = new SessionBrowserResetMessage();
- break;
- }
- case SESS_BROWSER_HASNEXTMESSAGE:
- {
- packet = new SessionBrowserHasNextMessageMessage();
- break;
- }
- case SESS_BROWSER_HASNEXTMESSAGE_RESP:
- {
- packet = new SessionBrowserHasNextMessageResponseMessage();
- break;
- }
- case SESS_BROWSER_NEXTMESSAGE:
- {
- packet = new SessionBrowserNextMessageMessage();
- break;
- }
case SESS_XA_START:
{
packet = new SessionXAStartMessage();
@@ -816,11 +779,6 @@
packet = new SessionProducerCloseMessage();
break;
}
- case SESS_BROWSER_CLOSE:
- {
- packet = new SessionBrowserCloseMessage();
- break;
- }
case SESS_SCHEDULED_SEND:
{
packet = new SessionScheduledSendMessage();
@@ -1004,9 +962,9 @@
{
if (closed)
{
- throw new MessagingException(MessagingException.NOT_CONNECTED, "Connection is destroyed");
+ throw new MessagingException(MessagingException.NOT_CONNECTED, "Connection is destroyed");
}
-
+
packet.setChannelID(id);
lock.lock();
@@ -1092,7 +1050,7 @@
* 2) A command must be processed on the live before the next command is processed on the live
* We replicate as follows:
* As a command arrives on the live, we replicate to the backup where it will get processed, we then immediately process
- * it on the live, but we stop short of sending the result back from the live until the result of completion has
+ * it on the live, but we stop short of sending the result back from the live until the result of completion has
* arrived back from the backup. This is what the DelayedResult is used for.
*/
public DelayedResult replicatePacket(final Packet packet)
@@ -1103,11 +1061,11 @@
synchronized (this)
{
DelayedResult result = new DelayedResult();
-
+
responseActions.add(result);
replicatingChannel.send(packet);
-
+
return result;
}
}
@@ -1130,12 +1088,12 @@
connection.doWrite(packet);
}
}
-
+
// This will never get called concurrently by more than one thread
public void replicateResponseReceived()
{
DelayedResult result = responseActions.poll();
-
+
if (result == null)
{
throw new IllegalStateException("Cannot find response action");
@@ -1159,7 +1117,7 @@
if (!connection.destroyed && connection.channels.remove(id) == null)
{
throw new IllegalArgumentException("Cannot find channel with id " + id + " to close");
- }
+ }
if (replicatingChannel != null)
{
@@ -1242,7 +1200,7 @@
}
// we need to do a thorough investigation of how packets confirmed get
- //
+ //
// a) replicated from client through live to backup without dealing with on live
// b) got redirected back to client from server
@@ -1309,10 +1267,10 @@
}
}
else if (handler != null)
- {
+ {
checkConfirmation(packet);
-
- handler.handlePacket(packet);
+
+ handler.handlePacket(packet);
}
else
{
Modified: trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/PacketImpl.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/PacketImpl.java 2008-10-17 16:53:35 UTC (rev 5137)
+++ trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/PacketImpl.java 2008-10-17 18:08:38 UTC (rev 5138)
@@ -47,10 +47,10 @@
public static final byte NULL_RESPONSE = 21;
public static final byte PACKETS_CONFIRMED = 22;
-
+
public static final byte REPLICATION_RESPONSE = 23;
-
+
// Server
public static final byte CREATESESSION = 30;
@@ -59,7 +59,7 @@
public static final byte REATTACH_SESSION = 32;
public static final byte REATTACH_SESSION_RESP = 33;
-
+
// Session
public static final byte SESS_CREATECONSUMER = 40;
@@ -69,10 +69,8 @@
public static final byte SESS_CREATEPRODUCER_RESP = 43;
- public static final byte SESS_CREATEBROWSER = 44;
+ public static final byte SESS_CREATEQUEUECOPY = 44;
- public static final byte SESS_CREATEBROWSER_RESP = 45;
-
public static final byte SESS_ACKNOWLEDGE = 46;
public static final byte SESS_COMMIT = 47;
@@ -95,16 +93,6 @@
public static final byte SESS_BINDINGQUERY_RESP = 56;
- public static final byte SESS_BROWSER_MESSAGE = 57;
-
- public static final byte SESS_BROWSER_RESET = 58;
-
- public static final byte SESS_BROWSER_HASNEXTMESSAGE = 59;
-
- public static final byte SESS_BROWSER_HASNEXTMESSAGE_RESP = 60;
-
- public static final byte SESS_BROWSER_NEXTMESSAGE = 61;
-
public static final byte SESS_XA_START = 62;
public static final byte SESS_XA_END = 63;
@@ -153,16 +141,14 @@
public static final byte SESS_PRODUCER_CLOSE = 85;
- public static final byte SESS_BROWSER_CLOSE = 86;
-
public static final byte SESS_RECEIVE_MSG = 87;
public static final byte SESS_MANAGEMENT_SEND = 88;
public static final byte SESS_SCHEDULED_SEND = 89;
-
+
public static final byte SESS_FAILOVER_COMPLETE = 90;
-
+
public static final byte SESS_REPLICATE_DELIVERY = 91;
// Static --------------------------------------------------------
Deleted: trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionBrowseMessage.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionBrowseMessage.java 2008-10-17 16:53:35 UTC (rev 5137)
+++ trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionBrowseMessage.java 2008-10-17 18:08:38 UTC (rev 5138)
@@ -1,107 +0,0 @@
-/*
- * JBoss, Home of Professional Open Source
- * Copyright 2005-2008, Red Hat Middleware LLC, and individual contributors
- * 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.core.remoting.impl.wireformat;
-
-import org.jboss.messaging.core.client.ClientMessage;
-import org.jboss.messaging.core.client.impl.ClientMessageImpl;
-import org.jboss.messaging.core.logging.Logger;
-import org.jboss.messaging.core.remoting.spi.MessagingBuffer;
-import org.jboss.messaging.core.server.ServerMessage;
-
-/**
- * @author <a href="mailto:tim.fox at jboss.com">Tim Fox</a>
- *
- * @version <tt>$Revision$</tt>
- */
-public class SessionBrowseMessage extends PacketImpl
-{
- // Constants -----------------------------------------------------
-
- private static final Logger log = Logger.getLogger(SessionReceiveMessage.class);
-
- // Attributes ----------------------------------------------------
-
- private ClientMessage clientMessage;
-
- private ServerMessage serverMessage;
-
-
- // Static --------------------------------------------------------
-
- // Constructors --------------------------------------------------
-
- public SessionBrowseMessage(final ServerMessage message)
- {
- super(SESS_BROWSER_MESSAGE);
-
- this.serverMessage = message;
-
- this.clientMessage = null;
- }
-
- public SessionBrowseMessage()
- {
- super(SESS_BROWSER_MESSAGE);
- }
-
- // Public --------------------------------------------------------
-
- public boolean isResponse()
- {
- return true;
- }
-
- public ClientMessage getClientMessage()
- {
- return clientMessage;
- }
-
- public ServerMessage getServerMessage()
- {
- return serverMessage;
- }
-
- public void encodeBody(final MessagingBuffer buffer)
- {
- serverMessage.encode(buffer);
- }
-
- public void decodeBody(final MessagingBuffer buffer)
- {
- //TODO can be optimised
-
- clientMessage = new ClientMessageImpl();
-
- clientMessage.decode(buffer);
-
- clientMessage.getBody().flip();
- }
-
- // Package protected ---------------------------------------------
-
- // Protected -----------------------------------------------------
-
- // Private -------------------------------------------------------
-
- // Inner classes -------------------------------------------------
-}
Deleted: trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionBrowserCloseMessage.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionBrowserCloseMessage.java 2008-10-17 16:53:35 UTC (rev 5137)
+++ trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionBrowserCloseMessage.java 2008-10-17 18:08:38 UTC (rev 5138)
@@ -1,99 +0,0 @@
-/*
- * JBoss, Home of Professional Open Source
- * Copyright 2005-2008, Red Hat Middleware LLC, and individual contributors
- * 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.core.remoting.impl.wireformat;
-
-import org.jboss.messaging.core.remoting.spi.MessagingBuffer;
-
-/**
- * @author <a href="mailto:tim.fox at jboss.com">Tim Fox</a>
- *
- * @version <tt>$Revision$</tt>
- */
-public class SessionBrowserCloseMessage extends PacketImpl
-{
- // Constants -----------------------------------------------------
-
- // Attributes ----------------------------------------------------
-
- private long browserID;
-
- // Static --------------------------------------------------------
-
- // Constructors --------------------------------------------------
-
- public SessionBrowserCloseMessage(final long browserID)
- {
- super(SESS_BROWSER_CLOSE);
-
- this.browserID = browserID;
- }
-
- public SessionBrowserCloseMessage()
- {
- super(SESS_BROWSER_CLOSE);
- }
-
- // Public --------------------------------------------------------
-
- public long getBrowserID()
- {
- return browserID;
- }
-
- public void encodeBody(final MessagingBuffer buffer)
- {
- buffer.putLong(browserID);
- }
-
- public void decodeBody(final MessagingBuffer buffer)
- {
- browserID = buffer.getLong();
- }
-
-
- @Override
- public String toString()
- {
- return getParentString() + ", browserID=" + browserID + "]";
- }
-
- public boolean equals(Object other)
- {
- if (other instanceof SessionBrowserCloseMessage == false)
- {
- return false;
- }
-
- SessionBrowserCloseMessage r = (SessionBrowserCloseMessage)other;
-
- return super.equals(other) && this.browserID == r.browserID;
- }
- // Package protected ---------------------------------------------
-
- // Protected -----------------------------------------------------
-
- // Private -------------------------------------------------------
-
- // Inner classes -------------------------------------------------
-}
-
Deleted: trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionBrowserHasNextMessageMessage.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionBrowserHasNextMessageMessage.java 2008-10-17 16:53:35 UTC (rev 5137)
+++ trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionBrowserHasNextMessageMessage.java 2008-10-17 18:08:38 UTC (rev 5138)
@@ -1,88 +0,0 @@
-/*
- * JBoss, Home of Professional Open Source
- * Copyright 2005-2008, Red Hat Middleware LLC, and individual contributors
- * 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.core.remoting.impl.wireformat;
-
-import org.jboss.messaging.core.remoting.spi.MessagingBuffer;
-
-/**
- * @author <a href="mailto:tim.fox at jboss.com">Tim Fox</a>
- *
- * @version <tt>$Revision$</tt>
- *
- */
-public class SessionBrowserHasNextMessageMessage extends PacketImpl
-{
- // Constants -----------------------------------------------------
-
- // Attributes ----------------------------------------------------
-
- private long browserID;
-
- // Static --------------------------------------------------------
-
- // Constructors --------------------------------------------------
-
- public SessionBrowserHasNextMessageMessage(final long browserID)
- {
- super(SESS_BROWSER_HASNEXTMESSAGE);
-
- this.browserID = browserID;
- }
-
- public SessionBrowserHasNextMessageMessage()
- {
- super(SESS_BROWSER_HASNEXTMESSAGE);
- }
-
- // Public --------------------------------------------------------
-
- public long getBrowserID()
- {
- return browserID;
- }
-
- public void encodeBody(final MessagingBuffer buffer)
- {
- buffer.putLong(browserID);
- }
-
- public void decodeBody(final MessagingBuffer buffer)
- {
- browserID = buffer.getLong();
- }
-
- @Override
- public String toString()
- {
- return getParentString() + ", browserID=" + browserID + "]";
- }
-
- // Package protected ---------------------------------------------
-
- // Protected -----------------------------------------------------
-
- // Private -------------------------------------------------------
-
- // Inner classes -------------------------------------------------
-}
-
Deleted: trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionBrowserHasNextMessageResponseMessage.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionBrowserHasNextMessageResponseMessage.java 2008-10-17 16:53:35 UTC (rev 5137)
+++ trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionBrowserHasNextMessageResponseMessage.java 2008-10-17 18:08:38 UTC (rev 5138)
@@ -1,93 +0,0 @@
-/*
- * JBoss, Home of Professional Open Source
- * Copyright 2005-2008, Red Hat Middleware LLC, and individual contributors
- * 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.core.remoting.impl.wireformat;
-
-import org.jboss.messaging.core.remoting.spi.MessagingBuffer;
-
-/**
- * @author <a href="mailto:tim.fox at jboss.com">Tim Fox</a>
- * @author <a href="mailto:jmesnil at redhat.com">Jeff Mesnil</a>
- *
- * @version <tt>$Revision$</tt>
- *
- */
-public class SessionBrowserHasNextMessageResponseMessage extends PacketImpl
-{
- // Constants -----------------------------------------------------
-
- // Attributes ----------------------------------------------------
-
- private boolean hasNext;
-
- // Static --------------------------------------------------------
-
- // Constructors --------------------------------------------------
-
- public SessionBrowserHasNextMessageResponseMessage(final boolean hasNext)
- {
- super(SESS_BROWSER_HASNEXTMESSAGE_RESP);
-
- this.hasNext = hasNext;
- }
-
- public SessionBrowserHasNextMessageResponseMessage()
- {
- super(SESS_BROWSER_HASNEXTMESSAGE_RESP);
- }
-
- // Public --------------------------------------------------------
-
- public boolean isResponse()
- {
- return true;
- }
-
- public boolean hasNext()
- {
- return hasNext;
- }
-
- public void encodeBody(final MessagingBuffer buffer)
- {
- buffer.putBoolean(hasNext);
- }
-
- public void decodeBody(final MessagingBuffer buffer)
- {
- hasNext = buffer.getBoolean();
- }
-
- @Override
- public String toString()
- {
- return getParentString() + ", hasNext=" + hasNext + "]";
- }
-
- // Package protected ---------------------------------------------
-
- // Protected -----------------------------------------------------
-
- // Private -------------------------------------------------------
-
- // Inner classes -------------------------------------------------
-}
Deleted: trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionBrowserNextMessageMessage.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionBrowserNextMessageMessage.java 2008-10-17 16:53:35 UTC (rev 5137)
+++ trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionBrowserNextMessageMessage.java 2008-10-17 18:08:38 UTC (rev 5138)
@@ -1,88 +0,0 @@
-/*
- * JBoss, Home of Professional Open Source
- * Copyright 2005-2008, Red Hat Middleware LLC, and individual contributors
- * 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.core.remoting.impl.wireformat;
-
-import org.jboss.messaging.core.remoting.spi.MessagingBuffer;
-
-/**
- * @author <a href="mailto:tim.fox at jboss.com">Tim Fox</a>
- *
- * @version <tt>$Revision$</tt>
- *
- */
-public class SessionBrowserNextMessageMessage extends PacketImpl
-{
- // Constants -----------------------------------------------------
-
- // Attributes ----------------------------------------------------
-
- private long browserID;
-
- // Static --------------------------------------------------------
-
- // Constructors --------------------------------------------------
-
- public SessionBrowserNextMessageMessage(final long browserID)
- {
- super(SESS_BROWSER_NEXTMESSAGE);
-
- this.browserID = browserID;
- }
-
- public SessionBrowserNextMessageMessage()
- {
- super(SESS_BROWSER_NEXTMESSAGE);
- }
-
- // Public --------------------------------------------------------
-
- public long getBrowserID()
- {
- return browserID;
- }
-
- public void encodeBody(final MessagingBuffer buffer)
- {
- buffer.putLong(browserID);
- }
-
- public void decodeBody(final MessagingBuffer buffer)
- {
- browserID = buffer.getLong();
- }
-
- @Override
- public String toString()
- {
- return getParentString() + ", browserID=" + browserID + "]";
- }
-
- // Package protected ---------------------------------------------
-
- // Protected -----------------------------------------------------
-
- // Private -------------------------------------------------------
-
- // Inner classes -------------------------------------------------
-}
-
Deleted: trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionBrowserResetMessage.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionBrowserResetMessage.java 2008-10-17 16:53:35 UTC (rev 5137)
+++ trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionBrowserResetMessage.java 2008-10-17 18:08:38 UTC (rev 5138)
@@ -1,89 +0,0 @@
-/*
- * JBoss, Home of Professional Open Source
- * Copyright 2005-2008, Red Hat Middleware LLC, and individual contributors
- * 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.core.remoting.impl.wireformat;
-
-import org.jboss.messaging.core.remoting.spi.MessagingBuffer;
-
-/**
- * @author <a href="mailto:tim.fox at jboss.com">Tim Fox</a>
- * @author <a href="mailto:jmesnil at redhat.com">Jeff Mesnil</a>
- *
- * @version <tt>$Revision$</tt>
- *
- */
-public class SessionBrowserResetMessage extends PacketImpl
-{
- // Constants -----------------------------------------------------
-
- // Attributes ----------------------------------------------------
-
- private long browserID;
-
- // Static --------------------------------------------------------
-
- // Constructors --------------------------------------------------
-
- public SessionBrowserResetMessage(final long browserID)
- {
- super(SESS_BROWSER_RESET);
-
- this.browserID = browserID;
- }
-
- public SessionBrowserResetMessage()
- {
- super(SESS_BROWSER_RESET);
- }
-
- // Public --------------------------------------------------------
-
- public long getBrowserID()
- {
- return browserID;
- }
-
- public void encodeBody(final MessagingBuffer buffer)
- {
- buffer.putLong(browserID);
- }
-
- public void decodeBody(final MessagingBuffer buffer)
- {
- browserID = buffer.getLong();
- }
-
- @Override
- public String toString()
- {
- return getParentString() + ", browserID=" + browserID + "]";
- }
-
- // Package protected ---------------------------------------------
-
- // Protected -----------------------------------------------------
-
- // Private -------------------------------------------------------
-
- // Inner classes -------------------------------------------------
-}
-
Deleted: trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionCreateBrowserMessage.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionCreateBrowserMessage.java 2008-10-17 16:53:35 UTC (rev 5137)
+++ trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionCreateBrowserMessage.java 2008-10-17 18:08:38 UTC (rev 5138)
@@ -1,113 +0,0 @@
-/*
- * JBoss, Home of Professional Open Source
- * Copyright 2005-2008, Red Hat Middleware LLC, and individual contributors
- * 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.core.remoting.impl.wireformat;
-
-import org.jboss.messaging.core.remoting.spi.MessagingBuffer;
-import org.jboss.messaging.util.SimpleString;
-
-/**
- * @author <a href="mailto:tim.fox at jboss.com">Tim Fox</a>
- * @author <a href="mailto:jmesnil at redhat.com">Jeff Mesnil</a>
- *
- * @version <tt>$Revision$</tt>
- *
- */
-public class SessionCreateBrowserMessage extends PacketImpl
-{
- // Constants -----------------------------------------------------
-
- // Attributes ----------------------------------------------------
-
- private SimpleString queueName;
-
- private SimpleString filterString;
-
- // Static --------------------------------------------------------
-
- // Constructors --------------------------------------------------
-
- public SessionCreateBrowserMessage(final SimpleString queueName, final SimpleString filterString)
- {
- super(SESS_CREATEBROWSER);
-
- this.queueName = queueName;
- this.filterString = filterString;
- }
-
- public SessionCreateBrowserMessage()
- {
- super(SESS_CREATEBROWSER);
- }
-
- // Public --------------------------------------------------------
-
- public SimpleString getQueueName()
- {
- return queueName;
- }
-
- public SimpleString getFilterString()
- {
- return filterString;
- }
-
- public void encodeBody(final MessagingBuffer buffer)
- {
- buffer.putSimpleString(queueName);
- buffer.putNullableSimpleString(filterString);
- }
-
- public void decodeBody(final MessagingBuffer buffer)
- {
- queueName = buffer.getSimpleString();
- filterString = buffer.getNullableSimpleString();
- }
-
- @Override
- public String toString()
- {
- return getParentString() + ", queueName=" + queueName + ", filterString="
- + filterString + "]";
- }
-
- public boolean equals(Object other)
- {
- if (other instanceof SessionCreateBrowserMessage == false)
- {
- return false;
- }
-
- SessionCreateBrowserMessage r = (SessionCreateBrowserMessage)other;
-
- return super.equals(other) && this.queueName.equals(r.queueName) &&
- this.filterString == null ? r.filterString == null : this.filterString.equals(r.filterString);
- }
-
- // Package protected ---------------------------------------------
-
- // Protected -----------------------------------------------------
-
- // Private -------------------------------------------------------
-
- // Inner classes -------------------------------------------------
-}
Modified: trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionCreateConsumerMessage.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionCreateConsumerMessage.java 2008-10-17 16:53:35 UTC (rev 5137)
+++ trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionCreateConsumerMessage.java 2008-10-17 18:08:38 UTC (rev 5138)
@@ -38,19 +38,22 @@
// Attributes ----------------------------------------------------
private SimpleString queueName;
-
+
private SimpleString filterString;
-
+
private int windowSize;
-
+
private int maxRate;
-
+
+ private boolean browseOnly;
+
+
// Static --------------------------------------------------------
// Constructors --------------------------------------------------
- public SessionCreateConsumerMessage(final SimpleString queueName, final SimpleString filterString,
- final int windowSize, final int maxRate)
+ public SessionCreateConsumerMessage(final SimpleString queueName, final SimpleString filterString,
+ final int windowSize, final int maxRate, final boolean browseOnly)
{
super(SESS_CREATECONSUMER);
@@ -58,12 +61,13 @@
this.filterString = filterString;
this.windowSize = windowSize;
this.maxRate = maxRate;
+ this.browseOnly = browseOnly;
}
-
+
public SessionCreateConsumerMessage()
{
- super(SESS_CREATECONSUMER);
- }
+ super(SESS_CREATECONSUMER);
+ }
// Public --------------------------------------------------------
@@ -93,26 +97,33 @@
{
return windowSize;
}
-
+
public int getMaxRate()
{
return maxRate;
}
-
+
+ public boolean isBrowseOnly()
+ {
+ return browseOnly;
+ }
+
public void encodeBody(final MessagingBuffer buffer)
{
buffer.putSimpleString(queueName);
buffer.putNullableSimpleString(filterString);
buffer.putInt(windowSize);
buffer.putInt(maxRate);
+ buffer.putBoolean(browseOnly);
}
-
+
public void decodeBody(final MessagingBuffer buffer)
{
queueName = buffer.getSimpleString();
filterString = buffer.getNullableSimpleString();
windowSize = buffer.getInt();
maxRate = buffer.getInt();
+ browseOnly = buffer.getBoolean();
}
public boolean equals(Object other)
Copied: trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionCreateQueueCopyMessage.java (from rev 5128, trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionCreateQueueMessage.java)
===================================================================
--- trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionCreateQueueCopyMessage.java (rev 0)
+++ trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionCreateQueueCopyMessage.java 2008-10-17 18:08:38 UTC (rev 5138)
@@ -0,0 +1,152 @@
+/*
+ * JBoss, Home of Professional Open Source
+ * Copyright 2005-2008, Red Hat Middleware LLC, and individual contributors
+ * 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.core.remoting.impl.wireformat;
+
+import org.jboss.messaging.core.remoting.spi.MessagingBuffer;
+import org.jboss.messaging.util.SimpleString;
+
+/**
+ * @author <a href="mailto:tim.fox at jboss.com">Tim Fox</a>
+
+ * @version <tt>$Revision$</tt>
+ */
+public class SessionCreateQueueCopyMessage extends PacketImpl
+{
+ // Constants -----------------------------------------------------
+
+ // Attributes ----------------------------------------------------
+
+ private SimpleString queueCopyName;
+ private SimpleString queueName;
+ private SimpleString filterString;
+ private boolean durable;
+ private boolean temporary;
+
+ // Static --------------------------------------------------------
+
+ // Constructors --------------------------------------------------
+
+ public SessionCreateQueueCopyMessage(final SimpleString queueCopyName, final SimpleString queueName,
+ final SimpleString filterString, final boolean durable, final boolean temporary)
+ {
+ super(SESS_CREATEQUEUECOPY);
+
+ this.queueCopyName = queueCopyName;
+ this.queueName = queueName;
+ this.filterString = filterString;
+ this.durable = durable;
+ this.temporary = temporary;
+ }
+
+ public SessionCreateQueueCopyMessage()
+ {
+ super(SESS_CREATEQUEUECOPY);
+ }
+
+ // Public --------------------------------------------------------
+
+ @Override
+ public String toString()
+ {
+ StringBuffer buff = new StringBuffer(getParentString());
+ buff.append(", queueCopyName=" + queueCopyName);
+ buff.append(", queueName=" + queueName);
+ buff.append(", filterString=" + filterString);
+ buff.append(", durable=" + durable);
+ buff.append(", temporary=" + temporary);
+ buff.append("]");
+ return buff.toString();
+ }
+
+ public SimpleString getQueueCopyName()
+ {
+ return queueCopyName;
+ }
+
+ public SimpleString getQueueName()
+ {
+ return queueName;
+ }
+
+ public SimpleString getFilterString()
+ {
+ return filterString;
+ }
+
+ public boolean isDurable()
+ {
+ return durable;
+ }
+
+ public boolean isTemporary()
+ {
+ return temporary;
+ }
+
+ public boolean isRequiresGlobalOrdering()
+ {
+ return true;
+ }
+
+ public void encodeBody(final MessagingBuffer buffer)
+ {
+ buffer.putSimpleString(queueCopyName);
+ buffer.putSimpleString(queueName);
+ buffer.putNullableSimpleString(filterString);
+ buffer.putBoolean(durable);
+ buffer.putBoolean(temporary);
+ }
+
+ public void decodeBody(final MessagingBuffer buffer)
+ {
+ queueCopyName = buffer.getSimpleString();
+ queueName = buffer.getSimpleString();
+ filterString = buffer.getNullableSimpleString();
+ durable = buffer.getBoolean();
+ temporary = buffer.getBoolean();
+ }
+
+ public boolean equals(Object other)
+ {
+ if (other instanceof SessionCreateQueueCopyMessage == false)
+ {
+ return false;
+ }
+
+ SessionCreateQueueCopyMessage r = (SessionCreateQueueCopyMessage)other;
+
+ return super.equals(other) && r.queueCopyName.equals(this.queueCopyName) &&
+ 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 ---------------------------------------------
+
+ // Protected -----------------------------------------------------
+
+ // Private -------------------------------------------------------
+
+ // Inner classes -------------------------------------------------
+}
\ No newline at end of file
Property changes on: trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionCreateQueueCopyMessage.java
___________________________________________________________________
Name: svn:mergeinfo
+
Modified: trunk/src/main/org/jboss/messaging/core/server/ServerSession.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/server/ServerSession.java 2008-10-17 16:53:35 UTC (rev 5137)
+++ trunk/src/main/org/jboss/messaging/core/server/ServerSession.java 2008-10-17 18:08:38 UTC (rev 5138)
@@ -22,10 +22,6 @@
package org.jboss.messaging.core.server;
-import java.util.List;
-
-import javax.transaction.xa.Xid;
-
import org.jboss.messaging.core.remoting.RemotingConnection;
import org.jboss.messaging.core.remoting.impl.wireformat.SessionBindingQueryResponseMessage;
import org.jboss.messaging.core.remoting.impl.wireformat.SessionCreateConsumerResponseMessage;
@@ -33,9 +29,11 @@
import org.jboss.messaging.core.remoting.impl.wireformat.SessionQueueQueryResponseMessage;
import org.jboss.messaging.core.remoting.impl.wireformat.SessionSendManagementMessage;
import org.jboss.messaging.core.remoting.impl.wireformat.SessionXAResponseMessage;
-import org.jboss.messaging.core.server.impl.ServerBrowserImpl;
import org.jboss.messaging.util.SimpleString;
+import javax.transaction.xa.Xid;
+import java.util.List;
+
/**
*
* A ServerSession
@@ -52,8 +50,6 @@
String getPassword();
- void removeBrowser(ServerBrowserImpl browser) throws Exception;
-
void removeConsumer(ServerConsumer consumer) throws Exception;
void removeProducer(ServerProducer producer) throws Exception;
@@ -108,12 +104,19 @@
boolean durable,
boolean temporary) throws Exception;
+ void createQueueCopy(final SimpleString queue,
+ final SimpleString queueCopy,
+ final SimpleString filterString,
+ final boolean durable,
+ final boolean temporary) throws Exception;
+
void deleteQueue(SimpleString queueName) throws Exception;
SessionCreateConsumerResponseMessage createConsumer(SimpleString queueName,
SimpleString filterString,
int windowSize,
- int maxRate) throws Exception;
+ int maxRate,
+ boolean browseOnly) throws Exception;
SessionCreateProducerResponseMessage createProducer(SimpleString address,
int windowSize,
@@ -124,31 +127,21 @@
SessionBindingQueryResponseMessage executeBindingQuery(SimpleString address) throws Exception;
- void createBrowser(SimpleString queueName, SimpleString filterString) throws Exception;
-
void closeConsumer(long consumerID) throws Exception;
void closeProducer(long producerID) throws Exception;
- void closeBrowser(long browserID) throws Exception;
-
void receiveConsumerCredits(long consumerID, int credits) throws Exception;
void sendProducerMessage(long producerID, ServerMessage message) throws Exception;
void sendScheduledProducerMessage(long producerID, ServerMessage serverMessage, long scheduledDeliveryTime) throws Exception;
- boolean browserHasNextMessage(long browserID) throws Exception;
-
- ServerMessage browserNextMessage(long browserID) throws Exception;
-
- void browserReset(long browserID) throws Exception;
-
int transferConnection(RemotingConnection newConnection, int lastReceivedCommandID);
void handleManagementMessage(SessionSendManagementMessage message) throws Exception;
void failedOver() throws Exception;
- void handleReplicatedDelivery(long consumerID, long messageID) throws Exception;
+ void handleReplicatedDelivery(long consumerID, long messageID) throws Exception;
}
Deleted: trunk/src/main/org/jboss/messaging/core/server/impl/ServerBrowserImpl.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/server/impl/ServerBrowserImpl.java 2008-10-17 16:53:35 UTC (rev 5137)
+++ trunk/src/main/org/jboss/messaging/core/server/impl/ServerBrowserImpl.java 2008-10-17 18:08:38 UTC (rev 5138)
@@ -1,191 +0,0 @@
-/*
- * JBoss, Home of Professional Open Source
- * Copyright 2005-2008, Red Hat Middleware LLC, and individual contributors
- * 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.core.server.impl;
-
-import java.util.ArrayList;
-import java.util.Iterator;
-import java.util.List;
-
-import org.jboss.messaging.core.exception.MessagingException;
-import org.jboss.messaging.core.filter.Filter;
-import org.jboss.messaging.core.filter.impl.FilterImpl;
-import org.jboss.messaging.core.logging.Logger;
-import org.jboss.messaging.core.message.Message;
-import org.jboss.messaging.core.server.MessageReference;
-import org.jboss.messaging.core.server.Queue;
-import org.jboss.messaging.core.server.ServerMessage;
-import org.jboss.messaging.core.server.ServerSession;
-import org.jboss.messaging.util.SimpleString;
-
-/**
- * Concrete implementation of BrowserEndpoint.
- *
- * @author <a href="mailto:tim.fox at jboss.com">Tim Fox</a>
- * @author <a href="mailto:ovidiu at feodorov.com">Ovidiu Feodorov</a>
- * @version <tt>$Revision: 3778 $</tt>
- *
- * $Id: ServerBrowserImpl.java 3778 2008-02-24 12:15:29Z timfox $
- */
-public class ServerBrowserImpl
-{
- // Constants ------------------------------------------------------------------------------------
-
- private static final Logger log = Logger.getLogger(ServerBrowserImpl.class);
-
- // Static ---------------------------------------------------------------------------------------
-
- // Attributes -----------------------------------------------------------------------------------
-
- private final long id;
- private final ServerSession session;
- private final Queue destination;
- private final Filter filter;
- private Iterator<ServerMessage> iterator;
-
- // Constructors ---------------------------------------------------------------------------------
-
- public ServerBrowserImpl(final long id, final ServerSession session,
- final Queue destination, final String messageFilter) throws MessagingException
- {
- this.id = id;
-
- this.session = session;
-
- this.destination = destination;
-
- if (messageFilter != null)
- {
- filter = new FilterImpl(new SimpleString(messageFilter));
- }
- else
- {
- filter = null;
- }
- }
-
- // BrowserEndpoint implementation ---------------------------------------------------------------
-
- public long getID()
- {
- return id;
- }
-
- public void reset() throws Exception
- {
- iterator = createIterator();
- }
-
- public boolean hasNextMessage() throws Exception
- {
- if (iterator == null)
- {
- iterator = createIterator();
- }
-
- boolean has = iterator.hasNext();
-
- return has;
- }
-
- public ServerMessage nextMessage() throws Exception
- {
- if (iterator == null)
- {
- iterator = createIterator();
- }
-
- ServerMessage r = iterator.next();
-
- return r;
- }
-
- public Message[] nextMessageBlock(int maxMessages) throws Exception
- {
- if (maxMessages < 2)
- {
- throw new IllegalArgumentException("maxMessages must be >=2 otherwise use nextMessage");
- }
-
- if (iterator == null)
- {
- iterator = createIterator();
- }
-
- List<ServerMessage> messages = new ArrayList<ServerMessage>(maxMessages);
- int i = 0;
- while (i < maxMessages)
- {
- if (iterator.hasNext())
- {
- ServerMessage m = iterator.next();
- messages.add(m);
- i++;
- }
- else
- {
- break;
- }
- }
- return messages.toArray(new Message[messages.size()]);
- }
-
- public void close() throws Exception
- {
- iterator = null;
-
- session.removeBrowser(this);
-
- log.trace(this + " closed");
- }
-
- // Public ---------------------------------------------------------------------------------------
-
- @Override
- public String toString()
- {
- return "BrowserEndpoint[" + id + "]";
- }
-
- // Package protected ----------------------------------------------------------------------------
-
- // Protected ------------------------------------------------------------------------------------
-
- // Private --------------------------------------------------------------------------------------
-
- private Iterator<ServerMessage> createIterator()
- {
- List<MessageReference> refs = destination.list(filter);
-
- List<ServerMessage> msgs = new ArrayList<ServerMessage>();
-
- for (MessageReference ref: refs)
- {
- msgs.add(ref.getMessage());
- }
-
- return msgs.iterator();
- }
-
- // Inner classes --------------------------------------------------------------------------------
-
-}
Modified: trunk/src/main/org/jboss/messaging/core/server/impl/ServerConsumerImpl.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/server/impl/ServerConsumerImpl.java 2008-10-17 16:53:35 UTC (rev 5137)
+++ trunk/src/main/org/jboss/messaging/core/server/impl/ServerConsumerImpl.java 2008-10-17 18:08:38 UTC (rev 5138)
@@ -22,11 +22,6 @@
package org.jboss.messaging.core.server.impl;
-import java.util.Iterator;
-import java.util.LinkedList;
-import java.util.concurrent.ConcurrentLinkedQueue;
-import java.util.concurrent.atomic.AtomicInteger;
-
import org.jboss.messaging.core.filter.Filter;
import org.jboss.messaging.core.logging.Logger;
import org.jboss.messaging.core.persistence.StorageManager;
@@ -44,12 +39,17 @@
import org.jboss.messaging.core.settings.HierarchicalRepository;
import org.jboss.messaging.core.settings.impl.QueueSettings;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.atomic.AtomicInteger;
+
/**
- * Concrete implementation of a ClientConsumer.
- *
+ * Concrete implementation of a ClientConsumer.
+ *
* @author <a href="mailto:tim.fox at jboss.com">Tim Fox</a>
* @author <a href="mailto:jmesnil at redhat.com">Jeff Mesnil</a>
- *
+ *
* @version <tt>$Revision: 3783 $</tt> $Id: ServerConsumerImpl.java 3783 2008-02-25 12:15:14Z timfox $
*/
public class ServerConsumerImpl implements ServerConsumer
@@ -81,6 +81,11 @@
private boolean started;
+ /**
+ * if we are a browse only consumer we don't need to worry about acknowledgemenets or being started/stopeed by the session.
+ */
+ private boolean browseOnly;
+
private final StorageManager storageManager;
private final HierarchicalRepository<QueueSettings> queueSettingsRepository;
@@ -101,21 +106,24 @@
final boolean enableFlowControl,
final int maxRate,
final boolean started,
+ final boolean browseOnly,
final StorageManager storageManager,
final HierarchicalRepository<QueueSettings> queueSettingsRepository,
final PostOffice postOffice,
final Channel channel)
{
this.id = id;
-
+
this.messageQueue = messageQueue;
this.filter = filter;
this.session = session;
- this.started = started;
+ this.started = browseOnly || started;
+ this.browseOnly = browseOnly;
+
if (enableFlowControl)
{
availableCredits = new AtomicInteger(0);
@@ -185,7 +193,10 @@
DelayedResult result = channel.replicatePacket(new SessionReplicateDeliveryMessage(id, message.getMessageID()));
- deliveringRefs.add(ref);
+ if(!browseOnly)
+ {
+ deliveringRefs.add(ref);
+ }
if (result == null)
{
@@ -210,6 +221,8 @@
public void close() throws Exception
{
+ browseOnly = false;
+
setStarted(false);
messageQueue.removeConsumer(this);
@@ -257,7 +270,7 @@
{
synchronized (startStopLock)
{
- this.started = started;
+ this.started = browseOnly || started;
}
// Outside the lock
Modified: trunk/src/main/org/jboss/messaging/core/server/impl/ServerSessionImpl.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/server/impl/ServerSessionImpl.java 2008-10-17 16:53:35 UTC (rev 5137)
+++ trunk/src/main/org/jboss/messaging/core/server/impl/ServerSessionImpl.java 2008-10-17 18:08:38 UTC (rev 5138)
@@ -12,23 +12,6 @@
package org.jboss.messaging.core.server.impl;
-import java.nio.ByteBuffer;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.Executor;
-
-import javax.management.Notification;
-import javax.management.NotificationListener;
-import javax.transaction.xa.XAException;
-import javax.transaction.xa.XAResource;
-import javax.transaction.xa.Xid;
-
import org.jboss.messaging.core.client.management.impl.ManagementHelper;
import org.jboss.messaging.core.exception.MessagingException;
import org.jboss.messaging.core.filter.Filter;
@@ -69,6 +52,22 @@
import org.jboss.messaging.util.SimpleString;
import org.jboss.messaging.util.SimpleStringIdGenerator;
+import javax.management.Notification;
+import javax.management.NotificationListener;
+import javax.transaction.xa.XAException;
+import javax.transaction.xa.XAResource;
+import javax.transaction.xa.Xid;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executor;
+
/*
* Session implementation
*
@@ -104,8 +103,6 @@
private final Map<Long, ServerConsumer> consumers = new ConcurrentHashMap<Long, ServerConsumer>();
- private final Map<Long, ServerBrowserImpl> browsers = new ConcurrentHashMap<Long, ServerBrowserImpl>();
-
private final Map<Long, ServerProducer> producers = new ConcurrentHashMap<Long, ServerProducer>();
private final Executor executor;
@@ -140,6 +137,8 @@
private final SimpleStringIdGenerator simpleStringIdGenerator;
+ private final Object queueCopyLock = new Object();
+
// Constructors ---------------------------------------------------------------------------------
public ServerSessionImpl(final String name,
@@ -220,14 +219,6 @@
return id;
}
- public void removeBrowser(final ServerBrowserImpl browser) throws Exception
- {
- if (browsers.remove(browser.getID()) == null)
- {
- throw new IllegalStateException("Cannot find browser with id " + browser.getID() + " to remove");
- }
- }
-
public void removeConsumer(final ServerConsumer consumer) throws Exception
{
if (consumers.remove(consumer.getID()) == null)
@@ -279,15 +270,6 @@
consumers.clear();
- Set<ServerBrowserImpl> browsersClone = new HashSet<ServerBrowserImpl>(browsers.values());
-
- for (ServerBrowserImpl browser : browsersClone)
- {
- browser.close();
- }
-
- browsers.clear();
-
Set<ServerProducer> producersClone = new HashSet<ServerProducer>(producers.values());
for (ServerProducer producer : producersClone)
@@ -314,16 +296,19 @@
{
if (!pager.page(msg))
{
- List<MessageReference> refs = postOffice.route(msg);
-
- if (msg.getDurableRefCount() != 0)
+ synchronized (queueCopyLock)
{
- storageManager.storeMessage(msg);
- }
+ List<MessageReference> refs = postOffice.route(msg);
- for (MessageReference ref : refs)
- {
- ref.getQueue().addLast(ref);
+ if (msg.getDurableRefCount() != 0)
+ {
+ storageManager.storeMessage(msg);
+ }
+
+ for (MessageReference ref : refs)
+ {
+ ref.getQueue().addLast(ref);
+ }
}
}
}
@@ -840,6 +825,37 @@
}
}
+ public void createQueueCopy(final SimpleString queue,
+ final SimpleString queueCopy,
+ final SimpleString filterString,
+ final boolean durable,
+ final boolean temporary) throws Exception
+ {
+ Binding binding = postOffice.getBinding(queue);
+ if (binding == null)
+ {
+ throw new MessagingException(MessagingException.QUEUE_DOES_NOT_EXIST);
+ }
+ //we need to stop messages being routed whilst we make a copy of the queue. altho this will be extremely quick
+ // even for large quantities of references
+ synchronized (queueCopyLock)
+ {
+ createQueue(binding.getAddress(), queueCopy, filterString, durable, temporary);
+ Queue newQ = postOffice.getBinding(queueCopy).getQueue();
+ Filter filter = null;
+
+ if (filterString != null)
+ {
+ filter = new FilterImpl(filterString);
+ }
+ List<MessageReference> refs = binding.getQueue().list(filter);
+ for (MessageReference ref : refs)
+ {
+ newQ.addLast(ref.getMessage().createReference(newQ));
+ }
+ }
+ }
+
public void deleteQueue(final SimpleString queueName) throws Exception
{
Binding binding = postOffice.removeBinding(queueName);
@@ -865,7 +881,8 @@
public SessionCreateConsumerResponseMessage createConsumer(final SimpleString queueName,
final SimpleString filterString,
int windowSize,
- int maxRate) throws Exception
+ int maxRate,
+ final boolean browseOnly) throws Exception
{
Binding binding = postOffice.getBinding(queueName);
@@ -903,6 +920,7 @@
windowSize != -1,
maxRate,
started,
+ browseOnly,
storageManager,
queueSettingsRepository,
postOffice,
@@ -972,25 +990,6 @@
return new SessionBindingQueryResponseMessage(exists, queueNames);
}
- public void createBrowser(final SimpleString queueName, final SimpleString filterString) throws Exception
- {
- Binding binding = postOffice.getBinding(queueName);
-
- if (binding == null)
- {
- throw new MessagingException(MessagingException.QUEUE_DOES_NOT_EXIST);
- }
-
- securityStore.check(binding.getAddress(), CheckType.READ, this);
-
- ServerBrowserImpl browser = new ServerBrowserImpl(idGenerator.generateID(),
- this,
- binding.getQueue(),
- filterString == null ? null : filterString.toString());
-
- browsers.put(browser.getID(), browser);
- }
-
/**
* Create a producer for the specified address
*
@@ -1042,26 +1041,7 @@
return new SessionCreateProducerResponseMessage(initialCredits, maxRateToUse, groupId);
}
- public boolean browserHasNextMessage(final long browserID) throws Exception
- {
- return browsers.get(browserID).hasNextMessage();
- }
- public ServerMessage browserNextMessage(final long browserID) throws Exception
- {
- return browsers.get(browserID).nextMessage();
- }
-
- public void browserReset(final long browserID) throws Exception
- {
- browsers.get(browserID).reset();
- }
-
- public void closeBrowser(final long browserID) throws Exception
- {
- browsers.get(browserID).close();
- }
-
public void closeConsumer(final long consumerID) throws Exception
{
consumers.get(consumerID).close();
Modified: trunk/src/main/org/jboss/messaging/core/server/impl/ServerSessionPacketHandler.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/server/impl/ServerSessionPacketHandler.java 2008-10-17 16:53:35 UTC (rev 5137)
+++ trunk/src/main/org/jboss/messaging/core/server/impl/ServerSessionPacketHandler.java 2008-10-17 18:08:38 UTC (rev 5138)
@@ -13,20 +13,25 @@
package org.jboss.messaging.core.server.impl;
+import org.jboss.messaging.core.exception.MessagingException;
+import org.jboss.messaging.core.logging.Logger;
+import org.jboss.messaging.core.persistence.StorageManager;
+import org.jboss.messaging.core.remoting.Channel;
+import org.jboss.messaging.core.remoting.ChannelHandler;
+import org.jboss.messaging.core.remoting.DelayedResult;
+import org.jboss.messaging.core.remoting.Packet;
+import org.jboss.messaging.core.remoting.impl.wireformat.MessagingExceptionMessage;
+import org.jboss.messaging.core.remoting.impl.wireformat.NullResponseMessage;
import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.SESS_ACKNOWLEDGE;
import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.SESS_ADD_DESTINATION;
import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.SESS_BINDINGQUERY;
-import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.SESS_BROWSER_CLOSE;
-import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.SESS_BROWSER_HASNEXTMESSAGE;
-import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.SESS_BROWSER_NEXTMESSAGE;
-import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.SESS_BROWSER_RESET;
import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.SESS_CLOSE;
import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.SESS_COMMIT;
import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.SESS_CONSUMER_CLOSE;
-import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.SESS_CREATEBROWSER;
import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.SESS_CREATECONSUMER;
import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.SESS_CREATEPRODUCER;
import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.SESS_CREATEQUEUE;
+import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.SESS_CREATEQUEUECOPY;
import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.SESS_DELETE_QUEUE;
import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.SESS_FAILOVER_COMPLETE;
import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.SESS_FLOWTOKEN;
@@ -34,6 +39,7 @@
import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.SESS_PRODUCER_CLOSE;
import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.SESS_QUEUEQUERY;
import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.SESS_REMOVE_DESTINATION;
+import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.SESS_REPLICATE_DELIVERY;
import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.SESS_ROLLBACK;
import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.SESS_SCHEDULED_SEND;
import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.SESS_SEND;
@@ -51,35 +57,14 @@
import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.SESS_XA_SET_TIMEOUT;
import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.SESS_XA_START;
import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.SESS_XA_SUSPEND;
-
-import java.util.List;
-
-import javax.transaction.xa.Xid;
-
-import org.jboss.messaging.core.exception.MessagingException;
-import org.jboss.messaging.core.logging.Logger;
-import org.jboss.messaging.core.persistence.StorageManager;
-import org.jboss.messaging.core.remoting.Channel;
-import org.jboss.messaging.core.remoting.ChannelHandler;
-import org.jboss.messaging.core.remoting.DelayedResult;
-import org.jboss.messaging.core.remoting.Packet;
-import org.jboss.messaging.core.remoting.impl.wireformat.MessagingExceptionMessage;
-import org.jboss.messaging.core.remoting.impl.wireformat.NullResponseMessage;
-import org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl;
import org.jboss.messaging.core.remoting.impl.wireformat.SessionAcknowledgeMessage;
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.SessionBrowseMessage;
-import org.jboss.messaging.core.remoting.impl.wireformat.SessionBrowserCloseMessage;
-import org.jboss.messaging.core.remoting.impl.wireformat.SessionBrowserHasNextMessageMessage;
-import org.jboss.messaging.core.remoting.impl.wireformat.SessionBrowserHasNextMessageResponseMessage;
-import org.jboss.messaging.core.remoting.impl.wireformat.SessionBrowserNextMessageMessage;
-import org.jboss.messaging.core.remoting.impl.wireformat.SessionBrowserResetMessage;
import org.jboss.messaging.core.remoting.impl.wireformat.SessionConsumerCloseMessage;
import org.jboss.messaging.core.remoting.impl.wireformat.SessionConsumerFlowCreditMessage;
-import org.jboss.messaging.core.remoting.impl.wireformat.SessionCreateBrowserMessage;
import org.jboss.messaging.core.remoting.impl.wireformat.SessionCreateConsumerMessage;
import org.jboss.messaging.core.remoting.impl.wireformat.SessionCreateProducerMessage;
+import org.jboss.messaging.core.remoting.impl.wireformat.SessionCreateQueueCopyMessage;
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.SessionProducerCloseMessage;
@@ -104,6 +89,9 @@
import org.jboss.messaging.core.server.ServerMessage;
import org.jboss.messaging.core.server.ServerSession;
+import javax.transaction.xa.Xid;
+import java.util.List;
+
/**
* A ServerSessionPacketHandler
*
@@ -120,7 +108,7 @@
private final Channel channel;
private final StorageManager storageManager;
-
+
public ServerSessionPacketHandler(final ServerSession session,
final Channel channel,
final StorageManager storageManager)
@@ -141,7 +129,7 @@
public void handlePacket(final Packet packet)
{
byte type = packet.getType();
-
+
if (type == SESS_SEND || type == SESS_SCHEDULED_SEND)
{
SessionSendMessage send = (SessionSendMessage)packet;
@@ -158,9 +146,9 @@
}
Packet response = null;
-
+
DelayedResult result = channel.replicatePacket(packet);
-
+
try
{
switch (type)
@@ -171,7 +159,8 @@
response = session.createConsumer(request.getQueueName(),
request.getFilterString(),
request.getWindowSize(),
- request.getMaxRate());
+ request.getMaxRate(),
+ request.isBrowseOnly());
break;
}
case SESS_CREATEQUEUE:
@@ -185,6 +174,17 @@
response = new NullResponseMessage();
break;
}
+ case SESS_CREATEQUEUECOPY:
+ {
+ SessionCreateQueueCopyMessage request = (SessionCreateQueueCopyMessage)packet;
+ session.createQueueCopy(request.getQueueName(),
+ request.getQueueCopyName(),
+ request.getFilterString(),
+ request.isDurable(),
+ request.isTemporary());
+ response = new NullResponseMessage();
+ break;
+ }
case SESS_DELETE_QUEUE:
{
SessionDeleteQueueMessage request = (SessionDeleteQueueMessage)packet;
@@ -204,13 +204,6 @@
response = session.executeBindingQuery(request.getAddress());
break;
}
- case SESS_CREATEBROWSER:
- {
- SessionCreateBrowserMessage request = (SessionCreateBrowserMessage)packet;
- session.createBrowser(request.getQueueName(), request.getFilterString());
- response = new NullResponseMessage();
- break;
- }
case SESS_CREATEPRODUCER:
{
SessionCreateProducerMessage request = (SessionCreateProducerMessage)packet;
@@ -274,7 +267,7 @@
}
case SESS_XA_ROLLBACK:
{
- SessionXARollbackMessage message = (SessionXARollbackMessage)packet;
+ SessionXARollbackMessage message = (SessionXARollbackMessage)packet;
response = session.XARollback(message.getXid());
break;
}
@@ -362,13 +355,6 @@
response = new NullResponseMessage();
break;
}
- case SESS_BROWSER_CLOSE:
- {
- SessionBrowserCloseMessage message = (SessionBrowserCloseMessage)packet;
- session.closeBrowser(message.getBrowserID());
- response = new NullResponseMessage();
- break;
- }
case SESS_FLOWTOKEN:
{
SessionConsumerFlowCreditMessage message = (SessionConsumerFlowCreditMessage)packet;
@@ -397,33 +383,13 @@
}
break;
}
- case SESS_BROWSER_HASNEXTMESSAGE:
- {
- SessionBrowserHasNextMessageMessage message = (SessionBrowserHasNextMessageMessage)packet;
- response = new SessionBrowserHasNextMessageResponseMessage(session.browserHasNextMessage(message.getBrowserID()));
- break;
- }
- case SESS_BROWSER_NEXTMESSAGE:
- {
- SessionBrowserNextMessageMessage message = (SessionBrowserNextMessageMessage)packet;
- ServerMessage smsg = session.browserNextMessage(message.getBrowserID());
- response = new SessionBrowseMessage(smsg);
- break;
- }
- case SESS_BROWSER_RESET:
- {
- SessionBrowserResetMessage message = (SessionBrowserResetMessage)packet;
- session.browserReset(message.getBrowserID());
- response = new NullResponseMessage();
- break;
- }
case SESS_MANAGEMENT_SEND:
{
SessionSendManagementMessage message = (SessionSendManagementMessage)packet;
session.handleManagementMessage(message);
break;
}
- case PacketImpl.SESS_REPLICATE_DELIVERY:
+ case SESS_REPLICATE_DELIVERY:
{
SessionReplicateDeliveryMessage message = (SessionReplicateDeliveryMessage)packet;
session.handleReplicatedDelivery(message.getConsumerID(), message.getMessageID());
@@ -457,7 +423,7 @@
if (response != null)
{
final boolean closeChannel = type == SESS_CLOSE;
-
+
if (result == null)
{
// Not clustered - just send now
Modified: trunk/src/main/org/jboss/messaging/jms/client/JBossQueueBrowser.java
===================================================================
--- trunk/src/main/org/jboss/messaging/jms/client/JBossQueueBrowser.java 2008-10-17 16:53:35 UTC (rev 5137)
+++ trunk/src/main/org/jboss/messaging/jms/client/JBossQueueBrowser.java 2008-10-17 18:08:38 UTC (rev 5138)
@@ -18,25 +18,30 @@
* 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.jms.client;
-import java.util.Enumeration;
+import org.jboss.messaging.core.client.ClientConsumer;
+import org.jboss.messaging.core.client.ClientMessage;
+import org.jboss.messaging.core.client.ClientSession;
+import org.jboss.messaging.core.exception.MessagingException;
+import org.jboss.messaging.core.logging.Logger;
+import org.jboss.messaging.jms.JBossQueue;
+import org.jboss.messaging.util.SimpleString;
import javax.jms.JMSException;
import javax.jms.Queue;
import javax.jms.QueueBrowser;
+import java.util.Enumeration;
+import java.util.NoSuchElementException;
+import java.util.UUID;
-import org.jboss.messaging.core.client.ClientBrowser;
-import org.jboss.messaging.core.client.ClientMessage;
-import org.jboss.messaging.core.exception.MessagingException;
-import org.jboss.messaging.core.logging.Logger;
-
/**
* @author <a href="mailto:tim.fox at jboss.com">Tim Fox</a>
- *
- * $Id$
+ * @author <a href="mailto:andy.taylor at jboss.org">Andy Taylor</a>
+ * <p/>
+ * $Id$
*/
public class JBossQueueBrowser implements QueueBrowser
{
@@ -44,53 +49,72 @@
private static final Logger log = Logger.getLogger(JBossQueueBrowser.class);
+ private static final long NEXT_MESSAGE_TIMEOUT = 1000;
+
// Static ---------------------------------------------------------------------------------------
// Attributes -----------------------------------------------------------------------------------
- private ClientBrowser browser;
- private Queue queue;
- private String messageSelector;
+ private ClientSession session;
+ private ClientConsumer consumer;
+
+ private JBossQueue queue;
+
+ private SimpleString messageSelector;
+
+ private SimpleString queueName;
+
// Constructors ---------------------------------------------------------------------------------
- public JBossQueueBrowser(Queue queue, String messageSelector, ClientBrowser browser)
+ public JBossQueueBrowser(JBossQueue queue, String messageSelector, ClientSession session)
{
- this.browser = browser;
+ this.session = session;
this.queue = queue;
- this.messageSelector = messageSelector;
+ if(messageSelector != null)
+ {
+ this. messageSelector = new SimpleString(SelectorTranslator.convertToJBMFilterString(messageSelector));
+ }
}
// QueueBrowser implementation -------------------------------------------------------------------
public void close() throws JMSException
{
- try
+ if (consumer != null)
{
- browser.close();
+ try
+ {
+ consumer.close();
+ session.deleteQueue(queueName);
+ }
+ catch (MessagingException e)
+ {
+ throw JMSExceptionHelper.convertFromMessagingException(e);
+ }
}
- catch (MessagingException e)
- {
- throw JMSExceptionHelper.convertFromMessagingException(e);
- }
}
public Enumeration getEnumeration() throws JMSException
{
try
{
- browser.reset();
+ close();
+ queueName = new SimpleString(UUID.randomUUID().toString());
+ session.createQueueCopy(queue.getSimpleAddress(), queueName, messageSelector, false, true);
+ consumer = session.createConsumer(queueName, null, false, true);
return new BrowserEnumeration();
}
catch (MessagingException e)
{
- throw JMSExceptionHelper.convertFromMessagingException(e);
+ throw JMSExceptionHelper.convertFromMessagingException(e);
}
+
}
public String getMessageSelector() throws JMSException
{
- return messageSelector;
+ return messageSelector == null?null:messageSelector.toString();
}
public Queue getQueue() throws JMSException
@@ -102,7 +126,7 @@
public String toString()
{
- return "JBossQueueBrowser->" + browser;
+ return "JBossQueueBrowser->" + consumer;
}
// Package protected ----------------------------------------------------------------------------
@@ -115,33 +139,48 @@
private class BrowserEnumeration implements Enumeration
{
+ ClientMessage current = null;
+
public boolean hasMoreElements()
{
- try
- {
- return browser.hasNextMessage();
- }
- catch (MessagingException e)
+ if (current == null)
{
- throw new IllegalStateException(e.getMessage());
+ try
+ {
+ //todo change this to consumer.receiveImmediate() once https://jira.jboss.org/jira/browse/JBMESSAGING-1432 is completed
+ current = consumer.receive(NEXT_MESSAGE_TIMEOUT);
+ }
+ catch (MessagingException e)
+ {
+ return false;
+ }
}
+ return current != null;
}
public Object nextElement()
{
- try
+ JBossMessage jbm;
+ if (hasMoreElements())
{
- ClientMessage message = browser.nextMessage();
+ ClientMessage next = current;
+ current = null;
+ jbm = JBossMessage.createMessage(next, session);
+ try
+ {
+ jbm.doBeforeReceive();
+ }
+ catch (Exception e)
+ {
+ log.error("Failed to prepare message", e);
- JBossMessage jbm = JBossMessage.createMessage(message, null);
-
- jbm.doBeforeReceive();
-
+ return null;
+ }
return jbm;
}
- catch (Exception e)
+ else
{
- throw new IllegalStateException(e.getMessage());
+ throw new NoSuchElementException();
}
}
}
Modified: trunk/src/main/org/jboss/messaging/jms/client/JBossSession.java
===================================================================
--- trunk/src/main/org/jboss/messaging/jms/client/JBossSession.java 2008-10-17 16:53:35 UTC (rev 5137)
+++ trunk/src/main/org/jboss/messaging/jms/client/JBossSession.java 2008-10-17 18:08:38 UTC (rev 5138)
@@ -22,10 +22,19 @@
package org.jboss.messaging.jms.client;
-import java.io.Serializable;
-import java.util.HashSet;
-import java.util.Set;
-import java.util.UUID;
+import org.jboss.messaging.core.client.ClientConsumer;
+import org.jboss.messaging.core.client.ClientProducer;
+import org.jboss.messaging.core.client.ClientSession;
+import org.jboss.messaging.core.exception.MessagingException;
+import org.jboss.messaging.core.logging.Logger;
+import org.jboss.messaging.core.remoting.impl.wireformat.SessionBindingQueryResponseMessage;
+import org.jboss.messaging.core.remoting.impl.wireformat.SessionQueueQueryResponseMessage;
+import org.jboss.messaging.jms.JBossDestination;
+import org.jboss.messaging.jms.JBossQueue;
+import org.jboss.messaging.jms.JBossTemporaryQueue;
+import org.jboss.messaging.jms.JBossTemporaryTopic;
+import org.jboss.messaging.jms.JBossTopic;
+import org.jboss.messaging.util.SimpleString;
import javax.jms.BytesMessage;
import javax.jms.Destination;
@@ -58,33 +67,22 @@
import javax.jms.XASession;
import javax.jms.XATopicSession;
import javax.transaction.xa.XAResource;
+import java.io.Serializable;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.UUID;
-import org.jboss.messaging.core.client.ClientBrowser;
-import org.jboss.messaging.core.client.ClientConsumer;
-import org.jboss.messaging.core.client.ClientProducer;
-import org.jboss.messaging.core.client.ClientSession;
-import org.jboss.messaging.core.exception.MessagingException;
-import org.jboss.messaging.core.logging.Logger;
-import org.jboss.messaging.core.remoting.impl.wireformat.SessionBindingQueryResponseMessage;
-import org.jboss.messaging.core.remoting.impl.wireformat.SessionQueueQueryResponseMessage;
-import org.jboss.messaging.jms.JBossDestination;
-import org.jboss.messaging.jms.JBossQueue;
-import org.jboss.messaging.jms.JBossTemporaryQueue;
-import org.jboss.messaging.jms.JBossTemporaryTopic;
-import org.jboss.messaging.jms.JBossTopic;
-import org.jboss.messaging.util.SimpleString;
-
/**
- *
+ *
* Note that we *do not* support JMS ASF (Application Server Facilities) optional
* constructs such as ConnectionConsumer
- *
+ *
* @author <a href="mailto:ovidiu at feodorov.com">Ovidiu Feodorov</a>
* @author <a href="mailto:tim.fox at jboss.com">Tim Fox</a>
* @author <a href="mailto:ataylor at redhat.com">Andy Taylor</a>
- *
+ *
* @version <tt>$Revision$</tt>
- *
+ *
* $Id$
*/
public class JBossSession implements Session, XASession, QueueSession, XAQueueSession, TopicSession, XATopicSession
@@ -638,17 +636,19 @@
try
{
- String coreSelector = SelectorTranslator.convertToJBMFilterString(filterString);
-
- ClientBrowser browser = session.createBrowser(jbq.getSimpleAddress(),
- coreSelector == null ? null : new SimpleString(coreSelector));
-
- return new JBossQueueBrowser(queue, filterString, browser);
+ SessionBindingQueryResponseMessage message = session.bindingQuery(new SimpleString(jbq.getAddress()));
+ if(!message.isExists())
+ {
+ throw new InvalidDestinationException(jbq.getAddress() + " does not exist");
+ }
}
catch (MessagingException e)
{
- throw JMSExceptionHelper.convertFromMessagingException(e);
+ JMSExceptionHelper.convertFromMessagingException(e);
}
+
+ return new JBossQueueBrowser(jbq, filterString, session);
+
}
public TemporaryQueue createTemporaryQueue() throws JMSException
Modified: trunk/tests/jms-tests/src/org/jboss/test/messaging/jms/BrowserTest.java
===================================================================
--- trunk/tests/jms-tests/src/org/jboss/test/messaging/jms/BrowserTest.java 2008-10-17 16:53:35 UTC (rev 5137)
+++ trunk/tests/jms-tests/src/org/jboss/test/messaging/jms/BrowserTest.java 2008-10-17 18:08:38 UTC (rev 5138)
@@ -21,7 +21,7 @@
*/
package org.jboss.test.messaging.jms;
-import java.util.Enumeration;
+import org.jboss.messaging.jms.JBossQueue;
import javax.jms.Connection;
import javax.jms.InvalidDestinationException;
@@ -31,10 +31,9 @@
import javax.jms.QueueBrowser;
import javax.jms.Session;
import javax.jms.TextMessage;
+import java.util.Enumeration;
-import org.jboss.messaging.jms.JBossQueue;
-
/**
* @author <a href="mailto:tim.fox at jboss.com">Tim Fox</a>
* @author <a href="mailto:ovidiu at feodorov.com">Ovidiu Feodorov</a>
@@ -138,7 +137,7 @@
}
QueueBrowser browser = session.createBrowser(queue1);
-
+
assertEquals(browser.getQueue(), queue1);
assertNull(browser.getMessageSelector());
Modified: trunk/tests/jms-tests/src/org/jboss/test/messaging/jms/SecurityTest.java
===================================================================
--- trunk/tests/jms-tests/src/org/jboss/test/messaging/jms/SecurityTest.java 2008-10-17 16:53:35 UTC (rev 5137)
+++ trunk/tests/jms-tests/src/org/jboss/test/messaging/jms/SecurityTest.java 2008-10-17 18:08:38 UTC (rev 5138)
@@ -21,9 +21,9 @@
*/
package org.jboss.test.messaging.jms;
-import java.util.ArrayList;
-import java.util.HashSet;
-import java.util.Set;
+import org.jboss.messaging.core.logging.Logger;
+import org.jboss.messaging.core.security.Role;
+import org.jboss.test.messaging.tools.ServerManagement;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
@@ -36,11 +36,10 @@
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.Topic;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.Set;
-import org.jboss.messaging.core.logging.Logger;
-import org.jboss.messaging.core.security.Role;
-import org.jboss.test.messaging.tools.ServerManagement;
-
/**
* Test JMS Security.
*
@@ -370,7 +369,7 @@
{
conn = cf.createConnection("nobody", "nobody");
Session sess = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
- sess.createBrowser(queue1);
+ sess.createBrowser(queue1).getEnumeration();
fail("should throw JMSSecurityException");
}
catch (JMSSecurityException e)
Deleted: trunk/tests/src/org/jboss/messaging/tests/timing/core/client/ScheduledMessageTest.java
===================================================================
--- trunk/tests/src/org/jboss/messaging/tests/timing/core/client/ScheduledMessageTest.java 2008-10-17 16:53:35 UTC (rev 5137)
+++ trunk/tests/src/org/jboss/messaging/tests/timing/core/client/ScheduledMessageTest.java 2008-10-17 18:08:38 UTC (rev 5138)
@@ -1,230 +0,0 @@
-/*
- * JBoss, Home of Professional Open Source
- * Copyright 2005-2008, Red Hat Middleware LLC, and individual contributors
- * 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.timing.core.client;
-
-import org.jboss.messaging.core.client.ClientConsumer;
-import org.jboss.messaging.core.client.ClientMessage;
-import org.jboss.messaging.core.client.ClientProducer;
-import org.jboss.messaging.core.client.ClientSession;
-import org.jboss.messaging.core.client.ClientSessionFactory;
-import org.jboss.messaging.core.client.impl.ClientSessionFactoryImpl;
-import org.jboss.messaging.core.config.TransportConfiguration;
-import org.jboss.messaging.core.config.impl.ConfigurationImpl;
-import org.jboss.messaging.core.server.MessagingService;
-import org.jboss.messaging.core.server.impl.MessagingServiceImpl;
-import org.jboss.messaging.core.transaction.impl.XidImpl;
-import org.jboss.messaging.jms.client.JBossTextMessage;
-import org.jboss.messaging.tests.util.UnitTestCase;
-import org.jboss.messaging.util.SimpleString;
-import org.jboss.util.id.GUID;
-
-import javax.transaction.xa.XAResource;
-import javax.transaction.xa.Xid;
-import java.io.File;
-import java.util.Calendar;
-
-/**
- * @author <a href="mailto:andy.taylor at jboss.org">Andy Taylor</a>
- */
-public class ScheduledMessageTest extends UnitTestCase
-{
- private static final String ACCEPTOR_FACTORY = "org.jboss.messaging.core.remoting.impl.invm.InVMAcceptorFactory";
-
- private static final String CONNECTOR_FACTORY = "org.jboss.messaging.core.remoting.impl.invm.InVMConnectorFactory";
-
- private String journalDir = System.getProperty("java.io.tmpdir", "/tmp") + "/ScheduledMessageRecoveryTest/journal";
-
- private String bindingsDir = System.getProperty("java.io.tmpdir", "/tmp") + "/ScheduledMessageRecoveryTest/bindings";
-
- private String pageDir = System.getProperty("java.io.tmpdir", "/tmp") + "/ScheduledMessageRecoveryTest/page";
-
- private SimpleString atestq = new SimpleString("ascheduledtestq");
-
- private MessagingService messagingService;
-
- private ConfigurationImpl configuration;
-
- protected void setUp() throws Exception
- {
- File file = new File(journalDir);
- File file2 = new File(bindingsDir);
- File file3 = new File(pageDir);
- deleteDirectory(file);
- file.mkdirs();
- deleteDirectory(file2);
- file2.mkdirs();
- deleteDirectory(file3);
- file3.mkdirs();
- configuration = new ConfigurationImpl();
- configuration.setSecurityEnabled(false);
- configuration.setJournalMinFiles(2);
- configuration.setPagingDirectory(pageDir);
- }
-
- protected void tearDown() throws Exception
- {
- if (messagingService != null)
- {
- try
- {
- messagingService.stop();
- }
- catch (Exception e)
- {
- //ignore
- }
- }
- new File(journalDir).delete();
- new File(bindingsDir).delete();
- new File(pageDir).delete();
- }
-
- public void testRecoveredMessageDeliveredCorrectly() throws Exception
- {
-
- TransportConfiguration transportConfig = new TransportConfiguration(ACCEPTOR_FACTORY);
- configuration.getAcceptorConfigurations().add(transportConfig);
- messagingService = MessagingServiceImpl.newNioStorageMessagingServer(configuration, journalDir, bindingsDir);
- //start the server
- messagingService.start();
- //then we create a client as normal
- ClientSessionFactory sessionFactory = new ClientSessionFactoryImpl(new TransportConfiguration(CONNECTOR_FACTORY));
- ClientSession session = sessionFactory.createSession(false, true, false, false);
- session.createQueue(atestq, atestq, null, true, true);
- ClientProducer producer = session.createProducer(atestq);
- ClientMessage message = session.createClientMessage(JBossTextMessage.TYPE, false, 0,
- System.currentTimeMillis(), (byte) 1);
- message.getBody().putString("testINVMCoreClient");
- message.getBody().flip();
- message.setDurable(true);
- Calendar cal = Calendar.getInstance();
- cal.roll(Calendar.SECOND, 10);
- producer.send(message, cal.getTimeInMillis());
-
- producer.close();
- session.close();
- messagingService.stop();
- messagingService = null;
- messagingService = MessagingServiceImpl.newNioStorageMessagingServer(configuration, journalDir, bindingsDir);
- messagingService.start();
-
- sessionFactory = new ClientSessionFactoryImpl(new TransportConfiguration(CONNECTOR_FACTORY));
-
- session = sessionFactory.createSession(false, true, true, false);
-
- ClientConsumer consumer = session.createConsumer(atestq);
-
- session.start();
-
- ClientMessage message2 = consumer.receive(10000);
- assertTrue(System.currentTimeMillis() >= cal.getTimeInMillis());
- assertEquals("testINVMCoreClient", message2.getBody().getString());
-
- message2.acknowledge();
- session.close();
- }
-
- public void testRecoveredTxMessageDeliveredCorrectly() throws Exception
- {
- Xid xid = new XidImpl("xa1".getBytes(), 1, new GUID().toString().getBytes());
- TransportConfiguration transportConfig = new TransportConfiguration(ACCEPTOR_FACTORY);
- configuration.getAcceptorConfigurations().add(transportConfig);
- messagingService = MessagingServiceImpl.newNioStorageMessagingServer(configuration, journalDir, bindingsDir);
- //start the server
- messagingService.start();
- //then we create a client as normal
- ClientSessionFactory sessionFactory = new ClientSessionFactoryImpl(new TransportConfiguration(CONNECTOR_FACTORY));
- ClientSession session = sessionFactory.createSession(true, false, false, false);
- session.createQueue(atestq, atestq, null, true, true);
- session.start(xid, XAResource.TMNOFLAGS);
- ClientProducer producer = session.createProducer(atestq);
- ClientMessage message = session.createClientMessage(JBossTextMessage.TYPE, false, 0,
- System.currentTimeMillis(), (byte) 1);
- message.getBody().putString("testINVMCoreClient");
- message.getBody().flip();
- message.setDurable(true);
- Calendar cal = Calendar.getInstance();
- cal.roll(Calendar.SECOND, 10);
- producer.send(message, cal.getTimeInMillis());
- session.end(xid, XAResource.TMSUCCESS);
- session.prepare(xid);
- producer.close();
- session.close();
- messagingService.stop();
- messagingService = null;
- messagingService = MessagingServiceImpl.newNioStorageMessagingServer(configuration, journalDir, bindingsDir);
- messagingService.start();
-
- sessionFactory = new ClientSessionFactoryImpl(new TransportConfiguration(CONNECTOR_FACTORY));
-
- session = sessionFactory.createSession(true, false, false, false);
- session.commit(xid, true);
- ClientConsumer consumer = session.createConsumer(atestq);
-
- session.start();
-
- ClientMessage message2 = consumer.receive(10000);
- assertTrue(System.currentTimeMillis() >= cal.getTimeInMillis());
- assertEquals("testINVMCoreClient", message2.getBody().getString());
-
- message2.acknowledge();
- session.close();
- }
-
- public void testPagedMessageDeliveredCorrectly() throws Exception
- {
-
- TransportConfiguration transportConfig = new TransportConfiguration(ACCEPTOR_FACTORY);
- configuration.getAcceptorConfigurations().add(transportConfig);
- configuration.setPagingMaxGlobalSizeBytes(0);
- messagingService = MessagingServiceImpl.newNioStorageMessagingServer(configuration, journalDir, bindingsDir);
- //start the server
- messagingService.start();
- //then we create a client as normal
- ClientSessionFactory sessionFactory = new ClientSessionFactoryImpl(new TransportConfiguration(CONNECTOR_FACTORY));
- ClientSession session = sessionFactory.createSession(false, true, false, false);
- session.createQueue(atestq, atestq, null, true, true);
- ClientProducer producer = session.createProducer(atestq);
- ClientMessage message = session.createClientMessage(JBossTextMessage.TYPE, false, 0,
- System.currentTimeMillis(), (byte) 1);
- message.getBody().putString("testINVMCoreClient");
- message.getBody().flip();
- message.setDurable(true);
- Calendar cal = Calendar.getInstance();
- cal.roll(Calendar.SECOND, 10);
- producer.send(message, cal.getTimeInMillis());
-
- producer.close();
-
-
- ClientConsumer consumer = session.createConsumer(atestq);
-
- session.start();
-
- ClientMessage message2 = consumer.receive(10000);
- assertTrue(System.currentTimeMillis() >= cal.getTimeInMillis());
- assertEquals("testINVMCoreClient", message2.getBody().getString());
-
- message2.acknowledge();
- session.close();
- }
-}
Modified: trunk/tests/src/org/jboss/messaging/tests/unit/jms/client/JBossQueueBrowserTest.java
===================================================================
--- trunk/tests/src/org/jboss/messaging/tests/unit/jms/client/JBossQueueBrowserTest.java 2008-10-17 16:53:35 UTC (rev 5137)
+++ trunk/tests/src/org/jboss/messaging/tests/unit/jms/client/JBossQueueBrowserTest.java 2008-10-17 18:08:38 UTC (rev 5138)
@@ -22,34 +22,30 @@
package org.jboss.messaging.tests.unit.jms.client;
+import junit.framework.TestCase;
+import org.easymock.EasyMock;
import static org.easymock.EasyMock.createStrictMock;
import static org.easymock.EasyMock.expect;
-import static org.easymock.EasyMock.expectLastCall;
import static org.easymock.EasyMock.replay;
import static org.easymock.EasyMock.verify;
-import static org.jboss.messaging.tests.util.RandomUtil.randomString;
-
-import java.util.Enumeration;
-
-import javax.jms.JMSException;
-import javax.jms.Message;
-import javax.jms.Queue;
-
-import junit.framework.TestCase;
-
-import org.jboss.messaging.core.client.ClientBrowser;
+import org.jboss.messaging.core.client.ClientConsumer;
import org.jboss.messaging.core.client.ClientMessage;
-import org.jboss.messaging.core.exception.MessagingException;
+import org.jboss.messaging.core.client.ClientSession;
import org.jboss.messaging.core.remoting.spi.MessagingBuffer;
import org.jboss.messaging.jms.JBossQueue;
import org.jboss.messaging.jms.client.JBossMessage;
import org.jboss.messaging.jms.client.JBossQueueBrowser;
+import static org.jboss.messaging.tests.util.RandomUtil.randomString;
+import org.jboss.messaging.util.SimpleString;
+import javax.jms.Message;
+import java.util.Enumeration;
+
/**
* @author <a href="mailto:jmesnil at redhat.com">Jeff Mesnil</a>
- *
+ *
* @version <tt>$Revision$</tt>
- *
+ *
*/
public class JBossQueueBrowserTest extends TestCase
{
@@ -66,8 +62,8 @@
public void testGetMessageSelector() throws Exception
{
String messageSelector = "color = 'green'";
- Queue queue = new JBossQueue(randomString());
- ClientBrowser clientBrowser = createStrictMock(ClientBrowser.class);
+ JBossQueue queue = new JBossQueue(randomString());
+ ClientSession clientBrowser = createStrictMock(ClientSession.class);
replay(clientBrowser);
JBossQueueBrowser browser = new JBossQueueBrowser(queue, messageSelector,
@@ -79,8 +75,8 @@
public void testGetQueue() throws Exception
{
- Queue queue = new JBossQueue(randomString());
- ClientBrowser clientBrowser = createStrictMock(ClientBrowser.class);
+ JBossQueue queue = new JBossQueue(randomString());
+ ClientSession clientBrowser = createStrictMock(ClientSession.class);
replay(clientBrowser);
JBossQueueBrowser browser = new JBossQueueBrowser(queue, null,
@@ -92,9 +88,8 @@
public void testClose() throws Exception
{
- Queue queue = new JBossQueue(randomString());
- ClientBrowser clientBrowser = createStrictMock(ClientBrowser.class);
- clientBrowser.close();
+ JBossQueue queue = new JBossQueue(randomString());
+ ClientSession clientBrowser = createStrictMock(ClientSession.class);
replay(clientBrowser);
JBossQueueBrowser browser = new JBossQueueBrowser(queue, null,
@@ -105,10 +100,10 @@
verify(clientBrowser);
}
- public void testCloseThrowsException() throws Exception
+/* public void testCloseThrowsException() throws Exception
{
- Queue queue = new JBossQueue(randomString());
- ClientBrowser clientBrowser = createStrictMock(ClientBrowser.class);
+ JBossQueue queue = new JBossQueue(randomString());
+ ClientSession clientBrowser = createStrictMock(ClientSession.class);
clientBrowser.close();
expectLastCall().andThrow(new MessagingException());
@@ -126,62 +121,43 @@
}
verify(clientBrowser);
- }
+ }*/
public void testGetEnumeration() throws Exception
{
- Queue queue = new JBossQueue(randomString());
- ClientBrowser clientBrowser = createStrictMock(ClientBrowser.class);
- clientBrowser.reset();
- replay(clientBrowser);
+ JBossQueue queue = new JBossQueue(randomString());
+ ClientConsumer consumer = createStrictMock(ClientConsumer.class);
+ ClientSession session = createStrictMock(ClientSession.class);
+ session.createQueueCopy((SimpleString) EasyMock.eq(queue.getSimpleAddress()), (SimpleString) EasyMock.anyObject(), (SimpleString) EasyMock.eq(null), EasyMock.anyBoolean(),EasyMock.anyBoolean());
+ expect(session.createConsumer((SimpleString) EasyMock.anyObject(), (SimpleString) EasyMock.anyObject(), EasyMock.anyBoolean(), EasyMock.anyBoolean())).andReturn(consumer);
+ replay(session, consumer);
JBossQueueBrowser browser = new JBossQueueBrowser(queue, null,
- clientBrowser);
+ session);
Enumeration enumeration = browser.getEnumeration();
assertNotNull(enumeration);
- verify(clientBrowser);
+ verify(session, consumer);
}
- public void testGetEnumerationThrowsException() throws Exception
- {
- Queue queue = new JBossQueue(randomString());
- ClientBrowser clientBrowser = createStrictMock(ClientBrowser.class);
- clientBrowser.reset();
- expectLastCall().andThrow(new MessagingException());
- replay(clientBrowser);
-
- JBossQueueBrowser browser = new JBossQueueBrowser(queue, null,
- clientBrowser);
-
- try
- {
- browser.getEnumeration();
- fail("JMSException");
- } catch (JMSException e)
- {
- }
-
- verify(clientBrowser);
- }
-
public void testGetEnumerationWithOneMessage() throws Exception
{
- Queue queue = new JBossQueue(randomString());
- ClientBrowser clientBrowser = createStrictMock(ClientBrowser.class);
+ JBossQueue queue = new JBossQueue(randomString());
ClientMessage clientMessage = createStrictMock(ClientMessage.class);
MessagingBuffer buffer = createStrictMock(MessagingBuffer.class);
- clientBrowser.reset();
- expect(clientBrowser.hasNextMessage()).andReturn(true);
+ ClientConsumer consumer = createStrictMock(ClientConsumer.class);
+ ClientSession session = createStrictMock(ClientSession.class);
+ session.createQueueCopy((SimpleString) EasyMock.eq(queue.getSimpleAddress()), (SimpleString) EasyMock.anyObject(), (SimpleString) EasyMock.eq(null), EasyMock.anyBoolean(),EasyMock.anyBoolean());
+ expect(session.createConsumer((SimpleString) EasyMock.anyObject(), (SimpleString) EasyMock.anyObject(), EasyMock.anyBoolean(), EasyMock.anyBoolean())).andReturn(consumer);
+ expect(consumer.receive(1000)).andReturn(clientMessage);
expect(clientMessage.getType()).andReturn(JBossMessage.TYPE);
expect(clientMessage.getBody()).andStubReturn(buffer);
- expect(clientBrowser.nextMessage()).andReturn(clientMessage);
- expect(clientBrowser.hasNextMessage()).andReturn(false);
- replay(clientMessage, clientBrowser);
+ expect(consumer.receive(1000)).andReturn(null);
+ replay(clientMessage, session, consumer);
JBossQueueBrowser browser = new JBossQueueBrowser(queue, null,
- clientBrowser);
+ session);
Enumeration enumeration = browser.getEnumeration();
assertNotNull(enumeration);
@@ -189,61 +165,9 @@
Message message = (Message) enumeration.nextElement();
assertFalse(enumeration.hasMoreElements());
- verify(clientMessage, clientBrowser);
+ verify(clientMessage, session, consumer);
}
- public void testGetEnumerationWithHasMoreElementsThrowsException()
- throws Exception
- {
- Queue queue = new JBossQueue(randomString());
- ClientBrowser clientBrowser = createStrictMock(ClientBrowser.class);
- clientBrowser.reset();
- expect(clientBrowser.hasNextMessage()).andThrow(new MessagingException());
- replay(clientBrowser);
-
- JBossQueueBrowser browser = new JBossQueueBrowser(queue, null,
- clientBrowser);
-
- Enumeration enumeration = browser.getEnumeration();
- assertNotNull(enumeration);
-
- try
- {
- enumeration.hasMoreElements();
- fail("IllegalStateException");
- } catch (IllegalStateException e)
- {
- }
-
- verify(clientBrowser);
- }
-
- public void testGetEnumerationWithNextThrowsException() throws Exception
- {
- Queue queue = new JBossQueue(randomString());
- ClientBrowser clientBrowser = createStrictMock(ClientBrowser.class);
- clientBrowser.reset();
- expect(clientBrowser.hasNextMessage()).andReturn(true);
- expect(clientBrowser.nextMessage()).andThrow(new MessagingException());
- replay(clientBrowser);
-
- JBossQueueBrowser browser = new JBossQueueBrowser(queue, null,
- clientBrowser);
-
- Enumeration enumeration = browser.getEnumeration();
- assertNotNull(enumeration);
- assertTrue(enumeration.hasMoreElements());
-
- try
- {
- enumeration.nextElement();
- fail("IllegalStateException");
- } catch (IllegalStateException e)
- {
- }
-
- verify(clientBrowser);
- }
// Package protected ---------------------------------------------
// Protected -----------------------------------------------------
Modified: trunk/tests/src/org/jboss/messaging/tests/unit/jms/client/JBossSessionTest.java
===================================================================
--- trunk/tests/src/org/jboss/messaging/tests/unit/jms/client/JBossSessionTest.java 2008-10-17 16:53:35 UTC (rev 5137)
+++ trunk/tests/src/org/jboss/messaging/tests/unit/jms/client/JBossSessionTest.java 2008-10-17 18:08:38 UTC (rev 5138)
@@ -22,6 +22,8 @@
package org.jboss.messaging.tests.unit.jms.client;
+import junit.framework.TestCase;
+import org.easymock.EasyMock;
import static org.easymock.EasyMock.createStrictMock;
import static org.easymock.EasyMock.eq;
import static org.easymock.EasyMock.expect;
@@ -30,13 +32,28 @@
import static org.easymock.EasyMock.isNull;
import static org.easymock.EasyMock.replay;
import static org.easymock.EasyMock.verify;
+import org.jboss.messaging.core.client.ClientConsumer;
+import org.jboss.messaging.core.client.ClientMessage;
+import org.jboss.messaging.core.client.ClientProducer;
+import org.jboss.messaging.core.client.ClientSession;
+import org.jboss.messaging.core.client.ClientSessionFactory;
+import org.jboss.messaging.core.client.impl.ClientMessageImpl;
+import org.jboss.messaging.core.exception.MessagingException;
+import org.jboss.messaging.core.remoting.impl.ByteBufferWrapper;
+import org.jboss.messaging.core.remoting.impl.wireformat.SessionBindingQueryResponseMessage;
+import org.jboss.messaging.core.remoting.impl.wireformat.SessionQueueQueryResponseMessage;
+import org.jboss.messaging.jms.JBossDestination;
+import org.jboss.messaging.jms.JBossQueue;
+import org.jboss.messaging.jms.JBossTemporaryQueue;
+import org.jboss.messaging.jms.JBossTemporaryTopic;
+import org.jboss.messaging.jms.JBossTopic;
+import org.jboss.messaging.jms.client.JBossConnection;
+import org.jboss.messaging.jms.client.JBossMessage;
+import org.jboss.messaging.jms.client.JBossSession;
import static org.jboss.messaging.tests.util.RandomUtil.randomSimpleString;
import static org.jboss.messaging.tests.util.RandomUtil.randomString;
+import org.jboss.messaging.util.SimpleString;
-import java.nio.ByteBuffer;
-import java.util.ArrayList;
-import java.util.List;
-
import javax.jms.BytesMessage;
import javax.jms.Destination;
import javax.jms.IllegalStateException;
@@ -50,7 +67,6 @@
import javax.jms.MessageProducer;
import javax.jms.ObjectMessage;
import javax.jms.Queue;
-import javax.jms.QueueBrowser;
import javax.jms.QueueSender;
import javax.jms.QueueSession;
import javax.jms.Session;
@@ -63,31 +79,10 @@
import javax.jms.TopicSession;
import javax.jms.TransactionInProgressException;
import javax.transaction.xa.XAResource;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
-import junit.framework.TestCase;
-
-import org.easymock.EasyMock;
-import org.jboss.messaging.core.client.ClientBrowser;
-import org.jboss.messaging.core.client.ClientConsumer;
-import org.jboss.messaging.core.client.ClientMessage;
-import org.jboss.messaging.core.client.ClientProducer;
-import org.jboss.messaging.core.client.ClientSession;
-import org.jboss.messaging.core.client.ClientSessionFactory;
-import org.jboss.messaging.core.client.impl.ClientMessageImpl;
-import org.jboss.messaging.core.exception.MessagingException;
-import org.jboss.messaging.core.remoting.impl.ByteBufferWrapper;
-import org.jboss.messaging.core.remoting.impl.wireformat.SessionBindingQueryResponseMessage;
-import org.jboss.messaging.core.remoting.impl.wireformat.SessionQueueQueryResponseMessage;
-import org.jboss.messaging.jms.JBossDestination;
-import org.jboss.messaging.jms.JBossQueue;
-import org.jboss.messaging.jms.JBossTemporaryQueue;
-import org.jboss.messaging.jms.JBossTemporaryTopic;
-import org.jboss.messaging.jms.JBossTopic;
-import org.jboss.messaging.jms.client.JBossConnection;
-import org.jboss.messaging.jms.client.JBossMessage;
-import org.jboss.messaging.jms.client.JBossSession;
-import org.jboss.messaging.util.SimpleString;
-
/**
* @author <a href="mailto:jmesnil at redhat.com">Jeff Mesnil</a>
*
@@ -1214,9 +1209,9 @@
verify(clientConsumer);
}
-
-
- public void testCreateBrowser() throws Exception
+
+
+ /*public void testCreateBrowser() throws Exception
{
JBossQueue queue = new JBossQueue(randomString());
ClientBrowser clientBrowser = createStrictMock(ClientBrowser.class);
@@ -1305,7 +1300,7 @@
assertNotNull(browser);
verify(clientBrowser);
- }
+ }*/
public void testCreateBrowserFromTopicSession() throws Exception
{
More information about the jboss-cvs-commits
mailing list