You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by rg...@apache.org on 2008/04/21 16:04:00 UTC
svn commit: r650148 [9/10] - in
/incubator/qpid/branches/broker-queue-refactor/java: broker/ broker/etc/
broker/src/main/java/org/apache/qpid/server/
broker/src/main/java/org/apache/qpid/server/ack/
broker/src/main/java/org/apache/qpid/server/configura...
Modified: incubator/qpid/branches/broker-queue-refactor/java/broker/src/test/java/org/apache/qpid/server/exchange/DestWildExchangeTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/test/java/org/apache/qpid/server/exchange/DestWildExchangeTest.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/test/java/org/apache/qpid/server/exchange/DestWildExchangeTest.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/test/java/org/apache/qpid/server/exchange/DestWildExchangeTest.java Mon Apr 21 07:03:35 2008
@@ -22,9 +22,7 @@
import junit.framework.TestCase;
import junit.framework.Assert;
-import org.apache.qpid.server.queue.AMQQueue;
-import org.apache.qpid.server.queue.AMQMessage;
-import org.apache.qpid.server.queue.MessageHandleFactory;
+import org.apache.qpid.server.queue.*;
import org.apache.qpid.server.virtualhost.VirtualHost;
import org.apache.qpid.server.registry.ApplicationRegistry;
import org.apache.qpid.server.txn.NonTransactionalContext;
@@ -33,69 +31,62 @@
import org.apache.qpid.server.store.MemoryMessageStore;
import org.apache.qpid.server.store.StoreContext;
import org.apache.qpid.server.RequiredDeliveryException;
+import org.apache.qpid.server.protocol.TestMinaProtocolSession;
import org.apache.qpid.AMQException;
import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.framing.ContentHeaderBody;
import org.apache.qpid.framing.abstraction.MessagePublishInfo;
-import java.util.HashSet;
-import java.util.List;
import java.util.LinkedList;
public class DestWildExchangeTest extends TestCase
{
- DestWildExchange _exchange;
+ TopicExchange _exchange;
VirtualHost _vhost;
MessageStore _store;
StoreContext _context;
+ TestMinaProtocolSession _protocolSession;
+
public void setUp() throws AMQException
{
- _exchange = new DestWildExchange();
+ _exchange = new TopicExchange();
_vhost = ApplicationRegistry.getInstance().getVirtualHostRegistry().getVirtualHosts().iterator().next();
_store = new MemoryMessageStore();
_context = new StoreContext();
+ _protocolSession = new TestMinaProtocolSession();
}
public void testNoRoute() throws AMQException
{
- AMQQueue queue = new AMQQueue(new AMQShortString("a*#b"), false, null, false, _vhost);
+ AMQQueue queue = AMQQueueFactory.createAMQQueueImpl(new AMQShortString("a*#b"), false, null, false, _vhost);
_exchange.registerQueue(new AMQShortString("a.*.#.b"), queue, null);
MessagePublishInfo info = new PublishInfo(new AMQShortString("a.b"));
- AMQMessage message = new AMQMessage(0L, info, null);
+ IncomingMessage message = new IncomingMessage(0L, info, null, _protocolSession);
- try
- {
- _exchange.route(message);
- fail("Message has no route and shouldn't be routed");
- }
- catch (NoRouteException nre)
- {
- //normal
- }
+ _exchange.route(message);
Assert.assertEquals(0, queue.getMessageCount());
}
public void testDirectMatch() throws AMQException
{
- AMQQueue queue = new AMQQueue(new AMQShortString("ab"), false, null, false, _vhost);
+ AMQQueue queue = AMQQueueFactory.createAMQQueueImpl(new AMQShortString("ab"), false, null, false, _vhost);
_exchange.registerQueue(new AMQShortString("a.b"), queue, null);
- AMQMessage message = createMessage("a.b");
+ IncomingMessage message = createMessage("a.b");
try
{
- _exchange.route(message);
- message.routingComplete(_store, _context, new MessageHandleFactory());
+ routeMessage(message);
}
catch (AMQException nre)
{
@@ -104,7 +95,7 @@
Assert.assertEquals(1, queue.getMessageCount());
- Assert.assertEquals("Wrong message recevied", message, queue.getMessagesOnTheQueue().get(0).getMessage());
+ Assert.assertEquals("Wrong message recevied", (Object) message.getMessageId(), queue.getMessagesOnTheQueue().get(0).getMessage().getMessageId());
queue.deleteMessageFromTop(_context);
Assert.assertEquals(0, queue.getMessageCount());
@@ -114,8 +105,7 @@
try
{
- _exchange.route(message);
- message.routingComplete(_store, _context, new MessageHandleFactory());
+ routeMessage(message);
fail("Message has no route and should fail to be routed");
}
catch (AMQException nre)
@@ -128,16 +118,15 @@
public void testStarMatch() throws AMQException
{
- AMQQueue queue = new AMQQueue(new AMQShortString("a*"), false, null, false, _vhost);
+ AMQQueue queue = AMQQueueFactory.createAMQQueueImpl(new AMQShortString("a*"), false, null, false, _vhost);
_exchange.registerQueue(new AMQShortString("a.*"), queue, null);
- AMQMessage message = createMessage("a.b");
+ IncomingMessage message = createMessage("a.b");
try
{
- _exchange.route(message);
- message.routingComplete(_store, _context, new MessageHandleFactory());
+ routeMessage(message);
}
catch (AMQException nre)
{
@@ -146,7 +135,7 @@
Assert.assertEquals(1, queue.getMessageCount());
- Assert.assertEquals("Wrong message recevied", message, queue.getMessagesOnTheQueue().get(0).getMessage());
+ Assert.assertEquals("Wrong message recevied", (Object) message.getMessageId(), queue.getMessagesOnTheQueue().get(0).getMessage().getMessageId());
queue.deleteMessageFromTop(_context);
Assert.assertEquals(0, queue.getMessageCount());
@@ -156,8 +145,7 @@
try
{
- _exchange.route(message);
- message.routingComplete(_store, _context, new MessageHandleFactory());
+ routeMessage(message);
}
catch (AMQException nre)
{
@@ -166,7 +154,7 @@
Assert.assertEquals(1, queue.getMessageCount());
- Assert.assertEquals("Wrong message recevied", message, queue.getMessagesOnTheQueue().get(0).getMessage());
+ Assert.assertEquals("Wrong message recevied", (Object) message.getMessageId(), queue.getMessagesOnTheQueue().get(0).getMessage().getMessageId());
queue.deleteMessageFromTop(_context);
Assert.assertEquals(0, queue.getMessageCount());
@@ -176,8 +164,7 @@
try
{
- _exchange.route(message);
- message.routingComplete(_store, _context, new MessageHandleFactory());
+ routeMessage(message);
fail("Message has no route and should fail to be routed");
}
catch (AMQException nre)
@@ -189,16 +176,15 @@
public void testHashMatch() throws AMQException
{
- AMQQueue queue = new AMQQueue(new AMQShortString("a#"), false, null, false, _vhost);
+ AMQQueue queue = AMQQueueFactory.createAMQQueueImpl(new AMQShortString("a#"), false, null, false, _vhost);
_exchange.registerQueue(new AMQShortString("a.#"), queue, null);
- AMQMessage message = createMessage("a.b.c");
+ IncomingMessage message = createMessage("a.b.c");
try
{
- _exchange.route(message);
- message.routingComplete(_store, _context, new MessageHandleFactory());
+ routeMessage(message);
}
catch (AMQException nre)
{
@@ -207,7 +193,7 @@
Assert.assertEquals(1, queue.getMessageCount());
- Assert.assertEquals("Wrong message recevied", message, queue.getMessagesOnTheQueue().get(0).getMessage());
+ Assert.assertEquals("Wrong message recevied", (Object) message.getMessageId(), queue.getMessagesOnTheQueue().get(0).getMessage().getMessageId());
queue.deleteMessageFromTop(_context);
Assert.assertEquals(0, queue.getMessageCount());
@@ -217,8 +203,7 @@
try
{
- _exchange.route(message);
- message.routingComplete(_store, _context, new MessageHandleFactory());
+ routeMessage(message);
}
catch (AMQException nre)
{
@@ -227,7 +212,7 @@
Assert.assertEquals(1, queue.getMessageCount());
- Assert.assertEquals("Wrong message recevied", message, queue.getMessagesOnTheQueue().get(0).getMessage());
+ Assert.assertEquals("Wrong message recevied", (Object) message.getMessageId(), queue.getMessagesOnTheQueue().get(0).getMessage().getMessageId());
queue.deleteMessageFromTop(_context);
Assert.assertEquals(0, queue.getMessageCount());
@@ -237,8 +222,7 @@
try
{
- _exchange.route(message);
- message.routingComplete(_store, _context, new MessageHandleFactory());
+ routeMessage(message);
}
catch (AMQException nre)
{
@@ -247,7 +231,7 @@
Assert.assertEquals(1, queue.getMessageCount());
- Assert.assertEquals("Wrong message recevied", message, queue.getMessagesOnTheQueue().get(0).getMessage());
+ Assert.assertEquals("Wrong message recevied", (Object) message.getMessageId(), queue.getMessagesOnTheQueue().get(0).getMessage().getMessageId());
queue.deleteMessageFromTop(_context);
Assert.assertEquals(0, queue.getMessageCount());
@@ -256,8 +240,7 @@
try
{
- _exchange.route(message);
- message.routingComplete(_store, _context, new MessageHandleFactory());
+ routeMessage(message);
}
catch (AMQException nre)
{
@@ -266,7 +249,7 @@
Assert.assertEquals(1, queue.getMessageCount());
- Assert.assertEquals("Wrong message recevied", message, queue.getMessagesOnTheQueue().get(0).getMessage());
+ Assert.assertEquals("Wrong message recevied", (Object) message.getMessageId(), queue.getMessagesOnTheQueue().get(0).getMessage().getMessageId());
queue.deleteMessageFromTop(_context);
Assert.assertEquals(0, queue.getMessageCount());
@@ -276,8 +259,7 @@
try
{
- _exchange.route(message);
- message.routingComplete(_store, _context, new MessageHandleFactory());
+ routeMessage(message);
fail("Message has no route and should fail to be routed");
}
catch (AMQException nre)
@@ -290,16 +272,15 @@
public void testMidHash() throws AMQException
{
- AMQQueue queue = new AMQQueue(new AMQShortString("a"), false, null, false, _vhost);
+ AMQQueue queue = AMQQueueFactory.createAMQQueueImpl(new AMQShortString("a"), false, null, false, _vhost);
_exchange.registerQueue(new AMQShortString("a.*.#.b"), queue, null);
- AMQMessage message = createMessage("a.c.d.b");
+ IncomingMessage message = createMessage("a.c.d.b");
try
{
- _exchange.route(message);
- message.routingComplete(_store, _context, new MessageHandleFactory());
+ routeMessage(message);
}
catch (AMQException nre)
{
@@ -308,7 +289,7 @@
Assert.assertEquals(1, queue.getMessageCount());
- Assert.assertEquals("Wrong message recevied", message, queue.getMessagesOnTheQueue().get(0).getMessage());
+ Assert.assertEquals("Wrong message recevied", (Object) message.getMessageId(), queue.getMessagesOnTheQueue().get(0).getMessage().getMessageId());
queue.deleteMessageFromTop(_context);
Assert.assertEquals(0, queue.getMessageCount());
@@ -317,8 +298,7 @@
try
{
- _exchange.route(message);
- message.routingComplete(_store, _context, new MessageHandleFactory());
+ routeMessage(message);
}
catch (AMQException nre)
{
@@ -327,7 +307,7 @@
Assert.assertEquals(1, queue.getMessageCount());
- Assert.assertEquals("Wrong message recevied", message, queue.getMessagesOnTheQueue().get(0).getMessage());
+ Assert.assertEquals("Wrong message recevied", (Object) message.getMessageId(), queue.getMessagesOnTheQueue().get(0).getMessage().getMessageId());
queue.deleteMessageFromTop(_context);
Assert.assertEquals(0, queue.getMessageCount());
@@ -336,16 +316,15 @@
public void testMatchafterHash() throws AMQException
{
- AMQQueue queue = new AMQQueue(new AMQShortString("a#"), false, null, false, _vhost);
+ AMQQueue queue = AMQQueueFactory.createAMQQueueImpl(new AMQShortString("a#"), false, null, false, _vhost);
_exchange.registerQueue(new AMQShortString("a.*.#.b.c"), queue, null);
- AMQMessage message = createMessage("a.c.b.b");
+ IncomingMessage message = createMessage("a.c.b.b");
try
{
- _exchange.route(message);
- message.routingComplete(_store, _context, new MessageHandleFactory());
+ routeMessage(message);
fail("Message has route and should not be routed");
}
catch (AMQException nre)
@@ -359,8 +338,7 @@
try
{
- _exchange.route(message);
- message.routingComplete(_store, _context, new MessageHandleFactory());
+ routeMessage(message);
}
catch (AMQException nre)
{
@@ -369,7 +347,7 @@
Assert.assertEquals(1, queue.getMessageCount());
- Assert.assertEquals("Wrong message recevied", message, queue.getMessagesOnTheQueue().get(0).getMessage());
+ Assert.assertEquals("Wrong message recevied", (Object) message.getMessageId(), queue.getMessagesOnTheQueue().get(0).getMessage().getMessageId());
queue.deleteMessageFromTop(_context);
Assert.assertEquals(0, queue.getMessageCount());
@@ -378,8 +356,7 @@
try
{
- _exchange.route(message);
- message.routingComplete(_store, _context, new MessageHandleFactory());
+ routeMessage(message);
fail("Message has route and should not be routed");
}
catch (AMQException nre)
@@ -392,8 +369,7 @@
try
{
- _exchange.route(message);
- message.routingComplete(_store, _context, new MessageHandleFactory());
+ routeMessage(message);
}
catch (AMQException nre)
{
@@ -403,7 +379,7 @@
Assert.assertEquals(1, queue.getMessageCount());
- Assert.assertEquals("Wrong message recevied", message, queue.getMessagesOnTheQueue().get(0).getMessage());
+ Assert.assertEquals("Wrong message recevied", (Object) message.getMessageId(), queue.getMessagesOnTheQueue().get(0).getMessage().getMessageId());
queue.deleteMessageFromTop(_context);
Assert.assertEquals(0, queue.getMessageCount());
@@ -413,16 +389,15 @@
public void testHashAfterHash() throws AMQException
{
- AMQQueue queue = new AMQQueue(new AMQShortString("a#"), false, null, false, _vhost);
+ AMQQueue queue = AMQQueueFactory.createAMQQueueImpl(new AMQShortString("a#"), false, null, false, _vhost);
_exchange.registerQueue(new AMQShortString("a.*.#.b.c.#.d"), queue, null);
- AMQMessage message = createMessage("a.c.b.b.c");
+ IncomingMessage message = createMessage("a.c.b.b.c");
try
{
- _exchange.route(message);
- message.routingComplete(_store, _context, new MessageHandleFactory());
+ routeMessage(message);
fail("Message has route and should not be routed");
}
catch (AMQException nre)
@@ -436,8 +411,7 @@
try
{
- _exchange.route(message);
- message.routingComplete(_store, _context, new MessageHandleFactory());
+ routeMessage(message);
}
catch (AMQException nre)
{
@@ -446,7 +420,7 @@
Assert.assertEquals(1, queue.getMessageCount());
- Assert.assertEquals("Wrong message recevied", message, queue.getMessagesOnTheQueue().get(0).getMessage());
+ Assert.assertEquals("Wrong message recevied", (Object) message.getMessageId(), queue.getMessagesOnTheQueue().get(0).getMessage().getMessageId());
queue.deleteMessageFromTop(_context);
Assert.assertEquals(0, queue.getMessageCount());
@@ -455,16 +429,15 @@
public void testHashHash() throws AMQException
{
- AMQQueue queue = new AMQQueue(new AMQShortString("a#"), false, null, false, _vhost);
+ AMQQueue queue = AMQQueueFactory.createAMQQueueImpl(new AMQShortString("a#"), false, null, false, _vhost);
_exchange.registerQueue(new AMQShortString("a.#.*.#.d"), queue, null);
- AMQMessage message = createMessage("a.c.b.b.c");
+ IncomingMessage message = createMessage("a.c.b.b.c");
try
{
- _exchange.route(message);
- message.routingComplete(_store, _context, new MessageHandleFactory());
+ routeMessage(message);
fail("Message has route and should not be routed");
}
catch (AMQException nre)
@@ -477,8 +450,7 @@
try
{
- _exchange.route(message);
- message.routingComplete(_store, _context, new MessageHandleFactory());
+ routeMessage(message);
}
catch (AMQException nre)
{
@@ -487,7 +459,7 @@
Assert.assertEquals(1, queue.getMessageCount());
- Assert.assertEquals("Wrong message recevied", message, queue.getMessagesOnTheQueue().get(0).getMessage());
+ Assert.assertEquals("Wrong message recevied", (Object) message.getMessageId(), queue.getMessagesOnTheQueue().get(0).getMessage().getMessageId());
queue.deleteMessageFromTop(_context);
Assert.assertEquals(0, queue.getMessageCount());
@@ -496,16 +468,15 @@
public void testSubMatchFails() throws AMQException
{
- AMQQueue queue = new AMQQueue(new AMQShortString("a"), false, null, false, _vhost);
+ AMQQueue queue = AMQQueueFactory.createAMQQueueImpl(new AMQShortString("a"), false, null, false, _vhost);
_exchange.registerQueue(new AMQShortString("a.b.c.d"), queue, null);
- AMQMessage message = createMessage("a.b.c");
+ IncomingMessage message = createMessage("a.b.c");
try
{
- _exchange.route(message);
- message.routingComplete(_store, _context, new MessageHandleFactory());
+ routeMessage(message);
fail("Message has route and should not be routed");
}
catch (AMQException nre)
@@ -516,18 +487,25 @@
}
+ private void routeMessage(final IncomingMessage message)
+ throws AMQException
+ {
+ _exchange.route(message);
+ message.routingComplete(_store, new MessageHandleFactory());
+ message.deliverToQueues();
+ }
+
public void testMoreRouting() throws AMQException
{
- AMQQueue queue = new AMQQueue(new AMQShortString("a"), false, null, false, _vhost);
+ AMQQueue queue = AMQQueueFactory.createAMQQueueImpl(new AMQShortString("a"), false, null, false, _vhost);
_exchange.registerQueue(new AMQShortString("a.b"), queue, null);
- AMQMessage message = createMessage("a.b.c");
+ IncomingMessage message = createMessage("a.b.c");
try
{
- _exchange.route(message);
- message.routingComplete(_store, _context, new MessageHandleFactory());
+ routeMessage(message);
fail("Message has route and should not be routed");
}
catch (AMQException nre)
@@ -540,16 +518,15 @@
public void testMoreQueue() throws AMQException
{
- AMQQueue queue = new AMQQueue(new AMQShortString("a"), false, null, false, _vhost);
+ AMQQueue queue = AMQQueueFactory.createAMQQueueImpl(new AMQShortString("a"), false, null, false, _vhost);
_exchange.registerQueue(new AMQShortString("a.b"), queue, null);
- AMQMessage message = createMessage("a");
+ IncomingMessage message = createMessage("a");
try
{
- _exchange.route(message);
- message.routingComplete(_store, _context, new MessageHandleFactory());
+ routeMessage(message);
fail("Message has route and should not be routed");
}
catch (AMQException nre)
@@ -560,16 +537,17 @@
}
- private AMQMessage createMessage(String s) throws AMQException
+ private IncomingMessage createMessage(String s) throws AMQException
{
MessagePublishInfo info = new PublishInfo(new AMQShortString(s));
TransactionalContext trancontext = new NonTransactionalContext(_store, _context, null,
- new LinkedList<RequiredDeliveryException>(),
- new HashSet<Long>());
+ new LinkedList<RequiredDeliveryException>()
+ );
+
+ IncomingMessage message = new IncomingMessage(0L, info, trancontext,_protocolSession);
+ message.setContentHeaderBody( new ContentHeaderBody());
- AMQMessage message = new AMQMessage(0L, info, trancontext);
- message.setContentHeaderBody(new ContentHeaderBody());
return message;
}
Modified: incubator/qpid/branches/broker-queue-refactor/java/broker/src/test/java/org/apache/qpid/server/exchange/ExchangeMBeanTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/test/java/org/apache/qpid/server/exchange/ExchangeMBeanTest.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/test/java/org/apache/qpid/server/exchange/ExchangeMBeanTest.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/test/java/org/apache/qpid/server/exchange/ExchangeMBeanTest.java Mon Apr 21 07:03:35 2008
@@ -21,8 +21,9 @@
package org.apache.qpid.server.exchange;
import junit.framework.TestCase;
-import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.server.queue.QueueRegistry;
+import org.apache.qpid.server.queue.AMQQueue;
+import org.apache.qpid.server.queue.AMQQueueFactory;
import org.apache.qpid.server.registry.ApplicationRegistry;
import org.apache.qpid.server.registry.IApplicationRegistry;
import org.apache.qpid.server.management.ManagedObject;
@@ -30,7 +31,6 @@
import org.apache.qpid.exchange.ExchangeDefaults;
import org.apache.qpid.framing.AMQShortString;
-import javax.management.openmbean.CompositeData;
import javax.management.openmbean.TabularData;
import java.util.ArrayList;
@@ -50,7 +50,7 @@
public void testDirectExchangeMBean() throws Exception
{
- DestNameExchange exchange = new DestNameExchange();
+ DirectExchange exchange = new DirectExchange();
exchange.initialise(_virtualHost, ExchangeDefaults.DIRECT_EXCHANGE_NAME, false, 0, true);
ManagedObject managedObj = exchange.getManagedObject();
ManagedExchange mbean = (ManagedExchange)managedObj;
@@ -77,7 +77,7 @@
public void testTopicExchangeMBean() throws Exception
{
- DestWildExchange exchange = new DestWildExchange();
+ TopicExchange exchange = new TopicExchange();
exchange.initialise(_virtualHost,ExchangeDefaults.TOPIC_EXCHANGE_NAME, false, 0, true);
ManagedObject managedObj = exchange.getManagedObject();
ManagedExchange mbean = (ManagedExchange)managedObj;
@@ -132,7 +132,7 @@
IApplicationRegistry applicationRegistry = ApplicationRegistry.getInstance();
_virtualHost = applicationRegistry.getVirtualHostRegistry().getVirtualHost("test");
_queueRegistry = _virtualHost.getQueueRegistry();
- _queue = new AMQQueue(new AMQShortString("testQueue"), false, new AMQShortString("ExchangeMBeanTest"), false, _virtualHost);
+ _queue = AMQQueueFactory.createAMQQueueImpl(new AMQShortString("testQueue"), false, new AMQShortString("ExchangeMBeanTest"), false, _virtualHost);
_queueRegistry.registerQueue(_queue);
}
}
Modified: incubator/qpid/branches/broker-queue-refactor/java/broker/src/test/java/org/apache/qpid/server/protocol/TestMinaProtocolSession.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/test/java/org/apache/qpid/server/protocol/TestMinaProtocolSession.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/test/java/org/apache/qpid/server/protocol/TestMinaProtocolSession.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/test/java/org/apache/qpid/server/protocol/TestMinaProtocolSession.java Mon Apr 21 07:03:35 2008
@@ -30,9 +30,11 @@
{
public TestMinaProtocolSession() throws AMQException
{
+
super(new TestIoSession(),
ApplicationRegistry.getInstance().getVirtualHostRegistry(),
new AMQCodecFactory(true));
+
}
public ProtocolOutputConverter getProtocolOutputConverter()
Modified: incubator/qpid/branches/broker-queue-refactor/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueAlertTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueAlertTest.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueAlertTest.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueAlertTest.java Mon Apr 21 07:03:35 2008
@@ -32,19 +32,22 @@
import org.apache.qpid.server.txn.NonTransactionalContext;
import org.apache.qpid.server.RequiredDeliveryException;
import org.apache.qpid.server.AMQChannel;
+import org.apache.qpid.server.subscription.Subscription;
+import org.apache.qpid.server.subscription.SubscriptionFactoryImpl;
import org.apache.qpid.server.protocol.TestMinaProtocolSession;
import org.apache.qpid.server.protocol.AMQMinaProtocolSession;
import org.apache.qpid.framing.ContentHeaderBody;
import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.framing.abstraction.MessagePublishInfo;
+import org.apache.qpid.framing.abstraction.ContentChunk;
+import org.apache.mina.common.ByteBuffer;
import javax.management.Notification;
import java.util.LinkedList;
-import java.util.HashSet;
/** This class tests all the alerts an AMQQueue can throw based on threshold values of different parameters */
public class AMQQueueAlertTest extends TestCase
-{
+{
private final static long MAX_MESSAGE_COUNT = 50;
private final static long MAX_MESSAGE_AGE = 250; // 0.25 sec
private final static long MAX_MESSAGE_SIZE = 2000; // 2 KB
@@ -52,13 +55,14 @@
private AMQQueue _queue;
private AMQQueueMBean _queueMBean;
private VirtualHost _virtualHost;
- private AMQMinaProtocolSession protocolSession = null;
+ private AMQMinaProtocolSession _protocolSession;
private MessageStore _messageStore = new MemoryMessageStore();
private StoreContext _storeContext = new StoreContext();
private TransactionalContext _transactionalContext = new NonTransactionalContext(_messageStore, _storeContext,
null,
- new LinkedList<RequiredDeliveryException>(),
- new HashSet<Long>());
+ new LinkedList<RequiredDeliveryException>()
+ );
+ private static final SubscriptionFactoryImpl SUBSCRIPTION_FACTORY = SubscriptionFactoryImpl.INSTANCE;
/**
* Tests if the alert gets thrown when message count increases the threshold limit
@@ -67,7 +71,7 @@
*/
public void testMessageCountAlert() throws Exception
{
- _queue = new AMQQueue(new AMQShortString("testQueue1"), false, new AMQShortString("AMQueueAlertTest"),
+ _queue = AMQQueueFactory.createAMQQueueImpl(new AMQShortString("testQueue1"), false, new AMQShortString("AMQueueAlertTest"),
false, _virtualHost);
_queueMBean = (AMQQueueMBean) _queue.getManagedObject();
@@ -90,7 +94,7 @@
*/
public void testMessageSizeAlert() throws Exception
{
- _queue = new AMQQueue(new AMQShortString("testQueue2"), false, new AMQShortString("AMQueueAlertTest"),
+ _queue = AMQQueueFactory.createAMQQueueImpl(new AMQShortString("testQueue2"), false, new AMQShortString("AMQueueAlertTest"),
false, _virtualHost);
_queueMBean = (AMQQueueMBean) _queue.getManagedObject();
_queueMBean.setMaximumMessageCount(MAX_MESSAGE_COUNT);
@@ -115,7 +119,7 @@
*/
public void testQueueDepthAlertNoSubscriber() throws Exception
{
- _queue = new AMQQueue(new AMQShortString("testQueue3"), false, new AMQShortString("AMQueueAlertTest"),
+ _queue = AMQQueueFactory.createAMQQueueImpl(new AMQShortString("testQueue3"), false, new AMQShortString("AMQueueAlertTest"),
false, _virtualHost);
_queueMBean = (AMQQueueMBean) _queue.getManagedObject();
_queueMBean.setMaximumMessageCount(MAX_MESSAGE_COUNT);
@@ -143,7 +147,7 @@
*/
public void testMessageAgeAlert() throws Exception
{
- _queue = new AMQQueue(new AMQShortString("testQueue4"), false, new AMQShortString("AMQueueAlertTest"),
+ _queue = AMQQueueFactory.createAMQQueueImpl(new AMQShortString("testQueue4"), false, new AMQShortString("AMQueueAlertTest"),
false, _virtualHost);
_queueMBean = (AMQQueueMBean) _queue.getManagedObject();
_queueMBean.setMaximumMessageCount(MAX_MESSAGE_COUNT);
@@ -168,18 +172,23 @@
This test sends some messages to the queue with subscribers needing message to be acknowledged.
The messages will not be acknowledged and will be required twice. Why we are checking this is because
the bug reported said that the queueDepth keeps increasing when messages are requeued.
+ // TODO - queue depth now includes unacknowledged messages so does not go down when messages are delivered
+
The QueueDepth should decrease when messages are delivered from the queue (QPID-408)
*/
public void testQueueDepthAlertWithSubscribers() throws Exception
{
- protocolSession = new TestMinaProtocolSession();
- AMQChannel channel = new AMQChannel(protocolSession, 2, _messageStore);
- protocolSession.addChannel(channel);
+ _protocolSession = new TestMinaProtocolSession();
+ AMQChannel channel = new AMQChannel(_protocolSession, 2, _messageStore);
+ _protocolSession.addChannel(channel);
// Create queue
_queue = getNewQueue();
- _queue.registerProtocolSession(protocolSession, channel.getChannelId(),
- new AMQShortString("consumer_tag"), true, null, false, false);
+ Subscription subscription =
+ SUBSCRIPTION_FACTORY.createSubscription(channel.getChannelId(), _protocolSession, new AMQShortString("consumer_tag"), true, null, false, channel.getCreditManager());
+
+ _queue.registerSubscription(
+ subscription, false);
_queueMBean = (AMQQueueMBean) _queue.getManagedObject();
_queueMBean.setMaximumMessageCount(9999l); // Set a high value, because this is not being tested
@@ -192,13 +201,13 @@
// Check queueDepth. There should be no messages on the queue and as the subscriber is listening
// so there should be no Queue_Deoth alert raised
- assertEquals(new Long(0), new Long(_queueMBean.getQueueDepth()));
+ assertEquals(new Long(totalSize), new Long(_queueMBean.getQueueDepth()));
Notification lastNotification = _queueMBean.getLastNotification();
- assertNull(lastNotification);
+// assertNull(lastNotification);
// Kill the subscriber and check for the queue depth values.
// Messages are unacknowledged, so those should get requeued. All messages should be on the Queue
- _queue.unregisterProtocolSession(protocolSession, channel.getChannelId(), new AMQShortString("consumer_tag"));
+ _queue.unregisterSubscription(subscription);
channel.requeue();
assertEquals(new Long(totalSize), new Long(_queueMBean.getQueueDepth()));
@@ -210,29 +219,32 @@
// Connect a consumer again and check QueueDepth values. The queue should get emptied.
// Messages will get delivered but still are unacknowledged.
- _queue.registerProtocolSession(protocolSession, channel.getChannelId(),
- new AMQShortString("consumer_tag"), true, null, false, false);
- _queue.deliverAsync();
- while (_queue.getMessageCount() != 0)
+ Subscription subscription2 =
+ SUBSCRIPTION_FACTORY.createSubscription(channel.getChannelId(), _protocolSession, new AMQShortString("consumer_tag"), true, null, false, channel.getCreditManager());
+
+ _queue.registerSubscription(
+ subscription2, false);
+
+ while (_queue.getUndeliveredMessageCount()!= 0)
{
Thread.sleep(100);
}
- assertEquals(new Long(0), new Long(_queueMBean.getQueueDepth()));
+// assertEquals(new Long(0), new Long(_queueMBean.getQueueDepth()));
// Kill the subscriber again. Now those messages should get requeued again. Check if the queue depth
// value is correct.
- _queue.unregisterProtocolSession(protocolSession, channel.getChannelId(), new AMQShortString("consumer_tag"));
+ _queue.unregisterSubscription(subscription2);
channel.requeue();
assertEquals(new Long(totalSize), new Long(_queueMBean.getQueueDepth()));
- protocolSession.closeSession();
+ _protocolSession.closeSession();
// Check the clear queue
_queueMBean.clearQueue();
assertEquals(new Long(0), new Long(_queueMBean.getQueueDepth()));
}
- protected AMQMessage message(final boolean immediate, long size) throws AMQException
+ protected IncomingMessage message(final boolean immediate, long size) throws AMQException
{
MessagePublishInfo publish = new MessagePublishInfo()
{
@@ -265,9 +277,9 @@
ContentHeaderBody contentHeaderBody = new ContentHeaderBody();
contentHeaderBody.bodySize = size; // in bytes
- AMQMessage message = new AMQMessage(_messageStore.getNewMessageId(), publish, _transactionalContext);
+ IncomingMessage message = new IncomingMessage(_messageStore.getNewMessageId(), publish, _transactionalContext, _protocolSession);
message.setContentHeaderBody(contentHeaderBody);
- message.setPublisher(protocolSession);
+
return message;
}
@@ -277,27 +289,49 @@
super.setUp();
IApplicationRegistry applicationRegistry = ApplicationRegistry.getInstance();
_virtualHost = applicationRegistry.getVirtualHostRegistry().getVirtualHost("test");
+ _protocolSession = new TestMinaProtocolSession();
+
}
- private void sendMessages(long messageCount, long size) throws AMQException
+ private void sendMessages(long messageCount, final long size) throws AMQException
{
- AMQMessage[] messages = new AMQMessage[(int) messageCount];
+ IncomingMessage[] messages = new IncomingMessage[(int) messageCount];
for (int i = 0; i < messages.length; i++)
{
messages[i] = message(false, size);
messages[i].enqueue(_queue);
- messages[i].routingComplete(_messageStore, _storeContext, new MessageHandleFactory());
+ messages[i].routingComplete(_messageStore, new MessageHandleFactory());
+
}
for (int i = 0; i < messageCount; i++)
{
- _queue.process(_storeContext, new QueueEntry(_queue,messages[i]), false);
+ messages[i].addContentBodyFrame(new ContentChunk(){
+
+ ByteBuffer _data = ByteBuffer.allocate((int)size);
+
+ public int getSize()
+ {
+ return (int) size;
+ }
+
+ public ByteBuffer getData()
+ {
+ return _data;
+ }
+
+ public void reduceToFit()
+ {
+
+ }
+ });
+ messages[i].deliverToQueues();
}
}
private AMQQueue getNewQueue() throws AMQException
{
- return new AMQQueue(new AMQShortString("testQueue" + Math.random()),
+ return AMQQueueFactory.createAMQQueueImpl(new AMQShortString("testQueue" + Math.random()),
false,
new AMQShortString("AMQueueAlertTest"),
false,
Modified: incubator/qpid/branches/broker-queue-refactor/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueMBeanTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueMBeanTest.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueMBeanTest.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueMBeanTest.java Mon Apr 21 07:03:35 2008
@@ -27,8 +27,12 @@
import org.apache.qpid.framing.BasicContentHeaderProperties;
import org.apache.qpid.framing.ContentBody;
import org.apache.qpid.framing.abstraction.MessagePublishInfo;
+import org.apache.qpid.framing.abstraction.ContentChunk;
import org.apache.qpid.server.AMQChannel;
import org.apache.qpid.server.RequiredDeliveryException;
+import org.apache.qpid.server.subscription.Subscription;
+import org.apache.qpid.server.subscription.SubscriptionFactory;
+import org.apache.qpid.server.subscription.SubscriptionFactoryImpl;
import org.apache.qpid.server.protocol.TestMinaProtocolSession;
import org.apache.qpid.server.protocol.AMQProtocolSession;
import org.apache.qpid.server.virtualhost.VirtualHost;
@@ -44,7 +48,6 @@
import javax.management.JMException;
import java.util.LinkedList;
-import java.util.HashSet;
/**
* Test class to test AMQQueueMBean attribtues and operations
@@ -59,6 +62,7 @@
private TransactionalContext _transactionalContext;
private VirtualHost _virtualHost;
private AMQProtocolSession _protocolSession;
+ private static final SubscriptionFactoryImpl SUBSCRIPTION_FACTORY = SubscriptionFactoryImpl.INSTANCE;
public void testMessageCountTransient() throws Exception
{
@@ -74,7 +78,7 @@
assertTrue(_queueMBean.getReceivedMessageCount() == messageCount);
_queueMBean.clearQueue();
- assertTrue(_queueMBean.getMessageCount() == 0);
+ assertEquals(0,(int)_queueMBean.getMessageCount());
assertTrue(_queueMBean.getReceivedMessageCount() == messageCount);
//Ensure that the data has been removed from the Store
@@ -117,8 +121,8 @@
public void testConsumerCount() throws AMQException
{
- SubscriptionManager mgr = _queue.getSubscribers();
- assertFalse(mgr.hasActiveSubscribers());
+
+ assertTrue(_queue.getActiveConsumerCount() == 0);
assertTrue(_queueMBean.getActiveConsumerCount() == 0);
@@ -126,18 +130,21 @@
AMQChannel channel = new AMQChannel(protocolSession, 1, _messageStore);
protocolSession.addChannel(channel);
- _queue.registerProtocolSession(protocolSession, 1, new AMQShortString("test"), false, null, false, false);
- assertTrue(_queueMBean.getActiveConsumerCount() == 1);
+ Subscription subscription =
+ SUBSCRIPTION_FACTORY.createSubscription(channel.getChannelId(), protocolSession, new AMQShortString("test"), false, null, false, channel.getCreditManager());
+
+ _queue.registerSubscription(subscription, false);
+ assertEquals(1,(int)_queueMBean.getActiveConsumerCount());
+
- SubscriptionSet _subscribers = (SubscriptionSet) mgr;
- SubscriptionFactory subscriptionFactory = new SubscriptionImpl.Factory();
+ SubscriptionFactory subscriptionFactory = SUBSCRIPTION_FACTORY;
Subscription s1 = subscriptionFactory.createSubscription(channel.getChannelId(),
protocolSession,
new AMQShortString("S1"),
false,
null,
true,
- _queue);
+ channel.getCreditManager());
Subscription s2 = subscriptionFactory.createSubscription(channel.getChannelId(),
protocolSession,
@@ -145,14 +152,14 @@
false,
null,
true,
- _queue);
- _subscribers.addSubscriber(s1);
- _subscribers.addSubscriber(s2);
+ channel.getCreditManager());
+ _queue.registerSubscription(s1,false);
+ _queue.registerSubscription(s2,false);
assertTrue(_queueMBean.getActiveConsumerCount() == 3);
assertTrue(_queueMBean.getConsumerCount() == 3);
s1.close();
- assertTrue(_queueMBean.getActiveConsumerCount() == 2);
+ assertEquals(2, (int) _queueMBean.getActiveConsumerCount());
assertTrue(_queueMBean.getConsumerCount() == 3);
}
@@ -205,13 +212,34 @@
}
- AMQMessage msg = message(false, false);
+ IncomingMessage msg = message(false, false);
long id = msg.getMessageId();
_queue.clearQueue(_storeContext);
msg.enqueue(_queue);
- msg.routingComplete(_messageStore, _storeContext, new MessageHandleFactory());
- _queue.process(_storeContext, new QueueEntry(_queue, msg), false);
+ msg.routingComplete(_messageStore, new MessageHandleFactory());
+
+ msg.addContentBodyFrame(new ContentChunk()
+ {
+ ByteBuffer _data = ByteBuffer.allocate((int)MESSAGE_SIZE);
+
+ public int getSize()
+ {
+ return (int) MESSAGE_SIZE;
+ }
+
+ public ByteBuffer getData()
+ {
+ return _data;
+ }
+
+ public void reduceToFit()
+ {
+
+ }
+ });
+ msg.deliverToQueues();
+// _queue.process(_storeContext, new QueueEntry(_queue, msg), false);
_queueMBean.viewMessageContent(id);
try
{
@@ -224,7 +252,7 @@
}
}
- private AMQMessage message(final boolean immediate, boolean persistent) throws AMQException
+ private IncomingMessage message(final boolean immediate, boolean persistent) throws AMQException
{
MessagePublishInfo publish = new MessagePublishInfo()
{
@@ -259,7 +287,10 @@
contentHeaderBody.bodySize = MESSAGE_SIZE; // in bytes
contentHeaderBody.properties = new BasicContentHeaderProperties();
((BasicContentHeaderProperties) contentHeaderBody.properties).setDeliveryMode((byte) (persistent ? 2 : 1));
- return new AMQMessage(_messageStore.getNewMessageId(), publish, _transactionalContext, contentHeaderBody);
+ IncomingMessage msg = new IncomingMessage(_messageStore.getNewMessageId(), publish, _transactionalContext, _protocolSession);
+ msg.setContentHeaderBody(contentHeaderBody);
+ return msg;
+
}
@Override
@@ -272,10 +303,10 @@
_transactionalContext = new NonTransactionalContext(_messageStore, _storeContext,
null,
- new LinkedList<RequiredDeliveryException>(),
- new HashSet<Long>());
+ new LinkedList<RequiredDeliveryException>()
+ );
- _queue = new AMQQueue(new AMQShortString("testQueue"), false, new AMQShortString("AMQueueMBeanTest"), false, _virtualHost);
+ _queue = AMQQueueFactory.createAMQQueueImpl(new AMQShortString("testQueue"), false, new AMQShortString("AMQueueMBeanTest"), false, _virtualHost);
_queueMBean = new AMQQueueMBean(_queue);
_protocolSession = new TestMinaProtocolSession();
@@ -285,19 +316,20 @@
{
for (int i = 0; i < messageCount; i++)
{
- AMQMessage currentMessage = message(false, persistent);
+ IncomingMessage currentMessage = message(false, persistent);
currentMessage.enqueue(_queue);
// route header
- currentMessage.routingComplete(_messageStore, _storeContext, new MessageHandleFactory());
+ currentMessage.routingComplete(_messageStore, new MessageHandleFactory());
// Add the body so we have somthing to test later
- currentMessage.addContentBodyFrame(_storeContext,
- _protocolSession.getMethodRegistry()
+ currentMessage.addContentBodyFrame(
+ _protocolSession.getMethodRegistry()
.getProtocolVersionMethodConverter()
.convertToContentChunk(
new ContentBody(ByteBuffer.allocate((int) MESSAGE_SIZE),
MESSAGE_SIZE)));
+ currentMessage.deliverToQueues();
}
Modified: incubator/qpid/branches/broker-queue-refactor/java/client/src/main/java/org/apache/qpid/client/AMQSession.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/client/src/main/java/org/apache/qpid/client/AMQSession.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/client/src/main/java/org/apache/qpid/client/AMQSession.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/client/src/main/java/org/apache/qpid/client/AMQSession.java Mon Apr 21 07:03:35 2008
@@ -182,6 +182,13 @@
_fastAccessConsumers[i] = null;
}
}
+
+
+
+ public String toString()
+ {
+ return "{ Fast: " + Arrays.asList(_fastAccessConsumers) + " ; Slow: " + _slowAccessConsumers + "}";
+ }
}
@@ -299,9 +306,6 @@
*/
private final IdToConsumerMap _consumers = new IdToConsumerMap();
- //Map<AMQShortString, BasicMessageConsumer> _consumers =
- //new ConcurrentHashMap<AMQShortString, BasicMessageConsumer>();
-
/**
* Contains a list of consumers which have been removed but which might still have
* messages to acknowledge, eg in client ack or transacted modes
@@ -1419,7 +1423,9 @@
if (message.isDeliverMessage())
{
_highestDeliveryTag.set(message.getDeliverBody().getDeliveryTag());
+
_queue.add(message);
+
}
else
{
Modified: incubator/qpid/branches/broker-queue-refactor/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java Mon Apr 21 07:03:35 2008
@@ -809,9 +809,9 @@
/** Acknowledge up to last message delivered (if any). Used when commiting. */
void acknowledgeDelivered()
{
- while (!_receivedDeliveryTags.isEmpty())
+ while (!_receivedDeliveryTags.isEmpty())
{
- _session.acknowledgeMessage(_receivedDeliveryTags.poll(), false);
+ _session.acknowledgeMessage(_receivedDeliveryTags.poll(), false);
}
}
@@ -1017,7 +1017,7 @@
{
_logger.warn("Queue was not empty after rejecting all messages Remaining:" + _synchronousQueue.size());
rollback();
- }
+ }
clearReceiveQueue();
}
Modified: incubator/qpid/branches/broker-queue-refactor/java/client/src/main/java/org/apache/qpid/client/handler/BasicDeliverMethodHandler.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/client/src/main/java/org/apache/qpid/client/handler/BasicDeliverMethodHandler.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/client/src/main/java/org/apache/qpid/client/handler/BasicDeliverMethodHandler.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/client/src/main/java/org/apache/qpid/client/handler/BasicDeliverMethodHandler.java Mon Apr 21 07:03:35 2008
@@ -46,7 +46,7 @@
{
final AMQProtocolSession session = stateManager.getProtocolSession();
final UnprocessedMessage msg = new UnprocessedMessage.UnprocessedDeliverMessage(body);
- _logger.debug("New JmsDeliver method received");
+ _logger.debug("New JmsDeliver method received");
session.unprocessedMessageReceived(channelId, msg);
}
}
Modified: incubator/qpid/branches/broker-queue-refactor/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolSession.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolSession.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolSession.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolSession.java Mon Apr 21 07:03:35 2008
@@ -297,15 +297,8 @@
throw new AMQException("Error: received content body without having received a ContentHeader frame first");
}
- /*try
- {*/
+
msg.receiveBody(contentBody);
- /*}
- catch (UnexpectedBodyReceivedException e)
- {
- _channelId2UnprocessedMsgMap.remove(channelId);
- throw e;
- }*/
if (msg.isAllBodyDataReceived())
{
@@ -324,9 +317,13 @@
* @param channelId the channel id the message should be delivered to
* @param msg the message
*/
- private void deliverMessageToAMQSession(int channelId, UnprocessedMessage msg)
+ private void deliverMessageToAMQSession(int channelId, UnprocessedMessage msg) throws AMQException
{
AMQSession session = getSession(channelId);
+ if(session == null)
+ {
+ throw new AMQException("Error: received message on non-existant channel:" + channelId);
+ }
session.messageReceived(msg);
if((channelId & FAST_CHANNEL_ACCESS_MASK) == 0)
{
Modified: incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/client/AMQQueueDeferredOrderingTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/client/AMQQueueDeferredOrderingTest.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/client/AMQQueueDeferredOrderingTest.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/client/AMQQueueDeferredOrderingTest.java Mon Apr 21 07:03:35 2008
@@ -126,7 +126,7 @@
_logger.info("Consuming messages");
for (int i = 0; i < NUM_MESSAGES; i++)
{
- Message msg = consumer.receive(3000);
+ Message msg = consumer.receive(90000);
assertNotNull("Message should not be null", msg);
assertTrue("Message should be a text message", msg instanceof TextMessage);
assertEquals("Message content does not match expected", Integer.toString(i), ((TextMessage) msg).getText());
Modified: incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/client/MessageListenerMultiConsumerTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/client/MessageListenerMultiConsumerTest.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/client/MessageListenerMultiConsumerTest.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/client/MessageListenerMultiConsumerTest.java Mon Apr 21 07:03:35 2008
@@ -128,7 +128,7 @@
{
int msg = 0;
int MAX_LOOPS = MSG_COUNT * 2;
- for (int loops = 0; (msg < MSG_COUNT) || (loops < MAX_LOOPS); loops++)
+ for (int loops = 0; (msg < MSG_COUNT) && (loops < MAX_LOOPS); loops++)
{
if (_consumer1.receive(100) != null)
@@ -220,12 +220,18 @@
for (int msg = 0; msg < (MSG_COUNT / 2); msg++)
{
- assertTrue(_consumer1.receive() != null);
+
+ final Message message = _consumer1.receive(100000);
+ if(message == null)
+ {
+ System.out.println("!!!!!!!! " + msg);
+ }
+ assertTrue(message != null);
}
for (int msg = 0; msg < (MSG_COUNT / 2); msg++)
{
- assertTrue(consumer2.receive() != null);
+ assertTrue(consumer2.receive(10000) != null);
}
}
else
@@ -235,12 +241,12 @@
for (int msg = 0; msg < (MSG_COUNT / 2); msg++)
{
- assertTrue(_consumer1.receive() != null);
+ assertTrue(_consumer1.receive(10000) != null);
}
for (int msg = 0; msg < (MSG_COUNT / 2); msg++)
{
- assertTrue(_consumer2.receive() != null);
+ assertTrue(_consumer2.receive(10000) != null);
}
}
}
Modified: incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/client/ResetMessageListenerTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/client/ResetMessageListenerTest.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/client/ResetMessageListenerTest.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/client/ResetMessageListenerTest.java Mon Apr 21 07:03:35 2008
@@ -116,6 +116,7 @@
{
_producer.send(_producerSession.createTextMessage("Message " + msg));
}
+// Thread.sleep(120000);
}
@@ -247,6 +248,14 @@
{
_producer.send(_producerSession.createTextMessage("Message " + msg));
}
+// try
+// {
+// Thread.sleep(120000);
+// }
+// catch (InterruptedException e)
+// {
+// e.printStackTrace(); //To change body of catch statement use File | Settings | File Templates.
+// }
}
catch (JMSException e)
{
@@ -257,7 +266,7 @@
try
{
- _allSecondMessagesSent.await(5000, TimeUnit.MILLISECONDS);
+ _allSecondMessagesSent.await(500000, TimeUnit.MILLISECONDS);
}
catch (InterruptedException e)
{
Modified: incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/test/unit/ack/RecoverTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/test/unit/ack/RecoverTest.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/test/unit/ack/RecoverTest.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/test/unit/ack/RecoverTest.java Mon Apr 21 07:03:35 2008
@@ -115,6 +115,7 @@
consumerSession.recover();
tm = (TextMessage) consumer.receiveNoWait();
+
assertNull(tm);
_logger.info("No messages redelivered as is expected");
Modified: incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/test/unit/basic/SessionStartTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/test/unit/basic/SessionStartTest.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/test/unit/basic/SessionStartTest.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/test/unit/basic/SessionStartTest.java Mon Apr 21 07:03:35 2008
@@ -28,6 +28,9 @@
import org.apache.qpid.client.AMQSession;
import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.testutil.VMBrokerSetup;
+import org.apache.qpid.AMQException;
+import org.apache.log4j.BasicConfigurator;
+import org.apache.log4j.Level;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -49,6 +52,7 @@
protected void setUp() throws Exception
{
super.setUp();
+
init(new AMQConnection(_connectionString, "guest", "guest", randomize("Client"), "test"));
}
Modified: incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/test/unit/client/channelclose/ChannelCloseOkTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/test/unit/client/channelclose/ChannelCloseOkTest.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/test/unit/client/channelclose/ChannelCloseOkTest.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/test/unit/client/channelclose/ChannelCloseOkTest.java Mon Apr 21 07:03:35 2008
@@ -69,6 +69,7 @@
private static final Logger _log = LoggerFactory.getLogger(ChannelCloseOkTest.class);
public String _connectionString = "vm://:1";
+ private static final int NUM_MESSAGES = 300;
protected void setUp() throws Exception
{
@@ -170,19 +171,18 @@
// Ensure both sessions are still ok.
// Send a bunch of messages as this give time for the sessions to be erroneously closed.
- final int num = 300;
- for (int i = 0; i < num; ++i)
+ for (int i = 0; i < NUM_MESSAGES; ++i)
{
send(_session1, _destination1, "" + i);
send(_session2, _destination2, "" + i);
}
- waitFor(_received1, num + 1);
- waitFor(_received2, num + 1);
+ waitFor(_received1, NUM_MESSAGES + 1);
+ waitFor(_received2, NUM_MESSAGES + 1);
// Note that the third message is never received as it is sent to an incorrect destination.
- assertEquals(num + 1, _received1.size());
- assertEquals(num + 1, _received2.size());
+ assertEquals(NUM_MESSAGES + 1, _received1.size());
+ assertEquals(NUM_MESSAGES + 1, _received2.size());
}
private void sendAndWait(Session session, Destination destination, String message, List<Message> received, int count)
@@ -199,15 +199,17 @@
producer1.send(session.createTextMessage(message));
}
- private void waitFor(List<Message> received, int count) throws InterruptedException
+ private void waitFor(List<Message> received, final int count) throws InterruptedException
{
+ int lastSeen = -1;
synchronized (received)
{
- while (received.size() < count)
+ while ((lastSeen != received.size()) && (lastSeen = received.size()) < count)
{
+
try
{
- received.wait();
+ received.wait(2000L);
}
catch (InterruptedException e)
{
@@ -215,6 +217,10 @@
throw e;
}
}
+ }
+ if(received.size() < count)
+ {
+ throw new RuntimeException("Expected: " + count + " got: " + received.size());
}
}
Modified: incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/test/unit/client/forwardall/Client.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/test/unit/client/forwardall/Client.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/test/unit/client/forwardall/Client.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/test/unit/client/forwardall/Client.java Mon Apr 21 07:03:35 2008
@@ -75,7 +75,9 @@
public synchronized void onMessage(Message response)
{
+
_logger.info("Received " + (++_count) + " of " + _expected + " responses.");
+
if (_count == _expected)
{
@@ -89,10 +91,10 @@
if (_count < _expected)
{
- wait(10000L);
+ wait(1000L);
}
- if (_count < _expected)
+ if (_count != _expected)
{
throw new Exception("Didn't receive all messages... got " + _count + " expected " + _expected);
}
Modified: incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/test/unit/client/forwardall/CombinedTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/test/unit/client/forwardall/CombinedTest.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/test/unit/client/forwardall/CombinedTest.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/test/unit/client/forwardall/CombinedTest.java Mon Apr 21 07:03:35 2008
@@ -22,9 +22,13 @@
import junit.framework.TestCase;
import org.apache.qpid.testutil.VMBrokerSetup;
+import org.apache.qpid.server.queue.SimpleAMQQueue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.Set;
+import java.util.HashSet;
+
/**
* Runs the Service's and Client parts of the test in the same process
* as the broker
@@ -34,6 +38,7 @@
private static final Logger _logger = LoggerFactory.getLogger(CombinedTest.class);
private int run = 0;
+
protected void setUp() throws Exception
{
super.setUp();
@@ -47,16 +52,16 @@
public void testForwardAll() throws Exception
{
- while (run < 10)
+ while (run < 100)
{
int services = 2;
ServiceCreator.start("vm://:1", services);
-
+ Thread.sleep(100);
_logger.info("Starting " + ++run + " client...");
new Client("vm://:1", services).shutdownWhenComplete();
-
+ ServiceCreator.closeAll();
_logger.info("Completed " + run + " successfully!");
}
}
Modified: incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/test/unit/client/forwardall/Service.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/test/unit/client/forwardall/Service.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/test/unit/client/forwardall/Service.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/test/unit/client/forwardall/Service.java Mon Apr 21 07:03:35 2008
@@ -37,14 +37,16 @@
{
private final AMQConnection _connection;
private final AMQSession _session;
+ private final int _id;
- Service(String broker) throws Exception
+ Service(String broker, int id) throws Exception
{
- this(connect(broker));
+ this(connect(broker), id);
}
- Service(AMQConnection connection) throws Exception
+ Service(AMQConnection connection, int id) throws Exception
{
+ _id = id;
_connection = connection;
AMQQueue queue = new SpecialQueue(connection, "ServiceQueue");
_session = (AMQSession) _connection.createSession(false, AMQSession.NO_ACKNOWLEDGE);
@@ -56,7 +58,7 @@
{
try
{
- Message response = _session.createTextMessage("Response!");
+ Message response = _session.createTextMessage("Response! " + _id);
Destination replyTo = request.getJMSReplyTo();
_session.createProducer(replyTo).send(response);
}
Modified: incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/test/unit/client/forwardall/ServiceCreator.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/test/unit/client/forwardall/ServiceCreator.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/test/unit/client/forwardall/ServiceCreator.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/test/unit/client/forwardall/ServiceCreator.java Mon Apr 21 07:03:35 2008
@@ -34,17 +34,19 @@
private final String broker;
private Service service;
+ private final int id;
- ServiceCreator(String broker)
+ ServiceCreator(String broker, final int id)
{
this.broker = broker;
+ this.id = id;
}
public void run()
{
try
{
- service = new Service(broker);
+ service = new Service(broker, id);
}
catch (Exception e)
{
@@ -76,11 +78,12 @@
{
threads = new Thread[services];
_services = new ServiceCreator[services];
- ServiceCreator runner = new ServiceCreator(broker);
+ //ServiceCreator runner = new ServiceCreator(broker);
// start services
_logger.info("Starting " + services + " services...");
for (int i = 0; i < services; i++)
{
+ ServiceCreator runner = new ServiceCreator(broker,i);
threads[i] = new Thread(runner);
_services[i] = runner;
threads[i].start();
Modified: incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/test/unit/close/MessageRequeueTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/test/unit/close/MessageRequeueTest.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/test/unit/close/MessageRequeueTest.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/test/unit/close/MessageRequeueTest.java Mon Apr 21 07:03:35 2008
@@ -40,30 +40,40 @@
import javax.jms.Session;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.Random;
+import java.util.UUID;
public class MessageRequeueTest extends TestCase
{
private static final Logger _logger = LoggerFactory.getLogger(MessageRequeueTest.class);
+
+
protected static AtomicInteger consumerIds = new AtomicInteger(0);
protected final Integer numTestMessages = 150;
protected final int consumeTimeout = 3000;
- protected final String queue = "direct://amq.direct//queue";
+ //protected final String queue = "direct://amq.direct//queue";
protected String payload = "Message:";
protected final String BROKER = "vm://:1";
private boolean testReception = true;
private long[] receieved = new long[numTestMessages + 1];
- private boolean passed = false;
+ //private boolean passed = false;
protected void setUp() throws Exception
{
super.setUp();
TransportConnection.createVMBroker(1);
+
+ }
+
+ private void putMessagesOnQueueThenClose(String queue)
+ throws JMSException, InterruptedException
+ {
QpidClientConnection conn = new QpidClientConnection(BROKER);
conn.connect();
@@ -76,20 +86,25 @@
conn.disconnect();
}
- protected void tearDown() throws Exception
+
+
+ private void tearDownQueue(String queue)
+ throws JMSException, InterruptedException
+
{
- super.tearDown();
+ QpidClientConnection conn = new QpidClientConnection(BROKER);
- if (!passed) // clean up
- {
- QpidClientConnection conn = new QpidClientConnection(BROKER);
+ conn.connect();
+ // clear queue
+ conn.consume(queue, consumeTimeout);
- conn.connect();
- // clear queue
- conn.consume(queue, consumeTimeout);
+ conn.disconnect();
+ }
- conn.disconnect();
- }
+
+ protected void tearDown() throws Exception
+ {
+ super.tearDown();
TransportConnection.killVMBroker(1);
}
@@ -102,6 +117,11 @@
*/
public void testDrain() throws JMSException, InterruptedException
{
+
+ String queue = "direct://amq.direct//queue" + UUID.randomUUID();
+
+ putMessagesOnQueueThenClose(queue);
+
QpidClientConnection conn = new QpidClientConnection(BROKER);
conn.connect();
@@ -172,18 +192,22 @@
assertEquals(list.toString(), 0, failed);
_logger.info("consumed: " + messagesReceived);
conn.disconnect();
- passed = true;
+ tearDownQueue(queue);
}
/** multiple consumers
* Based on code subbmitted by client FT-304
*/
- public void testCompetingConsumers()
+ public void testCompetingConsumers() throws JMSException, InterruptedException
{
- Consumer c1 = new Consumer();
- Consumer c2 = new Consumer();
- Consumer c3 = new Consumer();
- Consumer c4 = new Consumer();
+ String queue = "direct://amq.direct//queue" + UUID.randomUUID();
+
+ putMessagesOnQueueThenClose(queue);
+
+ Consumer c1 = new Consumer(queue);
+ Consumer c2 = new Consumer(queue);
+ Consumer c3 = new Consumer(queue);
+ Consumer c4 = new Consumer(queue);
Thread t1 = new Thread(c1);
Thread t2 = new Thread(c2);
@@ -237,16 +261,18 @@
assertEquals(list.toString() + "-" + numTestMessages + "-" + totalConsumed, 0, failed);
assertTrue("number of consumed messages does not match initial data: " + totalConsumed, numTestMessages <= totalConsumed);
- passed = true;
+ tearDownQueue(queue);
}
class Consumer implements Runnable
{
private Integer count = 0;
private Integer id;
+ private final String _queue;
- public Consumer()
+ public Consumer(String queue)
{
+ _queue = queue;
id = consumerIds.addAndGet(1);
}
@@ -263,7 +289,7 @@
Message result;
do
{
- result = conn.getNextMessage(queue, consumeTimeout);
+ result = conn.getNextMessage(_queue, consumeTimeout);
if (result != null)
{
@@ -322,8 +348,11 @@
}
}
- public void testRequeue() throws JMSException, AMQException, URLSyntaxException
+ public void testRequeue() throws JMSException, AMQException, URLSyntaxException, InterruptedException
{
+ String queue = "direct://amq.direct//queue" + UUID.randomUUID();
+ putMessagesOnQueueThenClose(queue);
+
int run = 0;
// while (run < 10)
{
Modified: incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/test/unit/topic/DurableSubscriptionTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/test/unit/topic/DurableSubscriptionTest.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/test/unit/topic/DurableSubscriptionTest.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/test/unit/topic/DurableSubscriptionTest.java Mon Apr 21 07:03:35 2008
@@ -88,7 +88,7 @@
Message msg;
_logger.info("Receive message on consumer 1:expecting A");
- msg = consumer1.receive();
+ msg = consumer1.receive(1000);
assertEquals("A", ((TextMessage) msg).getText());
_logger.info("Receive message on consumer 1 :expecting null");
msg = consumer1.receive(1000);
Modified: incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/test/unit/topic/TopicSessionTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/test/unit/topic/TopicSessionTest.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/test/unit/topic/TopicSessionTest.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/client/src/test/java/org/apache/qpid/test/unit/topic/TopicSessionTest.java Mon Apr 21 07:03:35 2008
@@ -20,28 +20,33 @@
*/
package org.apache.qpid.test.unit.topic;
-import javax.jms.InvalidDestinationException;
-import javax.jms.JMSException;
+import javax.jms.*;
import javax.jms.MessageConsumer;
import javax.jms.Session;
-import javax.jms.TemporaryTopic;
-import javax.jms.TextMessage;
-import javax.jms.TopicPublisher;
-import javax.jms.TopicSession;
-import javax.jms.TopicSubscriber;
+import javax.jms.Message;
import junit.framework.TestCase;
import org.apache.qpid.client.AMQConnection;
import org.apache.qpid.client.AMQSession;
import org.apache.qpid.client.AMQTopic;
+import org.apache.qpid.client.AMQQueue;
+import org.apache.qpid.client.BasicMessageProducer;
import org.apache.qpid.client.transport.TransportConnection;
+import org.apache.qpid.url.URLSyntaxException;
+import org.apache.qpid.AMQException;
+import org.apache.qpid.jms.*;
+
+import java.util.UUID;
/** @author Apache Software Foundation */
public class TopicSessionTest extends TestCase
{
private static final String BROKER = "vm://:1";
+ private static final int THREADS = 20;
+ private static final int MESSAGE_COUNT = 10000;
+ private static final int MESSAGE_SIZE = 128;
protected void setUp() throws Exception
{
@@ -102,6 +107,60 @@
subscriptionNameReuseForDifferentTopic(true);
}
+ public void notestSilly() throws Exception
+ {
+
+
+ final ExceptionListener listener = new ExceptionListener()
+ {
+ public void onException(JMSException jmsException)
+ {
+ //To change body of implemented methods use File | Settings | File Templates.
+ }
+ };
+
+
+ Thread[] threads = new Thread[100];
+
+ for(int j = 0; j < 20; j++)
+ {
+ threads[j] = new Thread(new Runnable() {
+ public void run()
+ {
+ try
+ {
+ AMQConnection con = new AMQConnection("tcp://127.0.0.1:5672?retries='0'", "guest", "guest", "test", "test");
+ AMQTopic topic = new AMQTopic(con, "MyTopic1" + UUID.randomUUID());
+
+
+ TopicSession session1 = con.createTopicSession(false, AMQSession.NO_ACKNOWLEDGE);
+
+ con.setExceptionListener(listener);
+
+ TopicPublisher publisher = session1.createPublisher(topic);
+
+ con.start();
+
+ while(true)
+ {
+ publisher.publish(session1.createTextMessage("hello"));
+ Thread.sleep(THREADS);
+ }
+ }
+ catch(Exception e)
+ {
+ e.printStackTrace();
+ }
+ }
+ });
+ threads[j].run();
+ }
+
+ threads[0].join();
+
+ }
+
+
private void subscriptionNameReuseForDifferentTopic(boolean shutdown) throws Exception
{
AMQConnection con = new AMQConnection("vm://:1?retries='0'", "guest", "guest", "test", "test");
@@ -368,8 +427,160 @@
con2.close();
}
+
+ public void noTestPublishToManyConsumers() throws Exception
+ {
+
+
+ final ExceptionListener exceptionListener = new ExceptionListener()
+ {
+ public void onException(JMSException jmsException)
+ {
+ jmsException.printStackTrace();
+ }
+ };
+
+
+
+ SubscribingThread[] threads = new SubscribingThread[100];
+
+ final String topicName = "MyTopic1" + UUID.randomUUID();
+ for(int j = 0; j < 21; j++)
+ {
+ final int threadId = j;
+ threads[threadId] = new SubscribingThread(threadId, topicName, exceptionListener);
+ threads[j].start();
+ Thread.sleep(100);
+ }
+
+
+ threads[1].join();
+
+ int totalMessages = 0;
+
+ for(int j = 1; j < 21; j++)
+ {
+
+ System.err.println("Thread " + j + ": " + threads[j].msgId);
+ totalMessages += threads[j].msgId;
+ }
+
+ System.err.println("****** Total: " + totalMessages);
+
+
+ }
+
+
+
+
public static junit.framework.Test suite()
{
return new junit.framework.TestSuite(TopicSessionTest.class);
+ }
+
+ private static class SubscribingThread extends Thread
+ {
+ private final int _threadId;
+ private final String _topicName;
+ private final ExceptionListener _exceptionListener;
+ int msgId = 0;
+
+ public SubscribingThread(final int threadId, final String topicName, final ExceptionListener exceptionListener)
+ {
+ _threadId = threadId;
+ _topicName = topicName;
+ _exceptionListener = exceptionListener;
+ }
+
+ public void run()
+ {
+ try
+ {
+ System.err.println("Thread: " + _threadId);
+
+
+ if(_threadId >0)
+ {
+
+ AMQConnection con2 = new AMQConnection("tcp://127.0.0.1:5672?retries='0'", "guest", "guest", "test", "test");
+ //AMQConnection con2 = new AMQConnection(BROKER + "?retries='0'", "guest", "guest", "test", "test");
+ AMQTopic topic2 = new AMQTopic(con2, _topicName);
+ TopicSession session2 = con2.createTopicSession(false, AMQSession.NO_ACKNOWLEDGE);
+ TopicSubscriber sub = session2.createSubscriber(topic2);
+ con2.setExceptionListener(_exceptionListener);
+
+
+
+ final MessageListener messageListener = new MessageListener()
+ {
+
+ public void onMessage(Message message)
+ {
+ try
+ {
+ msgId = message.getIntProperty("MessageId");
+ if(msgId % 1000 == 0)
+ {
+ System.err.println("Thread: " + _threadId + ": " + msgId + "messages");
+ }
+ }
+ catch (JMSException e)
+ {
+ e.printStackTrace(); //To change body of catch statement use File | Settings | File Templates.
+ }
+ }
+ };
+
+
+ sub.setMessageListener(messageListener);
+ con2.start();
+
+ Thread.sleep(125000);
+
+
+// Thread.sleep(1200000);
+ }
+ else
+ {
+ int messageId = 0;
+
+ AMQConnection con = new AMQConnection("tcp://127.0.0.1:5672?retries='0'", "guest", "guest", "test", "test");
+ //AMQConnection con = new AMQConnection(BROKER + "?retries='0'", "guest", "guest", "test", "test");
+
+
+ AMQTopic topic = new AMQTopic(con, _topicName);
+
+
+ TopicSession session1 = con.createTopicSession(false, AMQSession.NO_ACKNOWLEDGE);
+
+ con.setExceptionListener(_exceptionListener);
+
+ TopicPublisher publisher = session1.createPublisher(topic);
+ publisher.setDisableMessageID(true);
+ publisher.setDisableMessageTimestamp(true);
+ con.start();
+
+ Thread.sleep(5000);
+
+ publisher.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
+
+ while(messageId <= 240000)
+ //while(messageId <= 10000)
+ {
+ final TextMessage textMessage = session1.createTextMessage("hello");
+ textMessage.setIntProperty("MessageId", messageId++);
+
+
+ publisher.publish(textMessage);
+
+ }
+ }
+
+ }
+ catch(Exception e)
+ {
+ e.printStackTrace();
+ }
+ }
}
}
Modified: incubator/qpid/branches/broker-queue-refactor/java/common/src/main/java/org/apache/mina/common/FixedSizeByteBufferAllocator.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/common/src/main/java/org/apache/mina/common/FixedSizeByteBufferAllocator.java?rev=650148&r1=650147&r2=650148&view=diff
==============================================================================
--- incubator/qpid/branches/broker-queue-refactor/java/common/src/main/java/org/apache/mina/common/FixedSizeByteBufferAllocator.java (original)
+++ incubator/qpid/branches/broker-queue-refactor/java/common/src/main/java/org/apache/mina/common/FixedSizeByteBufferAllocator.java Mon Apr 21 07:03:35 2008
@@ -62,7 +62,6 @@
private static final class FixedSizeByteBuffer extends ByteBuffer
{
private java.nio.ByteBuffer buf;
- private int refCount = 1;
private int mark = -1;
@@ -70,36 +69,14 @@
{
this.buf = buf;
buf.order( ByteOrder.BIG_ENDIAN );
- refCount = 1;
}
public synchronized void acquire()
{
- if( refCount <= 0 )
- {
- throw new IllegalStateException( "Already released buffer." );
- }
-
- refCount ++;
}
public void release()
{
- synchronized( this )
- {
- if( refCount <= 0 )
- {
- refCount = 0;
- throw new IllegalStateException(
- "Already released buffer. You released the buffer too many times." );
- }
-
- refCount --;
- if( refCount > 0)
- {
- return;
- }
- }
}
public java.nio.ByteBuffer buf()
@@ -157,49 +134,11 @@
{
if( newCapacity > capacity() )
{
- // Allocate a new buffer and transfer all settings to it.
- int pos = position();
- int limit = limit();
- ByteOrder bo = order();
-
- capacity0( newCapacity );
- buf.limit( limit );
- if( mark >= 0 )
- {
- buf.position( mark );
- buf.mark();
- }
- buf.position( pos );
- buf.order( bo );
+ throw new IllegalArgumentException();
}
return this;
}
-
- protected void capacity0( int requestedCapacity )
- {
- int newCapacity = MINIMUM_CAPACITY;
- while( newCapacity < requestedCapacity )
- {
- newCapacity <<= 1;
- }
-
- java.nio.ByteBuffer oldBuf = this.buf;
- java.nio.ByteBuffer newBuf;
- if( isDirect() )
- {
- newBuf = java.nio.ByteBuffer.allocateDirect( newCapacity );
- }
- else
- {
- newBuf = java.nio.ByteBuffer.allocate( newCapacity );
- }
-
- newBuf.clear();
- oldBuf.clear();
- newBuf.put( oldBuf );
- this.buf = newBuf;
- }