You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ri...@apache.org on 2008/10/23 13:33:45 UTC
svn commit: r707344 - in /incubator/qpid/trunk/qpid/java/broker/src:
main/java/org/apache/qpid/server/ test/java/org/apache/qpid/server/
test/java/org/apache/qpid/server/queue/
test/java/org/apache/qpid/server/subscription/
Author: ritchiem
Date: Thu Oct 23 04:33:45 2008
New Revision: 707344
URL: http://svn.apache.org/viewvc?rev=707344&view=rev
Log:
QPID-1385 : Extracted annonymous class from AMQChannel.resend(boolean), Added new Mock Objects to the broke to allow direct testing of the new class.
Added:
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/ExtractResendAndRequeue.java
incubator/qpid/trunk/qpid/java/broker/src/test/java/org/apache/qpid/server/ExtractResendAndRequeueTest.java
incubator/qpid/trunk/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/MockAMQMessage.java
incubator/qpid/trunk/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/MockAMQMessageHandle.java
incubator/qpid/trunk/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/MockAMQQueue.java
incubator/qpid/trunk/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/MockMessagePublishInfo.java
Modified:
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java
incubator/qpid/trunk/qpid/java/broker/src/test/java/org/apache/qpid/server/subscription/MockSubscription.java
Modified: incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java?rev=707344&r1=707343&r2=707344&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java (original)
+++ incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java Thu Oct 23 04:33:45 2008
@@ -22,9 +22,7 @@
import org.apache.log4j.Logger;
import org.apache.qpid.AMQException;
-import org.apache.qpid.configuration.Configured;
import org.apache.qpid.framing.AMQShortString;
-import org.apache.qpid.framing.BasicContentHeaderProperties;
import org.apache.qpid.framing.ContentBody;
import org.apache.qpid.framing.ContentHeaderBody;
import org.apache.qpid.framing.FieldTable;
@@ -60,7 +58,6 @@
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicInteger;
public class AMQChannel
{
@@ -594,60 +591,9 @@
// Process the Unacked-Map.
// Marking messages who still have a consumer for to be resent
// and those that don't to be requeued.
+ _unacknowledgedMessageMap.visit(new ExtractResendAndRequeue(_unacknowledgedMessageMap, msgToRequeue,
+ msgToResend, requeue, _storeContext));
- _unacknowledgedMessageMap.visit(new UnacknowledgedMessageMap.Visitor()
- {
- public boolean callback(final long deliveryTag, QueueEntry message) throws AMQException
- {
-
- AMQMessage msg = message.getMessage();
- msg.setRedelivered(true);
- final Subscription subscription = message.getDeliveredSubscription();
- if (subscription != null)
- {
- // Consumer exists
- if (!subscription.isClosed())
- {
- msgToResend.put(deliveryTag, message);
- }
- else // consumer has gone
- {
- msgToRequeue.put(deliveryTag, message);
- }
- }
- else
- {
- // Message has no consumer tag, so was "delivered" to a GET
- // or consumer no longer registered
- // cannot resend, so re-queue.
- if (!message.isQueueDeleted())
- {
- if (requeue)
- {
- msgToRequeue.put(deliveryTag, message);
- }
- else
- {
- _log.info("No DeadLetter Queue and requeue not requested so dropping message:" + message);
- }
- }
- else
- {
- message.discard(_storeContext);
- _log.warn("Message.queue is null and no DeadLetter Queue so dropping message:" + message);
- }
- }
-
- // false means continue processing
- return false;
- }
-
- public void visitComplete()
- {
- }
- });
-
- _unacknowledgedMessageMap.clear();
// Process Messages to Resend
if (_log.isDebugEnabled())
Added: incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/ExtractResendAndRequeue.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/ExtractResendAndRequeue.java?rev=707344&view=auto
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/ExtractResendAndRequeue.java (added)
+++ incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/ExtractResendAndRequeue.java Thu Oct 23 04:33:45 2008
@@ -0,0 +1,110 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.server;
+
+import org.apache.qpid.server.ack.UnacknowledgedMessageMap;
+import org.apache.qpid.server.ack.UnacknowledgedMessageMapImpl;
+import org.apache.qpid.server.queue.QueueEntry;
+import org.apache.qpid.server.queue.AMQMessage;
+import org.apache.qpid.server.subscription.Subscription;
+import org.apache.qpid.server.store.StoreContext;
+import org.apache.qpid.AMQException;
+import org.apache.log4j.Logger;
+
+import java.util.Map;
+
+public class ExtractResendAndRequeue implements UnacknowledgedMessageMap.Visitor
+{
+ private static final Logger _log = Logger.getLogger(ExtractResendAndRequeue.class);
+
+ private Map<Long, QueueEntry> _msgToRequeue;
+ private Map<Long, QueueEntry> _msgToResend;
+ private boolean _requeueIfUnabletoResend;
+ private StoreContext _storeContext;
+ private UnacknowledgedMessageMap _unacknowledgedMessageMap;
+
+ public ExtractResendAndRequeue(UnacknowledgedMessageMap unacknowledgedMessageMap,
+ Map<Long, QueueEntry> msgToRequeue,
+ Map<Long, QueueEntry> msgToResend,
+ boolean requeueIfUnabletoResend,
+ StoreContext storeContext)
+ {
+ _unacknowledgedMessageMap = unacknowledgedMessageMap;
+ _msgToRequeue = msgToRequeue;
+ _msgToResend = msgToResend;
+ _requeueIfUnabletoResend = requeueIfUnabletoResend;
+ _storeContext = storeContext;
+ }
+
+ @Override
+ public boolean callback(final long deliveryTag, QueueEntry message) throws AMQException
+ {
+
+ AMQMessage msg = message.getMessage();
+ msg.setRedelivered(true);
+ final Subscription subscription = message.getDeliveredSubscription();
+ if (subscription != null)
+ {
+ // Consumer exists
+ if (!subscription.isClosed())
+ {
+ _msgToResend.put(deliveryTag, message);
+ }
+ else // consumer has gone
+ {
+ _msgToRequeue.put(deliveryTag, message);
+ }
+ }
+ else
+ {
+ // Message has no consumer tag, so was "delivered" to a GET
+ // or consumer no longer registered
+ // cannot resend, so re-queue.
+ if (!message.isQueueDeleted())
+ {
+ if (_requeueIfUnabletoResend)
+ {
+ _msgToRequeue.put(deliveryTag, message);
+ }
+ else
+ {
+ message.discard(_storeContext);
+ _log.info("No DeadLetter Queue and requeue not requested so dropping message:" + message);
+ }
+ }
+ else
+ {
+ message.discard(_storeContext);
+ _log.warn("Message.queue is null and no DeadLetter Queue so dropping message:" + message);
+ }
+ }
+
+ // false means continue processing
+ return false;
+ }
+
+ @Override
+ public void visitComplete()
+ {
+ _unacknowledgedMessageMap.clear();
+ }
+
+}
Added: incubator/qpid/trunk/qpid/java/broker/src/test/java/org/apache/qpid/server/ExtractResendAndRequeueTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/test/java/org/apache/qpid/server/ExtractResendAndRequeueTest.java?rev=707344&view=auto
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/test/java/org/apache/qpid/server/ExtractResendAndRequeueTest.java (added)
+++ incubator/qpid/trunk/qpid/java/broker/src/test/java/org/apache/qpid/server/ExtractResendAndRequeueTest.java Thu Oct 23 04:33:45 2008
@@ -0,0 +1,255 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.server;
+
+import junit.framework.TestCase;
+import org.apache.qpid.server.ack.UnacknowledgedMessageMapImpl;
+import org.apache.qpid.server.queue.MockQueueEntry;
+import org.apache.qpid.server.queue.QueueEntry;
+import org.apache.qpid.server.queue.SimpleQueueEntryList;
+import org.apache.qpid.server.queue.MockAMQMessage;
+import org.apache.qpid.server.queue.AMQQueue;
+import org.apache.qpid.server.queue.MockAMQQueue;
+import org.apache.qpid.server.queue.AMQMessage;
+import org.apache.qpid.server.queue.QueueEntryIterator;
+import org.apache.qpid.server.store.StoreContext;
+import org.apache.qpid.server.subscription.Subscription;
+import org.apache.qpid.server.subscription.MockSubscription;
+import org.apache.qpid.AMQException;
+
+import java.util.Map;
+import java.util.LinkedHashMap;
+import java.util.LinkedList;
+import java.util.Iterator;
+
+/**
+ * QPID-1385 : Race condition between added to unacked map and resending due to a rollback.
+ *
+ * In AMQChannel _unackedMap.clear() was done after the visit. This meant that the clear was not in the same
+ * synchronized block as as the preparation to resend.
+ *
+ * This clearing/prep for resend was done as a result of the rollback call. HOWEVER, the delivery thread was still
+ * in the process of sending messages to the client. It is therefore possible that a message could block on the
+ * _unackedMap lock waiting for the visit to compelete so that it can add the new message to the unackedMap....
+ * which is then cleared by the resend/rollback thread.
+ *
+ * This problem was encountered by the testSend2ThenRollback test.
+ *
+ * To try and increase the chance of the race condition occuring this test will send multiple messages so that the
+ * delivery thread will be in progress while the rollback method is called. Hopefully this will cause the
+ * deliveryTag to be lost
+ */
+public class ExtractResendAndRequeueTest extends TestCase
+{
+
+ UnacknowledgedMessageMapImpl _unacknowledgedMessageMap;
+ private static final int INITIAL_MSG_COUNT = 10;
+ private AMQQueue _queue = new MockAMQQueue();
+ private LinkedList<QueueEntry> _referenceList = new LinkedList<QueueEntry>();
+
+ @Override
+ public void setUp() throws AMQException
+ {
+ _unacknowledgedMessageMap = new UnacknowledgedMessageMapImpl(100);
+
+ long id = 0;
+ SimpleQueueEntryList list = new SimpleQueueEntryList(_queue);
+
+ // Add initial messages to QueueEntryList
+ for (int count = 0; count < INITIAL_MSG_COUNT; count++)
+ {
+ AMQMessage msg = new MockAMQMessage(id);
+
+ list.add(msg);
+
+ //Increment ID;
+ id++;
+ }
+
+ // Iterate through the QueueEntryList and add entries to unacknowledgeMessageMap and referecenList
+ QueueEntryIterator queueEntries = list.iterator();
+ while(queueEntries.advance())
+ {
+ QueueEntry entry = queueEntries.getNode();
+ _unacknowledgedMessageMap.add(entry.getMessage().getMessageId(), entry);
+
+ // Store the entry for future inspection
+ _referenceList.add(entry);
+ }
+
+ assertEquals("Map does not contain correct setup data", INITIAL_MSG_COUNT, _unacknowledgedMessageMap.size());
+ }
+
+ /**
+ * Helper method to create a new subscription and aquire the given messages.
+ *
+ * @param messageList The messages to aquire
+ *
+ * @return Subscription that performed the aquire
+ */
+ private Subscription createSubscriptionAndAquireMessages(LinkedList<QueueEntry> messageList)
+ {
+ Subscription subscription = new MockSubscription();
+
+ // Aquire messages in subscription
+ for (QueueEntry entry : messageList)
+ {
+ entry.acquire(subscription);
+ }
+
+ return subscription;
+ }
+
+ /**
+ * This is the normal consumer rollback method.
+ *
+ * An active consumer that has aquired messages expects those messasges to be reset when rollback is requested.
+ *
+ * This test validates that the msgToResend map includes all the messages and none are left behind.
+ *
+ * @throws AMQException the visit interface throws this
+ */
+ public void testResend() throws AMQException
+ {
+ //We don't need the subscription object here.
+ createSubscriptionAndAquireMessages(_referenceList);
+
+ final Map<Long, QueueEntry> msgToRequeue = new LinkedHashMap<Long, QueueEntry>();
+ final Map<Long, QueueEntry> msgToResend = new LinkedHashMap<Long, QueueEntry>();
+
+ // requeueIfUnabletoResend doesn't matter here.
+ _unacknowledgedMessageMap.visit(new ExtractResendAndRequeue(_unacknowledgedMessageMap, msgToRequeue,
+ msgToResend, true, new StoreContext()));
+
+ assertEquals("Message count for resend not correct.", INITIAL_MSG_COUNT, msgToResend.size());
+ assertEquals("Message count for requeue not correct.", 0, msgToRequeue.size());
+ assertEquals("Map was not emptied", 0, _unacknowledgedMessageMap.size());
+ }
+
+ /**
+ * This is the normal consumer close method.
+ *
+ * When a consumer that has aquired messages expects closes the messages that it has aquired should be removed from
+ * the unacknowledgeMap and placed in msgToRequeue
+ *
+ * This test validates that the msgToRequeue map includes all the messages and none are left behind.
+ *
+ * @throws AMQException the visit interface throws this
+ */
+ public void testRequeueDueToSubscriptionClosure() throws AMQException
+ {
+ Subscription subscription = createSubscriptionAndAquireMessages(_referenceList);
+
+ // Close subscription
+ subscription.close();
+
+ final Map<Long, QueueEntry> msgToRequeue = new LinkedHashMap<Long, QueueEntry>();
+ final Map<Long, QueueEntry> msgToResend = new LinkedHashMap<Long, QueueEntry>();
+
+ // requeueIfUnabletoResend doesn't matter here.
+ _unacknowledgedMessageMap.visit(new ExtractResendAndRequeue(_unacknowledgedMessageMap, msgToRequeue,
+ msgToResend, true, new StoreContext()));
+
+ assertEquals("Message count for resend not correct.", 0, msgToResend.size());
+ assertEquals("Message count for requeue not correct.", INITIAL_MSG_COUNT, msgToRequeue.size());
+ assertEquals("Map was not emptied", 0, _unacknowledgedMessageMap.size());
+ }
+
+ /**
+ * If the subscription is null, due to message being retrieved via a GET, And we request that messages are requeued
+ * requeueIfUnabletoResend(set to true) then all messages should be sent to the msgToRequeue map.
+ *
+ * @throws AMQException the visit interface throws this
+ */
+
+ public void testRequeueDueToMessageHavingNoConsumerTag() throws AMQException
+ {
+ final Map<Long, QueueEntry> msgToRequeue = new LinkedHashMap<Long, QueueEntry>();
+ final Map<Long, QueueEntry> msgToResend = new LinkedHashMap<Long, QueueEntry>();
+
+ // requeueIfUnabletoResend = true so all messages should go to msgToRequeue
+ _unacknowledgedMessageMap.visit(new ExtractResendAndRequeue(_unacknowledgedMessageMap, msgToRequeue,
+ msgToResend, true, new StoreContext()));
+
+ assertEquals("Message count for resend not correct.", 0, msgToResend.size());
+ assertEquals("Message count for requeue not correct.", INITIAL_MSG_COUNT, msgToRequeue.size());
+ assertEquals("Map was not emptied", 0, _unacknowledgedMessageMap.size());
+ }
+
+ /**
+ * If the subscription is null, due to message being retrieved via a GET, And we request that we don't
+ * requeueIfUnabletoResend(set to false) then all messages should be dropped as we do not have a dead letter queue.
+ *
+ * @throws AMQException the visit interface throws this
+ */
+
+ public void testDrop() throws AMQException
+ {
+ final Map<Long, QueueEntry> msgToRequeue = new LinkedHashMap<Long, QueueEntry>();
+ final Map<Long, QueueEntry> msgToResend = new LinkedHashMap<Long, QueueEntry>();
+
+ // requeueIfUnabletoResend = false so all messages should be dropped all maps should be empty
+ _unacknowledgedMessageMap.visit(new ExtractResendAndRequeue(_unacknowledgedMessageMap, msgToRequeue,
+ msgToResend, false, new StoreContext()));
+
+ assertEquals("Message count for resend not correct.", 0, msgToResend.size());
+ assertEquals("Message count for requeue not correct.", 0, msgToRequeue.size());
+ assertEquals("Map was not emptied", 0, _unacknowledgedMessageMap.size());
+
+
+ for (QueueEntry entry : _referenceList)
+ {
+ assertTrue("Message was not discarded", entry.isDeleted());
+ }
+
+ }
+
+ /**
+ * If the subscription is null, due to message being retrieved via a GET, AND the queue upon which the message was
+ * delivered has been deleted then it is not possible to requeue. Currently we simply discar the message but in the
+ * future we may wish to dead letter the message.
+ *
+ * Validate that at the end of the visit all Maps are empty and all messages are marked as deleted
+ *
+ * @throws AMQException the visit interface throws this
+ */
+ public void testDiscard() throws AMQException
+ {
+ final Map<Long, QueueEntry> msgToRequeue = new LinkedHashMap<Long, QueueEntry>();
+ final Map<Long, QueueEntry> msgToResend = new LinkedHashMap<Long, QueueEntry>();
+
+ _queue.delete();
+
+ // requeueIfUnabletoResend : value doesn't matter here as queue has been deleted
+ _unacknowledgedMessageMap.visit(new ExtractResendAndRequeue(_unacknowledgedMessageMap, msgToRequeue,
+ msgToResend, false, new StoreContext()));
+
+ assertEquals("Message count for resend not correct.", 0, msgToResend.size());
+ assertEquals("Message count for requeue not correct.", 0, msgToRequeue.size());
+ assertEquals("Map was not emptied", 0, _unacknowledgedMessageMap.size());
+
+ for (QueueEntry entry : _referenceList)
+ {
+ assertTrue("Message was not discarded", entry.isDeleted());
+ }
+ }
+
+}
Added: incubator/qpid/trunk/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/MockAMQMessage.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/MockAMQMessage.java?rev=707344&view=auto
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/MockAMQMessage.java (added)
+++ incubator/qpid/trunk/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/MockAMQMessage.java Thu Oct 23 04:33:45 2008
@@ -0,0 +1,49 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.server.queue;
+
+import org.apache.qpid.server.store.StoreContext;
+import org.apache.qpid.AMQException;
+import org.apache.qpid.framing.abstraction.MessagePublishInfo;
+
+public class MockAMQMessage extends AMQMessage
+{
+ public MockAMQMessage(long messageId)
+ throws AMQException
+ {
+ super(new MockAMQMessageHandle(messageId) ,
+ (StoreContext)null,
+ (MessagePublishInfo)new MockMessagePublishInfo());
+ }
+
+ protected MockAMQMessage(AMQMessage msg)
+ throws AMQException
+ {
+ super(msg);
+ }
+
+
+ @Override
+ public long getSize()
+ {
+ return 0l;
+ }
+}
Added: incubator/qpid/trunk/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/MockAMQMessageHandle.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/MockAMQMessageHandle.java?rev=707344&view=auto
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/MockAMQMessageHandle.java (added)
+++ incubator/qpid/trunk/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/MockAMQMessageHandle.java Thu Oct 23 04:33:45 2008
@@ -0,0 +1,37 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.server.queue;
+
+import org.apache.qpid.server.store.StoreContext;
+
+public class MockAMQMessageHandle extends InMemoryMessageHandle
+{
+ public MockAMQMessageHandle(final Long messageId)
+ {
+ super(messageId);
+ }
+
+ @Override
+ public long getBodySize(StoreContext store)
+ {
+ return 0l;
+ }
+}
Added: incubator/qpid/trunk/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/MockAMQQueue.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/MockAMQQueue.java?rev=707344&view=auto
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/MockAMQQueue.java (added)
+++ incubator/qpid/trunk/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/MockAMQQueue.java Thu Oct 23 04:33:45 2008
@@ -0,0 +1,313 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.server.queue;
+
+import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.framing.FieldTable;
+import org.apache.qpid.server.virtualhost.VirtualHost;
+import org.apache.qpid.server.exchange.Exchange;
+import org.apache.qpid.server.subscription.Subscription;
+import org.apache.qpid.server.store.StoreContext;
+import org.apache.qpid.server.management.ManagedObject;
+import org.apache.qpid.server.registry.ApplicationRegistry;
+import org.apache.qpid.AMQException;
+import org.apache.commons.configuration.Configuration;
+
+import java.util.List;
+import java.util.Set;
+import java.util.Map;
+import java.util.HashMap;
+import java.util.LinkedList;
+
+public class MockAMQQueue implements AMQQueue
+{
+ private boolean _deleted = false;
+
+ public AMQShortString getName()
+ {
+ return null; //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public boolean isDurable()
+ {
+ return false; //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public boolean isAutoDelete()
+ {
+ return false; //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public AMQShortString getOwner()
+ {
+ return null; //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public VirtualHost getVirtualHost()
+ {
+ return null; //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public void bind(Exchange exchange, AMQShortString routingKey, FieldTable arguments) throws AMQException
+ {
+ //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public void unBind(Exchange exchange, AMQShortString routingKey, FieldTable arguments) throws AMQException
+ {
+ //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public List<ExchangeBinding> getExchangeBindings()
+ {
+ return null; //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public void registerSubscription(Subscription subscription, boolean exclusive) throws AMQException
+ {
+ //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public void unregisterSubscription(Subscription subscription) throws AMQException
+ {
+ //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public int getConsumerCount()
+ {
+ return 0; //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public int getActiveConsumerCount()
+ {
+ return 0; //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public boolean isUnused()
+ {
+ return false; //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public boolean isEmpty()
+ {
+ return false; //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public int getMessageCount()
+ {
+ return 0; //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public int getUndeliveredMessageCount()
+ {
+ return 0; //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public long getQueueDepth()
+ {
+ return 0; //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public long getReceivedMessageCount()
+ {
+ return 0; //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public long getOldestMessageArrivalTime()
+ {
+ return 0; //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public boolean isDeleted()
+ {
+ return _deleted;
+ }
+
+ public int delete() throws AMQException
+ {
+ return 0; //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public QueueEntry enqueue(StoreContext storeContext, AMQMessage message) throws AMQException
+ {
+ return null; //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public void requeue(StoreContext storeContext, QueueEntry entry) throws AMQException
+ {
+ //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public void dequeue(StoreContext storeContext, QueueEntry entry) throws FailedDequeueException
+ {
+ //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public boolean resend(QueueEntry entry, Subscription subscription) throws AMQException
+ {
+ return false; //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public void addQueueDeleteTask(Task task)
+ {
+ //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public List<QueueEntry> getMessagesOnTheQueue()
+ {
+ return null; //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public List<QueueEntry> getMessagesOnTheQueue(long fromMessageId, long toMessageId)
+ {
+ return null; //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public List<Long> getMessagesOnTheQueue(int num)
+ {
+ return null; //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public List<Long> getMessagesOnTheQueue(int num, int offest)
+ {
+ return null; //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public QueueEntry getMessageOnTheQueue(long messageId)
+ {
+ return null; //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public void moveMessagesToAnotherQueue(long fromMessageId, long toMessageId, String queueName, StoreContext storeContext)
+ {
+ //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public void copyMessagesToAnotherQueue(long fromMessageId, long toMessageId, String queueName, StoreContext storeContext)
+ {
+ //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public void removeMessagesFromQueue(long fromMessageId, long toMessageId, StoreContext storeContext)
+ {
+ //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public long getMaximumMessageSize()
+ {
+ return 0; //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public void setMaximumMessageSize(long value)
+ {
+ //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public long getMaximumMessageCount()
+ {
+ return 0; //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public void setMaximumMessageCount(long value)
+ {
+ //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public long getMaximumQueueDepth()
+ {
+ return 0; //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public void setMaximumQueueDepth(long value)
+ {
+ //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public long getMaximumMessageAge()
+ {
+ return 0; //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public void setMaximumMessageAge(long maximumMessageAge)
+ {
+ //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public long getMinimumAlertRepeatGap()
+ {
+ return 0; //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public void deleteMessageFromTop(StoreContext storeContext) throws AMQException
+ {
+ //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public long clearQueue(StoreContext storeContext) throws AMQException
+ {
+ return 0; //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public void removeExpiredIfNoSubscribers() throws AMQException
+ {
+ //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public Set<NotificationCheck> getNotificationChecks()
+ {
+ return null; //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public void flushSubscription(Subscription sub) throws AMQException
+ {
+ //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public void deliverAsync(Subscription sub)
+ {
+ //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public void deliverAsync()
+ {
+ //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public void stop()
+ {
+ //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public void configure(Configuration virtualHostDefaultQueueConfiguration)
+ {
+ //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public ManagedObject getManagedObject()
+ {
+ return null; //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public int compareTo(AMQQueue o)
+ {
+ return 0; //To change body of implemented methods use File | Settings | File Templates.
+ }
+}
Added: incubator/qpid/trunk/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/MockMessagePublishInfo.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/MockMessagePublishInfo.java?rev=707344&view=auto
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/MockMessagePublishInfo.java (added)
+++ incubator/qpid/trunk/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/MockMessagePublishInfo.java Thu Oct 23 04:33:45 2008
@@ -0,0 +1,52 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.server.queue;
+
+import org.apache.qpid.framing.abstraction.MessagePublishInfo;
+import org.apache.qpid.framing.AMQShortString;
+
+public class MockMessagePublishInfo implements MessagePublishInfo
+{
+ public AMQShortString getExchange()
+ {
+ return null; //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public void setExchange(AMQShortString exchange)
+ {
+ //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public boolean isImmediate()
+ {
+ return false; //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public boolean isMandatory()
+ {
+ return false; //To change body of implemented methods use File | Settings | File Templates.
+ }
+
+ public AMQShortString getRoutingKey()
+ {
+ return null; //To change body of implemented methods use File | Settings | File Templates.
+ }
+}
Modified: incubator/qpid/trunk/qpid/java/broker/src/test/java/org/apache/qpid/server/subscription/MockSubscription.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/test/java/org/apache/qpid/server/subscription/MockSubscription.java?rev=707344&r1=707343&r2=707344&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/test/java/org/apache/qpid/server/subscription/MockSubscription.java (original)
+++ incubator/qpid/trunk/qpid/java/broker/src/test/java/org/apache/qpid/server/subscription/MockSubscription.java Thu Oct 23 04:33:45 2008
@@ -33,10 +33,10 @@
public class MockSubscription implements Subscription
{
- private boolean closed = false;
+ private boolean _closed = false;
private AMQShortString tag = new AMQShortString("mocktag");
private AMQQueue queue = null;
- private StateListener listener = null;
+ private StateListener _listener = null;
private QueueEntry lastSeen = null;
private State _state = State.ACTIVE;
private ArrayList<QueueEntry> messages = new ArrayList<QueueEntry>();
@@ -44,8 +44,11 @@
@Override
public void close()
{
- closed = true;
- listener.stateChange(this, _state, State.CLOSED);
+ _closed = true;
+ if (_listener != null)
+ {
+ _listener.stateChange(this, _state, State.CLOSED);
+ }
_state = State.CLOSED;
}
@@ -117,7 +120,7 @@
@Override
public boolean isClosed()
{
- return closed;
+ return _closed;
}
@Override
@@ -174,7 +177,7 @@
@Override
public void setStateListener(StateListener listener)
{
- this.listener = listener;
+ this._listener = listener;
}
public State getState()