diff options
| author | Robert Godfrey <rgodfrey@apache.org> | 2014-02-07 16:57:49 +0000 |
|---|---|---|
| committer | Robert Godfrey <rgodfrey@apache.org> | 2014-02-07 16:57:49 +0000 |
| commit | 65b1a1ddfe95b9e273d4cdaf23067a0aaff9b1d1 (patch) | |
| tree | 427d5ce851b9336fea70eb8fc6f87135ad239065 /qpid/java/broker-core | |
| parent | 3ab4f9bdc9bbc8375534f45022a02257eb6e030d (diff) | |
| download | qpid-python-65b1a1ddfe95b9e273d4cdaf23067a0aaff9b1d1.tar.gz | |
QPID-5504 : Refactoring to allow for nodes other than queues to be subscribed from, and nodes other than exchanges to be sent to (merged from separate branch)
git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1565726 13f79535-47bb-0310-9956-ffa450edef68
Diffstat (limited to 'qpid/java/broker-core')
78 files changed, 2770 insertions, 2129 deletions
diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/consumer/AbstractConsumerTarget.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/consumer/AbstractConsumerTarget.java new file mode 100644 index 0000000000..aa721e598a --- /dev/null +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/consumer/AbstractConsumerTarget.java @@ -0,0 +1,73 @@ +/* + * + * 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.consumer; + +import org.apache.qpid.server.util.StateChangeListener; + +import java.util.concurrent.atomic.AtomicReference; + +public abstract class AbstractConsumerTarget implements ConsumerTarget +{ + + private final AtomicReference<State> _state; + private final AtomicReference<StateChangeListener<ConsumerTarget, State>> _stateListener = + new AtomicReference<StateChangeListener<ConsumerTarget, State>>(); + + protected AbstractConsumerTarget(final State initialState) + { + _state = new AtomicReference<State>(initialState); + } + + + public final State getState() + { + return _state.get(); + } + + protected final boolean updateState(State from, State to) + { + if(_state.compareAndSet(from, to)) + { + StateChangeListener<ConsumerTarget, State> listener = _stateListener.get(); + if(listener != null) + { + listener.stateChanged(this, from, to); + } + return true; + } + else + { + return false; + } + } + + + public final void setStateListener(StateChangeListener<ConsumerTarget, State> listener) + { + _stateListener.set(listener); + } + + public final StateChangeListener<ConsumerTarget, State> getStateListener() + { + return _stateListener.get(); + } + +} diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/subscription/Subscription.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/consumer/Consumer.java index fde3d3809c..e082d6eee4 100644 --- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/subscription/Subscription.java +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/consumer/Consumer.java @@ -18,22 +18,29 @@ * under the License. * */ -package org.apache.qpid.server.subscription; +package org.apache.qpid.server.consumer; import java.util.concurrent.atomic.AtomicLong; import org.apache.qpid.AMQException; import org.apache.qpid.server.logging.LogActor; +import org.apache.qpid.server.message.MessageInstance; import org.apache.qpid.server.protocol.AMQSessionModel; -import org.apache.qpid.server.queue.AMQQueue; -import org.apache.qpid.server.queue.QueueEntry; +import org.apache.qpid.server.util.StateChangeListener; -public interface Subscription +public interface Consumer { AtomicLong SUB_ID_GENERATOR = new AtomicLong(0); - LogActor getLogActor(); + void externalStateChange(); - boolean isTransient(); + enum Option + { + ACQUIRES, + SEES_REQUEUES, + TRANSIENT, + EXCLUSIVE, + NO_LOCAL + } long getBytesOut(); @@ -43,49 +50,21 @@ public interface Subscription long getUnacknowledgedMessages(); - public static enum State - { - ACTIVE, - SUSPENDED, - CLOSED - } - - public static interface StateListener - { - public void stateChange(Subscription sub, State oldState, State newState); - } - - AMQQueue getQueue(); AMQSessionModel getSessionModel(); - QueueEntry.SubscriptionAcquiredState getOwningState(); - - void setQueue(AMQQueue queue, boolean exclusive); - void setNoLocal(boolean noLocal); - long getSubscriptionID(); + long getId(); boolean isSuspended(); - boolean hasInterest(QueueEntry msg); - boolean isClosed(); boolean acquires(); boolean seesRequeues(); - void close(); - - void send(QueueEntry entry, boolean batch) throws AMQException; - - void flushBatched(); - - void queueDeleted(AMQQueue queue); - - - boolean wouldSuspend(QueueEntry msg); + void close() throws AMQException; boolean trySendLock(); @@ -94,30 +73,9 @@ public interface Subscription void releaseSendLock(); - void releaseQueueEntry(final QueueEntry queueEntryImpl); - - void onDequeue(final QueueEntry queueEntry); - - void restoreCredit(final QueueEntry queueEntry); - - void setStateListener(final StateListener listener); - - public State getState(); - - AMQQueue.Context getQueueContext(); - - void setQueueContext(AMQQueue.Context queueContext); - - boolean isActive(); - public void set(String key, Object value); - - public Object get(String key); - - boolean isSessionTransactional(); - - void queueEmpty() throws AMQException; + String getName(); - String getConsumerName(); + void flush() throws AMQException; } diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/consumer/ConsumerTarget.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/consumer/ConsumerTarget.java new file mode 100644 index 0000000000..92579475ed --- /dev/null +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/consumer/ConsumerTarget.java @@ -0,0 +1,67 @@ +/* + * + * 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.consumer; + +import org.apache.qpid.AMQException; +import org.apache.qpid.server.message.MessageInstance; +import org.apache.qpid.server.message.ServerMessage; +import org.apache.qpid.server.protocol.AMQSessionModel; +import org.apache.qpid.server.util.StateChangeListener; + +public interface ConsumerTarget +{ + + + enum State + { + ACTIVE, SUSPENDED, CLOSED + } + + State getState(); + + void consumerAdded(Consumer sub); + + void consumerRemoved(Consumer sub); + + void setStateListener(StateChangeListener<ConsumerTarget, State> listener); + + long getUnacknowledgedBytes(); + + long getUnacknowledgedMessages(); + + AMQSessionModel getSessionModel(); + + void send(MessageInstance entry, boolean batch) throws AMQException; + + void flushBatched(); + + void queueDeleted(); + + void queueEmpty() throws AMQException; + + boolean allocateCredit(ServerMessage msg); + + void restoreCredit(ServerMessage queueEntry); + + boolean isSuspended(); + + boolean close(); +} diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/exchange/AbstractExchange.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/exchange/AbstractExchange.java index 6a959df440..bc670bd848 100644 --- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/exchange/AbstractExchange.java +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/exchange/AbstractExchange.java @@ -26,6 +26,7 @@ import org.apache.qpid.AMQException; import org.apache.qpid.AMQInternalException; import org.apache.qpid.AMQSecurityException; import org.apache.qpid.server.binding.Binding; +import org.apache.qpid.server.consumer.Consumer; import org.apache.qpid.server.logging.LogSubject; import org.apache.qpid.server.logging.actors.CurrentActor; import org.apache.qpid.server.logging.messages.BindingMessages; @@ -33,14 +34,17 @@ import org.apache.qpid.server.logging.messages.ExchangeMessages; import org.apache.qpid.server.logging.subjects.BindingLogSubject; import org.apache.qpid.server.logging.subjects.ExchangeLogSubject; import org.apache.qpid.server.message.InstanceProperties; +import org.apache.qpid.server.message.MessageInstance; import org.apache.qpid.server.message.MessageReference; import org.apache.qpid.server.message.ServerMessage; import org.apache.qpid.server.model.UUIDGenerator; import org.apache.qpid.server.plugin.ExchangeType; import org.apache.qpid.server.queue.AMQQueue; import org.apache.qpid.server.queue.BaseQueue; +import org.apache.qpid.server.queue.QueueEntry; import org.apache.qpid.server.store.DurableConfigurationStoreHelper; import org.apache.qpid.server.txn.ServerTransaction; +import org.apache.qpid.server.util.Action; import org.apache.qpid.server.virtualhost.VirtualHost; import java.util.Collection; @@ -66,7 +70,7 @@ public abstract class AbstractExchange implements Exchange private VirtualHost _virtualHost; - private final List<Task> _closeTaskList = new CopyOnWriteArrayList<Task>(); + private final List<Action<Exchange>> _closeTaskList = new CopyOnWriteArrayList<Action<Exchange>>(); /** * Whether the exchange is automatically deleted once all queues have detached from it @@ -138,6 +142,12 @@ public abstract class AbstractExchange implements Exchange if(_closed.compareAndSet(false,true)) { + List<Binding> bindings = new ArrayList<Binding>(_bindings); + for(Binding binding : bindings) + { + removeBinding(binding); + } + if(_alternateExchange != null) { _alternateExchange.removeReference(this); @@ -145,9 +155,9 @@ public abstract class AbstractExchange implements Exchange CurrentActor.get().message(_logSubject, ExchangeMessages.DELETED()); - for(Task task : _closeTaskList) + for(Action<Exchange> task : _closeTaskList) { - task.onClose(this); + task.performAction(this); } _closeTaskList.clear(); } @@ -300,12 +310,12 @@ public abstract class AbstractExchange implements Exchange return !_referrers.isEmpty(); } - public void addCloseTask(final Task task) + public void addCloseTask(final Action<Exchange> task) { _closeTaskList.add(task); } - public void removeCloseTask(final Task task) + public void removeCloseTask(final Action<Exchange> task) { _closeTaskList.remove(task); } @@ -421,7 +431,7 @@ public abstract class AbstractExchange implements Exchange public final int send(final ServerMessage message, final InstanceProperties instanceProperties, final ServerTransaction txn, - final BaseQueue.PostEnqueueAction postEnqueueAction) + final Action<MessageInstance<? extends Consumer>> postEnqueueAction) { List<? extends BaseQueue> queues = route(message, instanceProperties); @@ -579,8 +589,6 @@ public abstract class AbstractExchange implements Exchange { doRemoveBinding(b); queue.removeBinding(b); - removeCloseTask(b); - queue.removeQueueDeleteTask(b); if (b.isDurable()) { @@ -659,8 +667,6 @@ public abstract class AbstractExchange implements Exchange DurableConfigurationStoreHelper.createBinding(_virtualHost.getDurableConfigurationStore(), b); } - queue.addQueueDeleteTask(b); - addCloseTask(b); queue.addBinding(b); doAddBinding(b); b.logCreation(); @@ -673,7 +679,7 @@ public abstract class AbstractExchange implements Exchange } } - private final class BindingImpl extends Binding implements AMQQueue.Task, Task + private final class BindingImpl extends Binding { private final BindingLogSubject _logSubject; //TODO : persist creation time @@ -689,12 +695,6 @@ public abstract class AbstractExchange implements Exchange } - - public void doTask(final AMQQueue queue) throws AMQException - { - removeBinding(this); - } - public void onClose(final Exchange exchange) throws AMQSecurityException, AMQInternalException { removeBinding(this); @@ -729,11 +729,6 @@ public abstract class AbstractExchange implements Exchange } - public static interface Task - { - public void onClose(Exchange exchange) throws AMQSecurityException, AMQInternalException; - } - } diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/exchange/DefaultExchange.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/exchange/DefaultExchange.java index 71d0f8b4dd..33c5218b4c 100644 --- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/exchange/DefaultExchange.java +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/exchange/DefaultExchange.java @@ -32,18 +32,22 @@ import org.apache.qpid.AMQInternalException; import org.apache.qpid.AMQSecurityException; import org.apache.qpid.exchange.ExchangeDefaults; import org.apache.qpid.server.binding.Binding; +import org.apache.qpid.server.consumer.Consumer; import org.apache.qpid.server.logging.LogSubject; import org.apache.qpid.server.logging.actors.CurrentActor; import org.apache.qpid.server.logging.messages.ExchangeMessages; import org.apache.qpid.server.message.InstanceProperties; +import org.apache.qpid.server.message.MessageInstance; import org.apache.qpid.server.message.MessageReference; import org.apache.qpid.server.message.ServerMessage; import org.apache.qpid.server.model.UUIDGenerator; import org.apache.qpid.server.plugin.ExchangeType; import org.apache.qpid.server.queue.AMQQueue; import org.apache.qpid.server.queue.BaseQueue; +import org.apache.qpid.server.queue.QueueEntry; import org.apache.qpid.server.queue.QueueRegistry; import org.apache.qpid.server.txn.ServerTransaction; +import org.apache.qpid.server.util.Action; import org.apache.qpid.server.virtualhost.VirtualHost; public class DefaultExchange implements Exchange @@ -334,7 +338,7 @@ public class DefaultExchange implements Exchange public final int send(final ServerMessage message, final InstanceProperties instanceProperties, final ServerTransaction txn, - final BaseQueue.PostEnqueueAction postEnqueueAction) + final Action<MessageInstance<? extends Consumer>> postEnqueueAction) { final AMQQueue q = _virtualHost.getQueue(message.getRoutingKey()); if(q == null) diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/exchange/Exchange.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/exchange/Exchange.java index 18e912e972..6d83fdb2a1 100644 --- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/exchange/Exchange.java +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/exchange/Exchange.java @@ -24,20 +24,16 @@ import org.apache.qpid.AMQException; import org.apache.qpid.AMQInternalException; import org.apache.qpid.AMQSecurityException; import org.apache.qpid.server.binding.Binding; -import org.apache.qpid.server.message.InstanceProperties; -import org.apache.qpid.server.message.ServerMessage; +import org.apache.qpid.server.message.MessageDestination; import org.apache.qpid.server.plugin.ExchangeType; import org.apache.qpid.server.queue.AMQQueue; -import org.apache.qpid.server.queue.BaseQueue; -import org.apache.qpid.server.txn.ServerTransaction; import org.apache.qpid.server.virtualhost.VirtualHost; import java.util.Collection; -import java.util.List; import java.util.Map; import java.util.UUID; -public interface Exchange extends ExchangeReferrer +public interface Exchange extends ExchangeReferrer, MessageDestination { void initialise(UUID id, VirtualHost host, String name, boolean durable, boolean autoDelete) throws AMQException; @@ -95,19 +91,6 @@ public interface Exchange extends ExchangeReferrer void close() throws AMQException; /** - * Routes a message - * @param message the message to be routed - * @param instanceProperties the instance properties - * @param txn the transaction to enqueue within - * @param postEnqueueAction action to perform on the result of every enqueue (may be null) - * @return the number of queues in which the message was enqueued performed - */ - int send(ServerMessage message, - InstanceProperties instanceProperties, - ServerTransaction txn, - BaseQueue.PostEnqueueAction postEnqueueAction); - - /** * Determines whether a message would be isBound to a particular queue using a specific routing key and arguments * @param bindingKey * @param arguments diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/logging/actors/CurrentActor.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/logging/actors/CurrentActor.java index 6251471139..91d9ef7dbc 100644 --- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/logging/actors/CurrentActor.java +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/logging/actors/CurrentActor.java @@ -35,7 +35,6 @@ import org.apache.qpid.server.logging.LogSubject; * 2) We can set new actors at the point we have enough information. i.e. * - Set a low level ConnectionActor when processing bytes from the wire. * - Set a ChannelActor when we are processing the frame - * - Set a SubscriptionActor when we are handling the subscription. * <p/> * The code performing the logging need not worry about what type of actor is * currently set so can perform its logging. The resulting log entry though will diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/logging/actors/GenericActor.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/logging/actors/GenericActor.java index 0e418a95e2..4cf2bd4508 100644 --- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/logging/actors/GenericActor.java +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/logging/actors/GenericActor.java @@ -42,6 +42,24 @@ public class GenericActor extends AbstractActor _defaultMessageLogger = defaultMessageLogger; } + public GenericActor(final String logSubject) + { + this(new LogSubject() + { + @Override + public String toLogString() + { + return logSubject; + } + }); + } + + + public GenericActor(LogSubject logSubject) + { + this(logSubject, CurrentActor.get().getRootMessageLogger()); + } + public GenericActor(LogSubject logSubject, RootMessageLogger rootLogger) { super(rootLogger); @@ -53,6 +71,11 @@ public class GenericActor extends AbstractActor return _logSubject.toLogString(); } + public LogSubject getLogSubject() + { + return _logSubject; + } + public static LogActor getInstance(final String logMessage, RootMessageLogger rootLogger) { return new GenericActor(new LogSubject() diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/logging/subjects/SubscriptionLogSubject.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/logging/subjects/SubscriptionLogSubject.java deleted file mode 100644 index 0292fe3506..0000000000 --- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/logging/subjects/SubscriptionLogSubject.java +++ /dev/null @@ -1,55 +0,0 @@ -/* - * - * 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.logging.subjects; - -import org.apache.qpid.server.subscription.Subscription; - -import static org.apache.qpid.server.logging.subjects.LogSubjectFormat.SUBSCRIPTION_FORMAT; - -import java.text.MessageFormat; - -public class SubscriptionLogSubject extends AbstractLogSubject -{ - - /** - * Create an QueueLogSubject that Logs in the following format. - * - * @param subscription - */ - public SubscriptionLogSubject(Subscription subscription) - { - // Delegate the formatting of the Queue to the QueueLogSubject. So final - // log string format is: - // [ sub:<id>(vh(<vhost>)/qu(<queue>)) ] - - String queueString = new QueueLogSubject(subscription.getQueue()).toLogString(); - - setLogString("[" + MessageFormat.format(SUBSCRIPTION_FORMAT, - subscription.getSubscriptionID()) - + "(" - // queueString is [vh(/{0})/qu({1}) ] so need to trim - // ^ ^^ - + queueString.substring(1,queueString.length() - 3) - + ")" - + "] "); - - } -} diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/message/MessageDestination.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/message/MessageDestination.java new file mode 100644 index 0000000000..967c629749 --- /dev/null +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/message/MessageDestination.java @@ -0,0 +1,44 @@ +/* + * + * 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.message; + +import org.apache.qpid.server.consumer.Consumer; +import org.apache.qpid.server.txn.ServerTransaction; +import org.apache.qpid.server.util.Action; + +public interface MessageDestination extends MessageNode +{ + + public String getName(); + + /** + * Routes a message + * @param message the message to be routed + * @param instanceProperties the instance properties + * @param txn the transaction to enqueue within + * @param postEnqueueAction action to perform on the result of every enqueue (may be null) + * @return the number of queues in which the message was enqueued performed + */ + int send(ServerMessage message, + InstanceProperties instanceProperties, + ServerTransaction txn, + Action<MessageInstance<? extends Consumer>> postEnqueueAction); +} diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/message/MessageInstance.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/message/MessageInstance.java index afd7ff0269..97cb66cce4 100644 --- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/message/MessageInstance.java +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/message/MessageInstance.java @@ -21,9 +21,178 @@ package org.apache.qpid.server.message; -public interface MessageInstance +import org.apache.qpid.AMQException; +import org.apache.qpid.server.filter.Filterable; +import org.apache.qpid.server.queue.QueueEntry; +import org.apache.qpid.server.store.TransactionLogResource; +import org.apache.qpid.server.consumer.Consumer; +import org.apache.qpid.server.txn.ServerTransaction; +import org.apache.qpid.server.util.Action; +import org.apache.qpid.server.util.StateChangeListener; + +public interface MessageInstance<C extends Consumer> { + + /** + * Number of times this queue entry has been delivered. + * + * @return delivery count + */ + int getDeliveryCount(); + + void incrementDeliveryCount(); + + void decrementDeliveryCount(); + + void addStateChangeListener(StateChangeListener<MessageInstance<C>, State> listener); + + boolean removeStateChangeListener(StateChangeListener<MessageInstance<C>, State> listener); + + boolean acquiredByConsumer(); + + boolean isAcquiredBy(C consumer); + + void setRedelivered(); + + boolean isRedelivered(); + + C getDeliveredConsumer(); + + void reject(); + + boolean isRejectedBy(C consumer); + + boolean getDeliveredToConsumer(); + + boolean expired() throws AMQException; + + boolean acquire(C sub); + + int getMaximumDeliveryCount(); + + int routeToAlternate(Action<MessageInstance<? extends Consumer>> action, ServerTransaction txn); + + Filterable asFilterable(); + + public static enum State + { + AVAILABLE, + ACQUIRED, + DEQUEUED, + DELETED + } + + public abstract class EntryState + { + private EntryState() + { + } + + public abstract State getState(); + + /** + * Returns true if state is either DEQUEUED or DELETED. + * + * @return true if state is either DEQUEUED or DELETED. + */ + public boolean isDispensed() + { + State currentState = getState(); + return currentState == State.DEQUEUED || currentState == State.DELETED; + } + } + + + public final class AvailableState extends EntryState + { + + public State getState() + { + return State.AVAILABLE; + } + + public String toString() + { + return getState().name(); + } + } + + + public final class DequeuedState extends EntryState + { + + public State getState() + { + return State.DEQUEUED; + } + + public String toString() + { + return getState().name(); + } + } + + + public final class DeletedState extends EntryState + { + + public State getState() + { + return State.DELETED; + } + + public String toString() + { + return getState().name(); + } + } + + public final class NonConsumerAcquiredState extends EntryState + { + public State getState() + { + return State.ACQUIRED; + } + + public String toString() + { + return getState().name(); + } + } + + public final class ConsumerAcquiredState<C extends Consumer> extends EntryState + { + private final C _consumer; + + public ConsumerAcquiredState(C consumer) + { + _consumer = consumer; + } + + + public State getState() + { + return State.ACQUIRED; + } + + public C getConsumer() + { + return _consumer; + } + + public String toString() + { + return "{" + getState().name() + " : " + _consumer +"}"; + } + } + + + final static EntryState AVAILABLE_STATE = new AvailableState(); + final static EntryState DELETED_STATE = new DeletedState(); + final static EntryState DEQUEUED_STATE = new DequeuedState(); + final static EntryState NON_CONSUMER_ACQUIRED_STATE = new NonConsumerAcquiredState(); + boolean isAvailable(); boolean acquire(); @@ -32,6 +201,8 @@ public interface MessageInstance void release(); + boolean resend() throws AMQException; + void delete(); boolean isDeleted(); @@ -39,4 +210,6 @@ public interface MessageInstance ServerMessage getMessage(); InstanceProperties getInstanceProperties(); + + TransactionLogResource getOwningResource(); } diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/message/MessageNode.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/message/MessageNode.java new file mode 100644 index 0000000000..f4b751d2fd --- /dev/null +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/message/MessageNode.java @@ -0,0 +1,26 @@ +/* + * + * 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.message; + +public interface MessageNode +{ + String getName(); +} diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/message/MessageSource.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/message/MessageSource.java new file mode 100644 index 0000000000..06ff76f103 --- /dev/null +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/message/MessageSource.java @@ -0,0 +1,105 @@ +/* + * + * 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.message; + +import org.apache.qpid.AMQException; +import org.apache.qpid.server.consumer.Consumer; +import org.apache.qpid.server.consumer.ConsumerTarget; +import org.apache.qpid.server.filter.FilterManager; +import org.apache.qpid.server.protocol.AMQSessionModel; +import org.apache.qpid.server.queue.AMQQueue; +import org.apache.qpid.server.security.AuthorizationHolder; +import org.apache.qpid.server.store.TransactionLogResource; + +import java.util.Collection; +import java.util.EnumSet; + +public interface MessageSource<C extends Consumer> extends TransactionLogResource, MessageNode +{ + C addConsumer(ConsumerTarget target, FilterManager filters, + Class<? extends ServerMessage> messageClass, + String consumerName, EnumSet<Consumer.Option> options) throws AMQException; + + Collection<C> getConsumers(); + + void addConsumerRegistrationListener(ConsumerRegistrationListener listener); + + void removeConsumerRegistrationListener(ConsumerRegistrationListener listener); + + AuthorizationHolder getAuthorizationHolder(); + + void setAuthorizationHolder(AuthorizationHolder principalHolder); + + void setExclusiveOwningSession(AMQSessionModel owner); + + AMQSessionModel getExclusiveOwningSession(); + + boolean isExclusive(); + + interface ConsumerRegistrationListener + { + void consumerAdded(AMQQueue queue, Consumer consumer); + void consumerRemoved(AMQQueue queue, Consumer consumer); + } + + /** + * ExistingExclusiveConsumer signals a failure to create a consumer, because an exclusive consumer + * already exists. + * + * <p/><table id="crc"><caption>CRC Card</caption> + * <tr><th> Responsibilities <th> Collaborations + * <tr><td> Represent failure to create a consumer, because an exclusive consumer already exists. + * </table> + * + * @todo Not an AMQP exception as no status code. + * + * @todo Move to top level, used outside this class. + */ + static final class ExistingExclusiveConsumer extends AMQException + { + + public ExistingExclusiveConsumer() + { + super(""); + } + } + + /** + * ExistingConsumerPreventsExclusive signals a failure to create an exclusive consumer, as a consumer + * already exists. + * + * <p/><table id="crc"><caption>CRC Card</caption> + * <tr><th> Responsibilities <th> Collaborations + * <tr><td> Represent failure to create an exclusive consumer, as a consumer already exists. + * </table> + * + * @todo Not an AMQP exception as no status code. + * + * @todo Move to top level, used outside this class. + */ + static final class ExistingConsumerPreventsExclusive extends AMQException + { + public ExistingConsumerPreventsExclusive() + { + super(""); + } + } +} diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/Session.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/Session.java index e813d0c129..355a1cf3b1 100644 --- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/Session.java +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/Session.java @@ -77,6 +77,6 @@ public interface Session extends ConfiguredObject CHANNEL_ID, PRODUCER_FLOW_BLOCKED)); - Collection<Consumer> getSubscriptions(); + Collection<Consumer> getConsumers(); Collection<Publisher> getPublishers(); } diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/VirtualHost.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/VirtualHost.java index ae07005679..2b5176aa65 100644 --- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/VirtualHost.java +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/VirtualHost.java @@ -20,6 +20,7 @@ */ package org.apache.qpid.server.model; +import org.apache.qpid.server.message.MessageInstance; import org.apache.qpid.server.queue.QueueEntry; import org.apache.qpid.server.security.SecurityManager; import org.apache.qpid.server.store.MessageStore; @@ -144,11 +145,11 @@ public interface VirtualHost extends ConfiguredObject public static interface Transaction { - void dequeue(QueueEntry entry); + void dequeue(MessageInstance entry); - void copy(QueueEntry entry, Queue queue); + void copy(MessageInstance entry, Queue queue); - void move(QueueEntry entry, Queue queue); + void move(MessageInstance entry, Queue queue); } diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/ConsumerAdapter.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/ConsumerAdapter.java index 696c59783e..cf6874030b 100644 --- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/ConsumerAdapter.java +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/ConsumerAdapter.java @@ -22,33 +22,32 @@ package org.apache.qpid.server.model.adapter; import java.util.Map; import org.apache.qpid.server.model.ConfiguredObject; -import org.apache.qpid.server.model.Consumer; import org.apache.qpid.server.model.LifetimePolicy; import org.apache.qpid.server.model.State; import org.apache.qpid.server.model.Statistics; import org.apache.qpid.server.model.UUIDGenerator; -import org.apache.qpid.server.subscription.Subscription; +import org.apache.qpid.server.consumer.Consumer; import java.security.AccessControlException; import java.util.Collection; import java.util.Collections; -public class ConsumerAdapter extends AbstractAdapter implements Consumer +public class ConsumerAdapter extends AbstractAdapter implements org.apache.qpid.server.model.Consumer { - private final Subscription _subscription; + private final Consumer _consumer; private final QueueAdapter _queue; private final SessionAdapter _session; private final ConsumerStatistics _statistics; public ConsumerAdapter(final QueueAdapter queueAdapter, final SessionAdapter sessionAdapter, - final Subscription subscription) + final Consumer consumer) { super(UUIDGenerator.generateConsumerUUID(queueAdapter.getVirtualHost().getName(), queueAdapter.getName(), - subscription.getSessionModel().getConnectionModel().getRemoteAddressString(), - String.valueOf(subscription.getSessionModel().getChannelId()), - subscription.getConsumerName()), queueAdapter.getTaskExecutor()); - _subscription = subscription; + consumer.getSessionModel().getConnectionModel().getRemoteAddressString(), + String.valueOf(consumer.getSessionModel().getChannelId()), + consumer.getName()), queueAdapter.getTaskExecutor()); + _consumer = consumer; _queue = queueAdapter; _session = sessionAdapter; _statistics = new ConsumerStatistics(); @@ -57,7 +56,7 @@ public class ConsumerAdapter extends AbstractAdapter implements Consumer public String getName() { - return _subscription.getConsumerName(); + return _consumer.getName(); } public String setName(final String currentName, final String desiredName) @@ -107,7 +106,7 @@ public class ConsumerAdapter extends AbstractAdapter implements Consumer @Override public Collection<String> getAttributeNames() { - return Consumer.AVAILABLE_ATTRIBUTES; + return org.apache.qpid.server.model.Consumer.AVAILABLE_ATTRIBUTES; } @Override @@ -147,7 +146,7 @@ public class ConsumerAdapter extends AbstractAdapter implements Consumer } else if(DISTRIBUTION_MODE.equals(name)) { - return _subscription.acquires() ? "MOVE" : "COPY"; + return _consumer.acquires() ? "MOVE" : "COPY"; } else if(SETTLEMENT_MODE.equals(name)) { @@ -197,11 +196,11 @@ public class ConsumerAdapter extends AbstractAdapter implements Consumer { if(name.equals(BYTES_OUT)) { - return _subscription.getBytesOut(); + return _consumer.getBytesOut(); } else if(name.equals(MESSAGES_OUT)) { - return _subscription.getMessagesOut(); + return _consumer.getMessagesOut(); } else if(name.equals(STATE_CHANGED)) { @@ -209,11 +208,11 @@ public class ConsumerAdapter extends AbstractAdapter implements Consumer } else if(name.equals(UNACKNOWLEDGED_BYTES)) { - return _subscription.getUnacknowledgedBytes(); + return _consumer.getUnacknowledgedBytes(); } else if(name.equals(UNACKNOWLEDGED_MESSAGES)) { - return _subscription.getUnacknowledgedMessages(); + return _consumer.getUnacknowledgedMessages(); } return null; // TODO - Implement } diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/QueueAdapter.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/QueueAdapter.java index 074f7c243b..d59b13902b 100644 --- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/QueueAdapter.java +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/QueueAdapter.java @@ -35,7 +35,6 @@ import org.apache.qpid.server.binding.Binding; import org.apache.qpid.server.configuration.IllegalConfigurationException; import org.apache.qpid.server.model.ConfiguredObject; import org.apache.qpid.server.model.ConfiguredObjectFinder; -import org.apache.qpid.server.model.Consumer; import org.apache.qpid.server.model.Exchange; import org.apache.qpid.server.model.IllegalStateTransitionException; import org.apache.qpid.server.model.LifetimePolicy; @@ -47,10 +46,11 @@ import org.apache.qpid.server.protocol.AMQConnectionModel; import org.apache.qpid.server.protocol.AMQSessionModel; import org.apache.qpid.server.queue.*; import org.apache.qpid.server.store.DurableConfigurationStoreHelper; -import org.apache.qpid.server.subscription.Subscription; +import org.apache.qpid.server.consumer.Consumer; import org.apache.qpid.server.util.MapValueConverter; -final class QueueAdapter extends AbstractAdapter implements Queue, AMQQueue.SubscriptionRegistrationListener, AMQQueue.NotificationListener +final class QueueAdapter extends AbstractAdapter implements Queue, + AMQQueue.ConsumerRegistrationListener, AMQQueue.NotificationListener { @SuppressWarnings("serial") static final Map<String, Type> ATTRIBUTE_TYPES = Collections.unmodifiableMap(new HashMap<String, Type>(){{ @@ -69,8 +69,8 @@ final class QueueAdapter extends AbstractAdapter implements Queue, AMQQueue.Subs private final AMQQueue _queue; private final Map<Binding, BindingAdapter> _bindingAdapters = new HashMap<Binding, BindingAdapter>(); - private Map<org.apache.qpid.server.subscription.Subscription, ConsumerAdapter> _consumerAdapters = - new HashMap<org.apache.qpid.server.subscription.Subscription, ConsumerAdapter>(); + private Map<Consumer, ConsumerAdapter> _consumerAdapters = + new HashMap<Consumer, ConsumerAdapter>(); private final VirtualHostAdapter _vhost; @@ -84,7 +84,7 @@ final class QueueAdapter extends AbstractAdapter implements Queue, AMQQueue.Subs addParent(org.apache.qpid.server.model.VirtualHost.class, virtualHostAdapter); _queue = queue; - _queue.addSubscriptionRegistrationListener(this); + _queue.addConsumerRegistrationListener(this); populateConsumers(); _statistics = new QueueStatisticsAdapter(queue); _queue.setNotificationListener(this); @@ -124,21 +124,21 @@ final class QueueAdapter extends AbstractAdapter implements Queue, AMQQueue.Subs private void populateConsumers() { - Collection<org.apache.qpid.server.subscription.Subscription> actualSubscriptions = _queue.getConsumers(); + Collection<Consumer> actualConsumers = _queue.getConsumers(); synchronized (_consumerAdapters) { - Iterator<org.apache.qpid.server.subscription.Subscription> iter = _consumerAdapters.keySet().iterator(); - for(org.apache.qpid.server.subscription.Subscription subscription : actualSubscriptions) + Iterator<Consumer> iter = _consumerAdapters.keySet().iterator(); + for(Consumer consumer : actualConsumers) { - if(!_consumerAdapters.containsKey(subscription)) + if(!_consumerAdapters.containsKey(consumer)) { - SessionAdapter sessionAdapter = getSessionAdapter(subscription.getSessionModel()); - ConsumerAdapter adapter = new ConsumerAdapter(this, sessionAdapter, subscription); - _consumerAdapters.put(subscription, adapter); + SessionAdapter sessionAdapter = getSessionAdapter(consumer.getSessionModel()); + ConsumerAdapter adapter = new ConsumerAdapter(this, sessionAdapter, consumer); + _consumerAdapters.put(consumer, adapter); if (sessionAdapter != null) { // Register ConsumerAdapter with the SessionAdapter. - sessionAdapter.subscriptionRegistered(subscription, adapter); + sessionAdapter.consumerRegistered(consumer, adapter); } } } @@ -153,11 +153,11 @@ final class QueueAdapter extends AbstractAdapter implements Queue, AMQQueue.Subs } } - public Collection<Consumer> getConsumers() + public Collection<org.apache.qpid.server.model.Consumer> getConsumers() { synchronized (_consumerAdapters) { - return new ArrayList<Consumer>(_consumerAdapters.values()); + return new ArrayList<org.apache.qpid.server.model.Consumer>(_consumerAdapters.values()); } } @@ -502,7 +502,7 @@ final class QueueAdapter extends AbstractAdapter implements Queue, AMQQueue.Subs @Override public <C extends ConfiguredObject> Collection<C> getChildren(Class<C> clazz) { - if(clazz == Consumer.class) + if(clazz == org.apache.qpid.server.model.Consumer.class) { return (Collection<C>) getConsumers(); } @@ -587,19 +587,19 @@ final class QueueAdapter extends AbstractAdapter implements Queue, AMQQueue.Subs return _queue; } - public void subscriptionRegistered(final AMQQueue queue, final Subscription subscription) + public void consumerAdded(final AMQQueue queue, final Consumer consumer) { ConsumerAdapter adapter = null; synchronized (_consumerAdapters) { - if(!_consumerAdapters.containsKey(subscription)) + if(!_consumerAdapters.containsKey(consumer)) { - SessionAdapter sessionAdapter = getSessionAdapter(subscription.getSessionModel()); - adapter = new ConsumerAdapter(this, sessionAdapter, subscription); - _consumerAdapters.put(subscription, adapter); + SessionAdapter sessionAdapter = getSessionAdapter(consumer.getSessionModel()); + adapter = new ConsumerAdapter(this, sessionAdapter, consumer); + _consumerAdapters.put(consumer, adapter); if (sessionAdapter != null) { // Register ConsumerAdapter with the SessionAdapter. - sessionAdapter.subscriptionRegistered(subscription, adapter); + sessionAdapter.consumerRegistered(consumer, adapter); } } } @@ -609,20 +609,20 @@ final class QueueAdapter extends AbstractAdapter implements Queue, AMQQueue.Subs } } - public void subscriptionUnregistered(final AMQQueue queue, final Subscription subscription) + public void consumerRemoved(final AMQQueue queue, final Consumer consumer) { ConsumerAdapter adapter = null; synchronized (_consumerAdapters) { - adapter = _consumerAdapters.remove(subscription); + adapter = _consumerAdapters.remove(consumer); } if(adapter != null) { - SessionAdapter sessionAdapter = getSessionAdapter(subscription.getSessionModel()); + SessionAdapter sessionAdapter = getSessionAdapter(consumer.getSessionModel()); if (sessionAdapter != null) { // Unregister ConsumerAdapter with the SessionAdapter. - sessionAdapter.subscriptionUnregistered(subscription); + sessionAdapter.consumerUnregistered(consumer); } childRemoved(adapter); } diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/SessionAdapter.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/SessionAdapter.java index 31ce7e56fd..6b76eeefa0 100644 --- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/SessionAdapter.java +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/SessionAdapter.java @@ -34,9 +34,8 @@ import org.apache.qpid.server.model.Publisher; import org.apache.qpid.server.model.Session; import org.apache.qpid.server.model.State; import org.apache.qpid.server.model.Statistics; -import org.apache.qpid.server.model.Consumer; import org.apache.qpid.server.model.UUIDGenerator; -import org.apache.qpid.server.subscription.Subscription; +import org.apache.qpid.server.consumer.Consumer; import org.apache.qpid.server.configuration.updater.TaskExecutor; import org.apache.qpid.server.protocol.AMQSessionModel; @@ -47,7 +46,7 @@ final class SessionAdapter extends AbstractAdapter implements Session private AMQSessionModel _session; private SessionStatistics _statistics; - private Map<Subscription, ConsumerAdapter> _consumerAdapters = new HashMap<Subscription, ConsumerAdapter>(); + private Map<Consumer, ConsumerAdapter> _consumerAdapters = new HashMap<Consumer, ConsumerAdapter>(); public SessionAdapter(final AMQSessionModel session, TaskExecutor taskExecutor) { @@ -56,11 +55,11 @@ final class SessionAdapter extends AbstractAdapter implements Session _statistics = new SessionStatistics(); } - public Collection<Consumer> getSubscriptions() + public Collection<org.apache.qpid.server.model.Consumer> getConsumers() { synchronized (_consumerAdapters) { - return new ArrayList<Consumer>(_consumerAdapters.values()); + return new ArrayList<org.apache.qpid.server.model.Consumer>(_consumerAdapters.values()); } } @@ -119,29 +118,29 @@ final class SessionAdapter extends AbstractAdapter implements Session } /** - * Register a ConsumerAdapter (Subscription) with this Session keyed by the Subscription. - * @param subscription the org.apache.qpid.server.subscription.Subscription used to key the ConsumerAdapter. + * Register a ConsumerAdapter with this Session keyed by the Consumer. + * @param consumer the org.apache.qpid.server.consumer.Consumer used to key the ConsumerAdapter. * @param adapter the registered ConsumerAdapter. */ - void subscriptionRegistered(Subscription subscription, ConsumerAdapter adapter) + void consumerRegistered(Consumer consumer, ConsumerAdapter adapter) { synchronized (_consumerAdapters) { - _consumerAdapters.put(subscription, adapter); + _consumerAdapters.put(consumer, adapter); } childAdded(adapter); } /** - * Unregister a ConsumerAdapter (Subscription) with this Session keyed by the Subscription. - * @param subscription the org.apache.qpid.server.subscription.Subscription used to key the ConsumerAdapter. + * Unregister a ConsumerAdapter with this Session keyed by the Consumer. + * @param consumer the org.apache.qpid.server.consumer.Consumer used to key the ConsumerAdapter. */ - void subscriptionUnregistered(Subscription subscription) + void consumerUnregistered(Consumer consumer) { ConsumerAdapter adapter = null; synchronized (_consumerAdapters) { - adapter = _consumerAdapters.remove(subscription); + adapter = _consumerAdapters.remove(consumer); } if (adapter != null) { @@ -188,9 +187,9 @@ final class SessionAdapter extends AbstractAdapter implements Session @Override public <C extends ConfiguredObject> Collection<C> getChildren(Class<C> clazz) { - if(clazz == Consumer.class) + if(clazz == org.apache.qpid.server.model.Consumer.class) { - return (Collection<C>) getSubscriptions(); + return (Collection<C>) getConsumers(); } else if(clazz == Publisher.class) { diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/VirtualHostAdapter.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/VirtualHostAdapter.java index 300b6e6618..f2ce20c74d 100644 --- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/VirtualHostAdapter.java +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/VirtualHostAdapter.java @@ -44,6 +44,7 @@ import org.apache.qpid.AMQException; import org.apache.qpid.server.configuration.IllegalConfigurationException; import org.apache.qpid.server.configuration.VirtualHostConfiguration; import org.apache.qpid.server.configuration.XmlConfigurationUtilities.MyConfiguration; +import org.apache.qpid.server.message.MessageInstance; import org.apache.qpid.server.message.ServerMessage; import org.apache.qpid.server.model.Broker; import org.apache.qpid.server.model.ConfiguredObject; @@ -759,11 +760,11 @@ public final class VirtualHostAdapter extends AbstractAdapter implements Virtual op.withinTransaction(new Transaction() { - public void dequeue(final QueueEntry entry) + public void dequeue(final MessageInstance entry) { if(entry.acquire()) { - txn.dequeue(entry.getQueue(), entry.getMessage(), new ServerTransaction.Action() + txn.dequeue(entry.getOwningResource(), entry.getMessage(), new ServerTransaction.Action() { public void postCommit() { @@ -777,7 +778,7 @@ public final class VirtualHostAdapter extends AbstractAdapter implements Virtual } } - public void copy(QueueEntry entry, Queue queue) + public void copy(MessageInstance entry, Queue queue) { final ServerMessage message = entry.getMessage(); final AMQQueue toQueue = ((QueueAdapter)queue).getAMQQueue(); @@ -788,7 +789,7 @@ public final class VirtualHostAdapter extends AbstractAdapter implements Virtual { try { - toQueue.enqueue(message); + toQueue.enqueue(message, null); } catch(AMQException e) { @@ -803,7 +804,7 @@ public final class VirtualHostAdapter extends AbstractAdapter implements Virtual } - public void move(final QueueEntry entry, Queue queue) + public void move(final MessageInstance entry, Queue queue) { final ServerMessage message = entry.getMessage(); final AMQQueue toQueue = ((QueueAdapter)queue).getAMQQueue(); @@ -817,7 +818,7 @@ public final class VirtualHostAdapter extends AbstractAdapter implements Virtual { try { - toQueue.enqueue(message); + toQueue.enqueue(message, null); } catch (AMQException e) { @@ -830,7 +831,7 @@ public final class VirtualHostAdapter extends AbstractAdapter implements Virtual entry.release(); } }); - txn.dequeue(entry.getQueue(), message, + txn.dequeue(entry.getOwningResource(), message, new ServerTransaction.Action() { diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/logging/actors/SubscriptionActor.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/plugin/SystemNodeCreator.java index 906cd6db3a..a117524254 100644 --- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/logging/actors/SubscriptionActor.java +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/plugin/SystemNodeCreator.java @@ -18,29 +18,20 @@ * under the License. * */ -package org.apache.qpid.server.logging.actors; +package org.apache.qpid.server.plugin; -import org.apache.qpid.server.logging.RootMessageLogger; -import org.apache.qpid.server.logging.subjects.SubscriptionLogSubject; -import org.apache.qpid.server.subscription.Subscription; +import org.apache.qpid.server.message.MessageNode; +import org.apache.qpid.server.virtualhost.VirtualHost; -/** - * The subscription actor provides formatted logging for actions that are - * performed by the subscription. Such as STATE changes. - */ -public class SubscriptionActor extends AbstractActor +public interface SystemNodeCreator extends Pluggable { - private SubscriptionLogSubject _logSubject; - - public SubscriptionActor(RootMessageLogger logger, Subscription subscription) + interface SystemNodeRegistry { - super(logger); + void registerSystemNode(MessageNode node); + void removeSystemNode(MessageNode node); - _logSubject = new SubscriptionLogSubject(subscription); + VirtualHost getVirtualHost(); } - public String getLogMessage() - { - return _logSubject.toLogString(); - } + void register(SystemNodeRegistry registry); } diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/protocol/CapacityChecker.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/protocol/CapacityChecker.java new file mode 100644 index 0000000000..0ba3095243 --- /dev/null +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/protocol/CapacityChecker.java @@ -0,0 +1,26 @@ +/* + * + * 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.protocol; + +public interface CapacityChecker +{ + void checkCapacity(AMQSessionModel channel); +} diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/AMQQueue.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/AMQQueue.java index 0cddd1ed3b..62927edc29 100644 --- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/AMQQueue.java +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/AMQQueue.java @@ -26,19 +26,20 @@ import org.apache.qpid.server.configuration.QueueConfiguration; import org.apache.qpid.server.exchange.Exchange; import org.apache.qpid.server.exchange.ExchangeReferrer; import org.apache.qpid.server.logging.LogSubject; -import org.apache.qpid.server.protocol.AMQSessionModel; -import org.apache.qpid.server.security.AuthorizationHolder; -import org.apache.qpid.server.store.TransactionLogResource; -import org.apache.qpid.server.subscription.Subscription; +import org.apache.qpid.server.message.MessageDestination; +import org.apache.qpid.server.message.MessageSource; +import org.apache.qpid.server.protocol.CapacityChecker; +import org.apache.qpid.server.consumer.Consumer; +import org.apache.qpid.server.util.Action; import org.apache.qpid.server.virtualhost.VirtualHost; import java.util.Collection; import java.util.List; import java.util.Set; -public interface AMQQueue extends Comparable<AMQQueue>, ExchangeReferrer, TransactionLogResource, BaseQueue +public interface AMQQueue<C extends Consumer> extends Comparable<AMQQueue<C>>, ExchangeReferrer, BaseQueue, + MessageSource<C>, CapacityChecker, MessageDestination { - String getName(); public interface NotificationListener { @@ -65,45 +66,20 @@ public interface AMQQueue extends Comparable<AMQQueue>, ExchangeReferrer, Transa long getTotalEnqueueCount(); - public interface Context - { - QueueEntry getLastSeenEntry(); - } - void setNoLocal(boolean b); boolean isAutoDelete(); String getOwner(); - AuthorizationHolder getAuthorizationHolder(); - void setAuthorizationHolder(AuthorizationHolder principalHolder); - - void setExclusiveOwningSession(AMQSessionModel owner); - AMQSessionModel getExclusiveOwningSession(); VirtualHost getVirtualHost(); - void registerSubscription(final Subscription subscription, final boolean exclusive) throws AMQException; - - void unregisterSubscription(final Subscription subscription) throws AMQException; - - Collection<Subscription> getConsumers(); - - interface SubscriptionRegistrationListener - { - void subscriptionRegistered(AMQQueue queue, Subscription subscription); - void subscriptionUnregistered(AMQQueue queue, Subscription subscription); - } - - void addSubscriptionRegistrationListener(SubscriptionRegistrationListener listener); - void removeSubscriptionRegistrationListener(SubscriptionRegistrationListener listener); - int getConsumerCount(); int getActiveConsumerCount(); - boolean hasExclusiveSubscriber(); + boolean hasExclusiveConsumer(); boolean isUnused(); @@ -126,14 +102,14 @@ public interface AMQQueue extends Comparable<AMQQueue>, ExchangeReferrer, Transa void requeue(QueueEntry entry); - void dequeue(QueueEntry entry, Subscription sub); + void dequeue(QueueEntry entry, Consumer sub); void decrementUnackedMsgCount(QueueEntry queueEntry); - boolean resend(final QueueEntry entry, final Subscription subscription) throws AMQException; + boolean resend(final QueueEntry entry, final Consumer consumer) throws AMQException; - void addQueueDeleteTask(final Task task); - void removeQueueDeleteTask(final Task task); + void addQueueDeleteTask(Action<AMQQueue> task); + void removeQueueDeleteTask(Action<AMQQueue> task); @@ -209,16 +185,10 @@ public interface AMQQueue extends Comparable<AMQQueue>, ExchangeReferrer, Transa Set<NotificationCheck> getNotificationChecks(); - void flushSubscription(final Subscription sub) throws AMQException; - - void deliverAsync(final Subscription sub); - void deliverAsync(); void stop(); - boolean isExclusive(); - Exchange getAlternateExchange(); void setAlternateExchange(Exchange exchange); @@ -226,56 +196,6 @@ public interface AMQQueue extends Comparable<AMQQueue>, ExchangeReferrer, Transa Collection<String> getAvailableAttributes(); Object getAttribute(String attrName); - void checkCapacity(AMQSessionModel channel); - - /** - * ExistingExclusiveSubscription signals a failure to create a subscription, because an exclusive subscription - * already exists. - * - * <p/><table id="crc"><caption>CRC Card</caption> - * <tr><th> Responsibilities <th> Collaborations - * <tr><td> Represent failure to create a subscription, because an exclusive subscription already exists. - * </table> - * - * @todo Not an AMQP exception as no status code. - * - * @todo Move to top level, used outside this class. - */ - static final class ExistingExclusiveSubscription extends AMQException - { - - public ExistingExclusiveSubscription() - { - super(""); - } - } - - /** - * ExistingSubscriptionPreventsExclusive signals a failure to create an exclusive subscription, as a subscription - * already exists. - * - * <p/><table id="crc"><caption>CRC Card</caption> - * <tr><th> Responsibilities <th> Collaborations - * <tr><td> Represent failure to create an exclusive subscription, as a subscription already exists. - * </table> - * - * @todo Not an AMQP exception as no status code. - * - * @todo Move to top level, used outside this class. - */ - static final class ExistingSubscriptionPreventsExclusive extends AMQException - { - public ExistingSubscriptionPreventsExclusive() - { - super(""); - } - } - - static interface Task - { - public void doTask(AMQQueue queue) throws AMQException; - } - void configure(QueueConfiguration config); void setExclusive(boolean exclusive); diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/subscription/AssignedSubscriptionMessageGroupManager.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/AssignedConsumerMessageGroupManager.java index ae7e11afa4..a9b36c1b24 100644 --- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/subscription/AssignedSubscriptionMessageGroupManager.java +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/AssignedConsumerMessageGroupManager.java @@ -18,28 +18,26 @@ * under the License. * */ -package org.apache.qpid.server.subscription; +package org.apache.qpid.server.queue; -import org.apache.qpid.server.queue.QueueEntryVisitor; +import org.apache.qpid.server.consumer.Consumer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import org.apache.qpid.server.queue.QueueEntry; - import java.util.Iterator; import java.util.concurrent.ConcurrentHashMap; -public class AssignedSubscriptionMessageGroupManager implements MessageGroupManager +public class AssignedConsumerMessageGroupManager implements MessageGroupManager { - private static final Logger _logger = LoggerFactory.getLogger(AssignedSubscriptionMessageGroupManager.class); + private static final Logger _logger = LoggerFactory.getLogger(AssignedConsumerMessageGroupManager.class); private final String _groupId; - private final ConcurrentHashMap<Integer, Subscription> _groupMap = new ConcurrentHashMap<Integer, Subscription>(); + private final ConcurrentHashMap<Integer, QueueConsumer> _groupMap = new ConcurrentHashMap<Integer, QueueConsumer>(); private final int _groupMask; - public AssignedSubscriptionMessageGroupManager(final String groupId, final int maxGroups) + public AssignedConsumerMessageGroupManager(final String groupId, final int maxGroups) { _groupId = groupId; _groupMask = pow2(maxGroups)-1; @@ -55,13 +53,13 @@ public class AssignedSubscriptionMessageGroupManager implements MessageGroupMana return val; } - public Subscription getAssignedSubscription(final QueueEntry entry) + public QueueConsumer getAssignedConsumer(final QueueEntry entry) { Object groupVal = entry.getMessage().getMessageHeader().getHeader(_groupId); return groupVal == null ? null : _groupMap.get(groupVal.hashCode() & _groupMask); } - public boolean acceptMessage(Subscription sub, QueueEntry entry) + public boolean acceptMessage(QueueConsumer sub, QueueEntry entry) { if(assignMessage(sub, entry)) { @@ -73,7 +71,7 @@ public class AssignedSubscriptionMessageGroupManager implements MessageGroupMana } } - private boolean assignMessage(Subscription sub, QueueEntry entry) + private boolean assignMessage(QueueConsumer sub, QueueEntry entry) { Object groupVal = entry.getMessage().getMessageHeader().getHeader(_groupId); if(groupVal == null) @@ -83,7 +81,7 @@ public class AssignedSubscriptionMessageGroupManager implements MessageGroupMana else { Integer group = groupVal.hashCode() & _groupMask; - Subscription assignedSub = _groupMap.get(group); + QueueConsumer assignedSub = _groupMap.get(group); if(assignedSub == sub) { return true; @@ -107,7 +105,7 @@ public class AssignedSubscriptionMessageGroupManager implements MessageGroupMana } } - public QueueEntry findEarliestAssignedAvailableEntry(Subscription sub) + public QueueEntry findEarliestAssignedAvailableEntry(QueueConsumer sub) { EntryFinder visitor = new EntryFinder(sub); sub.getQueue().visit(visitor); @@ -117,9 +115,9 @@ public class AssignedSubscriptionMessageGroupManager implements MessageGroupMana private class EntryFinder implements QueueEntryVisitor { private QueueEntry _entry; - private Subscription _sub; + private QueueConsumer _sub; - public EntryFinder(final Subscription sub) + public EntryFinder(final QueueConsumer sub) { _sub = sub; } @@ -138,7 +136,7 @@ public class AssignedSubscriptionMessageGroupManager implements MessageGroupMana } Integer group = groupId.hashCode() & _groupMask; - Subscription assignedSub = _groupMap.get(group); + Consumer assignedSub = _groupMap.get(group); if(assignedSub == _sub) { _entry = entry; @@ -156,9 +154,9 @@ public class AssignedSubscriptionMessageGroupManager implements MessageGroupMana } } - public void clearAssignments(Subscription sub) + public void clearAssignments(QueueConsumer sub) { - Iterator<Subscription> subIter = _groupMap.values().iterator(); + Iterator<QueueConsumer> subIter = _groupMap.values().iterator(); while(subIter.hasNext()) { if(subIter.next() == sub) diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/BaseQueue.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/BaseQueue.java index 7aba1a2342..c1c3bd37e6 100644 --- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/BaseQueue.java +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/BaseQueue.java @@ -22,19 +22,15 @@ package org.apache.qpid.server.queue; import org.apache.qpid.AMQException; +import org.apache.qpid.server.consumer.Consumer; +import org.apache.qpid.server.message.MessageInstance; import org.apache.qpid.server.message.ServerMessage; import org.apache.qpid.server.store.TransactionLogResource; +import org.apache.qpid.server.util.Action; public interface BaseQueue extends TransactionLogResource { - public static interface PostEnqueueAction - { - public void onEnqueue(QueueEntry entry); - } - - void enqueue(ServerMessage message) throws AMQException; - void enqueue(ServerMessage message, PostEnqueueAction action) throws AMQException; - void enqueue(ServerMessage message, boolean transactional, PostEnqueueAction action) throws AMQException; + void enqueue(ServerMessage message, Action<MessageInstance<? extends Consumer>> action) throws AMQException; boolean isDurable(); boolean isDeleted(); diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/ConflationQueueList.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/ConflationQueueList.java index e4725e0e2a..7469e95394 100644 --- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/ConflationQueueList.java +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/ConflationQueueList.java @@ -43,7 +43,7 @@ public class ConflationQueueList extends SimpleQueueEntryList private final QueueEntry _deleteInProgress = new SimpleQueueEntryImpl(this); private final QueueEntry _newerEntryAlreadyBeenAndGone = new SimpleQueueEntryImpl(this); - public ConflationQueueList(AMQQueue queue, String conflationKey) + public ConflationQueueList(AMQQueue<QueueConsumer> queue, String conflationKey) { super(queue); _conflationKey = conflationKey; diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/subscription/DefinedGroupMessageGroupManager.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/DefinedGroupMessageGroupManager.java index 55110c46de..4c74e5ba0b 100644 --- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/subscription/DefinedGroupMessageGroupManager.java +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/DefinedGroupMessageGroupManager.java @@ -18,15 +18,16 @@ * under the License. * */ -package org.apache.qpid.server.subscription; +package org.apache.qpid.server.queue; -import org.apache.qpid.server.queue.QueueEntryVisitor; +import org.apache.qpid.server.consumer.Consumer; +import org.apache.qpid.server.message.MessageInstance; +import org.apache.qpid.server.util.StateChangeListener; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.apache.qpid.server.message.AMQMessageHeader; import org.apache.qpid.server.message.ServerMessage; -import org.apache.qpid.server.queue.QueueEntry; import java.util.HashMap; import java.util.Map; @@ -38,23 +39,23 @@ public class DefinedGroupMessageGroupManager implements MessageGroupManager private final String _groupId; private final String _defaultGroup; private final Map<Object, Group> _groupMap = new HashMap<Object, Group>(); - private final SubscriptionResetHelper _resetHelper; + private final ConsumerResetHelper _resetHelper; private final class Group { private final Object _group; - private Subscription _subscription; + private QueueConsumer _consumer; private int _activeCount; - private Group(final Object key, final Subscription subscription) + private Group(final Object key, final QueueConsumer consumer) { _group = key; - _subscription = subscription; + _consumer = consumer; } public boolean add() { - if(_subscription != null) + if(_consumer != null) { _activeCount++; return true; @@ -69,8 +70,8 @@ public class DefinedGroupMessageGroupManager implements MessageGroupManager { if(--_activeCount == 0) { - _resetHelper.resetSubPointersForGroups(_subscription, false); - _subscription = null; + _resetHelper.resetSubPointersForGroups(_consumer, false); + _consumer = null; _groupMap.remove(_group); } } @@ -100,12 +101,12 @@ public class DefinedGroupMessageGroupManager implements MessageGroupManager public boolean isValid() { - return !(_subscription == null || (_activeCount == 0 && _subscription.isClosed())); + return !(_consumer == null || (_activeCount == 0 && _consumer.isClosed())); } - public Subscription getSubscription() + public QueueConsumer getConsumer() { - return _subscription; + return _consumer; } @Override @@ -113,28 +114,28 @@ public class DefinedGroupMessageGroupManager implements MessageGroupManager { return "Group{" + "_group=" + _group + - ", _subscription=" + _subscription + + ", _consumer=" + _consumer + ", _activeCount=" + _activeCount + '}'; } } - public DefinedGroupMessageGroupManager(final String groupId, String defaultGroup, SubscriptionResetHelper resetHelper) + public DefinedGroupMessageGroupManager(final String groupId, String defaultGroup, ConsumerResetHelper resetHelper) { _groupId = groupId; _defaultGroup = defaultGroup; _resetHelper = resetHelper; } - public synchronized Subscription getAssignedSubscription(final QueueEntry entry) + public synchronized QueueConsumer getAssignedConsumer(final QueueEntry entry) { Object groupId = getKey(entry); Group group = _groupMap.get(groupId); - return group == null || !group.isValid() ? null : group.getSubscription(); + return group == null || !group.isValid() ? null : group.getConsumer(); } - public synchronized boolean acceptMessage(final Subscription sub, final QueueEntry entry) + public synchronized boolean acceptMessage(final QueueConsumer sub, final QueueEntry entry) { if(assignMessage(sub, entry)) { @@ -146,7 +147,7 @@ public class DefinedGroupMessageGroupManager implements MessageGroupManager } } - private boolean assignMessage(final Subscription sub, final QueueEntry entry) + private boolean assignMessage(final QueueConsumer sub, final QueueEntry entry) { Object groupId = getKey(entry); Group group = _groupMap.get(groupId); @@ -158,15 +159,15 @@ public class DefinedGroupMessageGroupManager implements MessageGroupManager _groupMap.put(groupId, group); // there's a small change that the group became empty between the point at which getNextAvailable() was - // called on the subscription, and when accept message is called... in that case we want to avoid delivering + // called on the consumer, and when accept message is called... in that case we want to avoid delivering // out of order - if(_resetHelper.isEntryAheadOfSubscription(entry, sub)) + if(_resetHelper.isEntryAheadOfConsumer(entry, sub)) { return false; } } - Subscription assignedSub = group.getSubscription(); + Consumer assignedSub = group.getConsumer(); if(assignedSub == sub) { @@ -179,7 +180,7 @@ public class DefinedGroupMessageGroupManager implements MessageGroupManager } } - public synchronized QueueEntry findEarliestAssignedAvailableEntry(final Subscription sub) + public synchronized QueueEntry findEarliestAssignedAvailableEntry(final QueueConsumer sub) { EntryFinder visitor = new EntryFinder(sub); sub.getQueue().visit(visitor); @@ -189,9 +190,9 @@ public class DefinedGroupMessageGroupManager implements MessageGroupManager private class EntryFinder implements QueueEntryVisitor { private QueueEntry _entry; - private Subscription _sub; + private QueueConsumer _sub; - public EntryFinder(final Subscription sub) + public EntryFinder(final QueueConsumer sub) { _sub = sub; } @@ -206,7 +207,7 @@ public class DefinedGroupMessageGroupManager implements MessageGroupManager Object groupId = getKey(entry); Group group = _groupMap.get(groupId); - if(group != null && group.getSubscription() == _sub) + if(group != null && group.getConsumer() == _sub) { _entry = entry; return true; @@ -224,7 +225,7 @@ public class DefinedGroupMessageGroupManager implements MessageGroupManager } - public void clearAssignments(final Subscription sub) + public void clearAssignments(final QueueConsumer sub) { } @@ -240,19 +241,19 @@ public class DefinedGroupMessageGroupManager implements MessageGroupManager return groupVal; } - private class GroupStateChangeListener implements QueueEntry.StateChangeListener + private class GroupStateChangeListener implements StateChangeListener<MessageInstance<QueueConsumer>, QueueEntry.State> { private final Group _group; public GroupStateChangeListener(final Group group, - final QueueEntry entry) + final MessageInstance<QueueConsumer> entry) { _group = group; } - public void stateChanged(final QueueEntry entry, - final QueueEntry.State oldState, - final QueueEntry.State newState) + public void stateChanged(final MessageInstance<QueueConsumer> entry, + final MessageInstance.State oldState, + final MessageInstance.State newState) { synchronized (DefinedGroupMessageGroupManager.this) { diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/subscription/MessageGroupManager.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/MessageGroupManager.java index 8ce4ce3344..740a96bf2d 100644 --- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/subscription/MessageGroupManager.java +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/MessageGroupManager.java @@ -18,24 +18,22 @@ * under the License. * */ -package org.apache.qpid.server.subscription; - -import org.apache.qpid.server.queue.QueueEntry; +package org.apache.qpid.server.queue; public interface MessageGroupManager { - public interface SubscriptionResetHelper + public interface ConsumerResetHelper { - public void resetSubPointersForGroups(Subscription subscription, boolean clearAssignments); + public void resetSubPointersForGroups(QueueConsumer consumer, boolean clearAssignments); - boolean isEntryAheadOfSubscription(QueueEntry entry, Subscription sub); + boolean isEntryAheadOfConsumer(QueueEntry entry, QueueConsumer sub); } - Subscription getAssignedSubscription(QueueEntry entry); + QueueConsumer getAssignedConsumer(QueueEntry entry); - boolean acceptMessage(Subscription sub, QueueEntry entry); + boolean acceptMessage(QueueConsumer sub, QueueEntry entry); - QueueEntry findEarliestAssignedAvailableEntry(Subscription sub); + QueueEntry findEarliestAssignedAvailableEntry(QueueConsumer sub); - void clearAssignments(Subscription sub); + void clearAssignments(QueueConsumer sub); } diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/OutOfOrderQueue.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/OutOfOrderQueue.java index daa5db393a..6918ae683c 100644 --- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/OutOfOrderQueue.java +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/OutOfOrderQueue.java @@ -20,8 +20,6 @@ */ package org.apache.qpid.server.queue; -import org.apache.qpid.server.subscription.Subscription; -import org.apache.qpid.server.subscription.SubscriptionList; import org.apache.qpid.server.virtualhost.VirtualHost; import java.util.Map; @@ -38,16 +36,16 @@ public abstract class OutOfOrderQueue extends SimpleAMQQueue } @Override - protected void checkSubscriptionsNotAheadOfDelivery(final QueueEntry entry) + protected void checkConsumersNotAheadOfDelivery(final QueueEntry entry) { - // check that all subscriptions are not in advance of the entry - SubscriptionList.SubscriptionNodeIterator subIter = getSubscriptionList().iterator(); + // check that all consumers are not in advance of the entry + QueueConsumerList.ConsumerNodeIterator subIter = getConsumerList().iterator(); while(subIter.advance() && !entry.isAcquired()) { - final Subscription subscription = subIter.getNode().getSubscription(); - if(!subscription.isClosed()) + final QueueConsumer consumer = subIter.getNode().getConsumer(); + if(!consumer.isClosed()) { - QueueContext context = (QueueContext) subscription.getQueueContext(); + QueueContext context = consumer.getQueueContext(); if(context != null) { QueueEntry released = context.getReleasedEntry(); diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/PriorityQueueList.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/PriorityQueueList.java index 66315af9fb..05d84327d4 100644 --- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/PriorityQueueList.java +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/PriorityQueueList.java @@ -24,7 +24,7 @@ import org.apache.qpid.server.message.ServerMessage; public class PriorityQueueList implements QueueEntryList<SimpleQueueEntryImpl> { - private final AMQQueue _queue; + private final AMQQueue<QueueConsumer> _queue; private final PriorityQueueEntrySubList[] _priorityLists; private final int _priorities; private final int _priorityOffset; @@ -46,7 +46,7 @@ public class PriorityQueueList implements QueueEntryList<SimpleQueueEntryImpl> return _priorities; } - public AMQQueue getQueue() + public AMQQueue<QueueConsumer> getQueue() { return _queue; } @@ -166,7 +166,7 @@ public class PriorityQueueList implements QueueEntryList<SimpleQueueEntryImpl> { private int _listPriority; - public PriorityQueueEntrySubList(AMQQueue queue, int listPriority) + public PriorityQueueEntrySubList(AMQQueue<QueueConsumer> queue, int listPriority) { super(queue); _listPriority = listPriority; diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueConsumer.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueConsumer.java new file mode 100644 index 0000000000..ff7840255a --- /dev/null +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueConsumer.java @@ -0,0 +1,480 @@ +/* + * + * 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.log4j.Logger; +import org.apache.qpid.AMQException; +import org.apache.qpid.server.filter.FilterManager; +import org.apache.qpid.server.logging.LogActor; +import org.apache.qpid.server.logging.LogSubject; +import org.apache.qpid.server.logging.actors.CurrentActor; +import org.apache.qpid.server.logging.actors.GenericActor; +import org.apache.qpid.server.logging.messages.SubscriptionMessages; +import org.apache.qpid.server.logging.subjects.QueueLogSubject; +import org.apache.qpid.server.message.MessageInstance; +import org.apache.qpid.server.message.ServerMessage; +import org.apache.qpid.server.protocol.AMQSessionModel; +import org.apache.qpid.server.protocol.MessageConverterRegistry; +import org.apache.qpid.server.consumer.Consumer; +import org.apache.qpid.server.consumer.ConsumerTarget; +import org.apache.qpid.server.util.StateChangeListener; + +import java.text.MessageFormat; +import java.util.EnumMap; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicReference; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; + +import static org.apache.qpid.server.logging.subjects.LogSubjectFormat.SUBSCRIPTION_FORMAT; + +class QueueConsumer<T extends ConsumerTarget> implements Consumer +{ + + public static enum State + { + ACTIVE, + SUSPENDED, + CLOSED + } + + private static final Logger _logger = Logger.getLogger(QueueConsumer.class); + private final AtomicBoolean _targetClosed = new AtomicBoolean(false); + private final AtomicBoolean _closed = new AtomicBoolean(false); + private final long _id; + private final AtomicReference<State> _state = new AtomicReference<State>(State.ACTIVE); + private final Lock _stateChangeLock = new ReentrantLock(); + private final long _createTime = System.currentTimeMillis(); + private final MessageInstance.ConsumerAcquiredState _owningState = new MessageInstance.ConsumerAcquiredState(this); + private final boolean _acquires; + private final boolean _seesRequeues; + private final String _consumerName; + private final boolean _isTransient; + private final AtomicLong _deliveredCount = new AtomicLong(0); + private final AtomicLong _deliveredBytes = new AtomicLong(0); + private final FilterManager _filters; + private final Class<? extends ServerMessage> _messageClass; + private final Object _sessionReference; + private SimpleAMQQueue _queue; + private GenericActor _logActor; + + static final EnumMap<ConsumerTarget.State, State> STATE_MAP = + new EnumMap<ConsumerTarget.State, State>(ConsumerTarget.State.class); + + static + { + STATE_MAP.put(ConsumerTarget.State.ACTIVE, State.ACTIVE); + STATE_MAP.put(ConsumerTarget.State.SUSPENDED, State.SUSPENDED); + STATE_MAP.put(ConsumerTarget.State.CLOSED, State.CLOSED); + } + + private final T _target; + private final SubFlushRunner _runner = new SubFlushRunner(this); + private volatile QueueContext _queueContext; + private StateChangeListener<? extends Consumer, State> _stateListener = new StateChangeListener<Consumer, State>() + { + public void stateChanged(Consumer sub, State oldState, State newState) + { + CurrentActor.get().message(SubscriptionMessages.STATE(newState.toString())); + } + }; + private boolean _noLocal; + + QueueConsumer(final FilterManager filters, + final Class<? extends ServerMessage> messageClass, + final boolean acquires, + final boolean seesRequeues, + final String consumerName, + final boolean isTransient, + T target) + { + _messageClass = messageClass; + _sessionReference = target.getSessionModel().getConnectionReference(); + _id = SUB_ID_GENERATOR.getAndIncrement(); + _filters = filters; + _acquires = acquires; + _seesRequeues = seesRequeues; + _consumerName = consumerName; + _isTransient = isTransient; + _target = target; + _target.setStateListener( + new StateChangeListener<ConsumerTarget, ConsumerTarget.State>() + { + @Override + public void stateChanged(final ConsumerTarget object, + final ConsumerTarget.State oldState, + final ConsumerTarget.State newState) + { + targetStateChanged(oldState, newState); + } + }); + } + + private void targetStateChanged(final ConsumerTarget.State oldState, final ConsumerTarget.State newState) + { + if(oldState != newState) + { + if(newState == ConsumerTarget.State.CLOSED) + { + if(_targetClosed.compareAndSet(false,true)) + { + CurrentActor.get().message(getLogSubject(), SubscriptionMessages.CLOSE()); + } + } + else + { + CurrentActor.get().message(getLogSubject(),SubscriptionMessages.STATE(newState.toString())); + } + } + + if(newState == ConsumerTarget.State.CLOSED && oldState != newState && !_closed.get()) + { + try + { + close(); + } + catch (AMQException e) + { + _logger.error("Unable to remove to remove consumer", e); + throw new RuntimeException(e); + } + } + final StateChangeListener<Consumer, State> stateListener = + (StateChangeListener<Consumer, State>) getStateListener(); + if(stateListener != null) + { + stateListener.stateChanged(this, STATE_MAP.get(oldState), STATE_MAP.get(newState)); + } + } + + public T getTarget() + { + return _target; + } + + @Override + public void externalStateChange() + { + getQueue().deliverAsync(this); + } + + @Override + public long getUnacknowledgedBytes() + { + return _target.getUnacknowledgedBytes(); + } + + @Override + public long getUnacknowledgedMessages() + { + return _target.getUnacknowledgedMessages(); + } + + @Override + public AMQSessionModel getSessionModel() + { + return _target.getSessionModel(); + } + + @Override + public boolean isSuspended() + { + return _target.isSuspended(); + } + + @Override + public void close() throws AMQException + { + if(_closed.compareAndSet(false,true)) + { + getSendLock(); + try + { + _target.close(); + _target.consumerRemoved(this); + _queue.unregisterConsumer(this); + } + finally + { + releaseSendLock(); + } + + } + } + + void flushBatched() + { + _target.flushBatched(); + } + + void queueDeleted() + { + _target.queueDeleted(); + } + + boolean wouldSuspend(final MessageInstance msg) + { + return !_target.allocateCredit(msg.getMessage()); + } + + void restoreCredit(final MessageInstance queueEntry) + { + _target.restoreCredit(queueEntry.getMessage()); + } + + void queueEmpty() throws AMQException + { + _target.queueEmpty(); + } + + State getState() + { + return STATE_MAP.get(_target.getState()); + } + + public final SimpleAMQQueue getQueue() + { + return _queue; + } + + final void setQueue(SimpleAMQQueue queue, boolean exclusive) + { + if(getQueue() != null) + { + throw new IllegalStateException("Attempt to set queue for consumer " + this + " to " + queue + "when already set to " + getQueue()); + } + _queue = queue; + + String queueString = new QueueLogSubject(_queue).toLogString(); + + _logActor = new GenericActor("[" + MessageFormat.format(SUBSCRIPTION_FORMAT, getId()) + + "(" + // queueString is [vh(/{0})/qu({1}) ] so need to trim + // ^ ^^ + + queueString.substring(1,queueString.length() - 3) + + ")" + + "] "); + + + if (CurrentActor.get().getRootMessageLogger().isMessageEnabled(_logActor, _logActor.getLogSubject(), SubscriptionMessages.CREATE_LOG_HIERARCHY)) + { + final String filterLogString = getFilterLogString(); + CurrentActor.get().message(_logActor.getLogSubject(), SubscriptionMessages.CREATE(filterLogString, queue.isDurable() && exclusive, + filterLogString.length() > 0)); + } + } + + protected final LogSubject getLogSubject() + { + return _logActor.getLogSubject(); + } + + final LogActor getLogActor() + { + return _logActor; + } + + + @Override + public final void flush() throws AMQException + { + getQueue().flushConsumer(this); + } + + boolean resend(final MessageInstance entry) throws AMQException + { + return getQueue().resend((QueueEntry)entry, this); + } + + final SubFlushRunner getRunner() + { + return _runner; + } + + public final long getId() + { + return _id; + } + + public final StateChangeListener<? extends Consumer, State> getStateListener() + { + return _stateListener; + } + + public final void setStateListener(StateChangeListener<? extends Consumer, State> listener) + { + _stateListener = listener; + } + + final QueueContext getQueueContext() + { + return _queueContext; + } + + final void setQueueContext(QueueContext queueContext) + { + _queueContext = queueContext; + } + + protected boolean updateState(State from, State to) + { + return _state.compareAndSet(from, to); + } + + public final boolean isActive() + { + return getState() == State.ACTIVE; + } + + public final boolean isClosed() + { + return getState() == State.CLOSED; + } + + public final void setNoLocal(boolean noLocal) + { + _noLocal = noLocal; + } + + public final boolean hasInterest(MessageInstance entry) + { + //check that the message hasn't been rejected + if (entry.isRejectedBy(this)) + { + + return false; + } + + if (entry.getMessage().getClass() == _messageClass) + { + if(_noLocal) + { + Object connectionRef = entry.getMessage().getConnectionReference(); + if (connectionRef != null && connectionRef == _sessionReference) + { + return false; + } + } + } + else + { + // no interest in messages we can't convert + if(_messageClass != null && MessageConverterRegistry.getConverter(entry.getMessage().getClass(), + _messageClass)==null) + { + return false; + } + } + return (_filters == null) || _filters.allAllow(entry.asFilterable()); + } + + protected String getFilterLogString() + { + StringBuilder filterLogString = new StringBuilder(); + String delimiter = ", "; + boolean hasEntries = false; + if (_filters != null && _filters.hasFilters()) + { + filterLogString.append(_filters.toString()); + hasEntries = true; + } + + if (!acquires()) + { + if (hasEntries) + { + filterLogString.append(delimiter); + } + filterLogString.append("Browser"); + hasEntries = true; + } + + return filterLogString.toString(); + } + + public final boolean trySendLock() + { + return _stateChangeLock.tryLock(); + } + + public final void getSendLock() + { + _stateChangeLock.lock(); + } + + public final void releaseSendLock() + { + _stateChangeLock.unlock(); + } + + public final long getCreateTime() + { + return _createTime; + } + + final MessageInstance.ConsumerAcquiredState getOwningState() + { + return _owningState; + } + + public final boolean acquires() + { + return _acquires; + } + + public final boolean seesRequeues() + { + return _seesRequeues; + } + + public final String getName() + { + return _consumerName; + } + + public final boolean isTransient() + { + return _isTransient; + } + + public final long getBytesOut() + { + return _deliveredBytes.longValue(); + } + + public final long getMessagesOut() + { + return _deliveredCount.longValue(); + } + + final void send(final QueueEntry entry, final boolean batch) throws AMQException + { + _deliveredCount.incrementAndGet(); + ServerMessage message = entry.getMessage(); + if(message == null) + { + throw new AMQException("message was null!"); + } + _deliveredBytes.addAndGet(message.getSize()); + _target.send(entry, batch); + } +} diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/subscription/SubscriptionList.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueConsumerList.java index 23d4fb241f..82e9d58cf3 100644 --- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/subscription/SubscriptionList.java +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueConsumerList.java @@ -18,34 +18,34 @@ * under the License. * */ -package org.apache.qpid.server.subscription; +package org.apache.qpid.server.queue; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; -public class SubscriptionList +class QueueConsumerList { - private final SubscriptionNode _head = new SubscriptionNode(); + private final ConsumerNode _head = new ConsumerNode(); - private final AtomicReference<SubscriptionNode> _tail = new AtomicReference<SubscriptionNode>(_head); - private final AtomicReference<SubscriptionNode> _subNodeMarker = new AtomicReference<SubscriptionNode>(_head); + private final AtomicReference<ConsumerNode> _tail = new AtomicReference<ConsumerNode>(_head); + private final AtomicReference<ConsumerNode> _subNodeMarker = new AtomicReference<ConsumerNode>(_head); private final AtomicInteger _size = new AtomicInteger(); - public static final class SubscriptionNode + public static final class ConsumerNode { private final AtomicBoolean _deleted = new AtomicBoolean(); - private final AtomicReference<SubscriptionNode> _next = new AtomicReference<SubscriptionNode>(); - private final Subscription _sub; + private final AtomicReference<ConsumerNode> _next = new AtomicReference<ConsumerNode>(); + private final QueueConsumer _sub; - public SubscriptionNode() + public ConsumerNode() { //used for sentinel head and dummy node construction _sub = null; _deleted.set(true); } - public SubscriptionNode(final Subscription sub) + public ConsumerNode(final QueueConsumer sub) { //used for regular node construction _sub = sub; @@ -57,12 +57,12 @@ public class SubscriptionList * * @return the next non-deleted node, or null if none was found. */ - public SubscriptionNode findNext() + public ConsumerNode findNext() { - SubscriptionNode next = nextNode(); + ConsumerNode next = nextNode(); while(next != null && next.isDeleted()) { - final SubscriptionNode newNext = next.nextNode(); + final ConsumerNode newNext = next.nextNode(); if(newNext != null) { //try to move our _next reference forward to the 'newNext' @@ -86,7 +86,7 @@ public class SubscriptionList * * @return the immediately next node in the structure, or null if at the tail. */ - protected SubscriptionNode nextNode() + protected ConsumerNode nextNode() { return _next.get(); } @@ -94,10 +94,10 @@ public class SubscriptionList /** * Used to initialise the 'next' reference. Will only succeed if the reference was not previously set. * - * @param node the SubscriptionNode to set as 'next' + * @param node the ConsumerNode to set as 'next' * @return whether the operation succeeded */ - private boolean setNext(final SubscriptionNode node) + private boolean setNext(final ConsumerNode node) { return _next.compareAndSet(null, node); } @@ -112,18 +112,18 @@ public class SubscriptionList return _deleted.compareAndSet(false,true); } - public Subscription getSubscription() + public QueueConsumer getConsumer() { return _sub; } } - private void insert(final SubscriptionNode node, final boolean count) + private void insert(final ConsumerNode node, final boolean count) { for (;;) { - SubscriptionNode tail = _tail.get(); - SubscriptionNode next = tail.nextNode(); + ConsumerNode tail = _tail.get(); + ConsumerNode next = tail.nextNode(); if (tail == _tail.get()) { if (next == null) @@ -146,35 +146,35 @@ public class SubscriptionList } } - public void add(final Subscription sub) + public void add(final QueueConsumer sub) { - SubscriptionNode node = new SubscriptionNode(sub); + ConsumerNode node = new ConsumerNode(sub); insert(node, true); } - public boolean remove(final Subscription sub) + public boolean remove(final QueueConsumer sub) { - SubscriptionNode prevNode = _head; - SubscriptionNode node = _head.nextNode(); + ConsumerNode prevNode = _head; + ConsumerNode node = _head.nextNode(); while(node != null) { - if(sub.equals(node.getSubscription()) && node.delete()) + if(sub.equals(node.getConsumer()) && node.delete()) { _size.decrementAndGet(); - SubscriptionNode tail = _tail.get(); + ConsumerNode tail = _tail.get(); if(node == tail) { //we cant remove the last node from the structure for //correctness reasons, however we have just 'deleted' //the tail. Inserting an empty dummy node after it will - //let us scavenge the node containing the Subscription. - insert(new SubscriptionNode(), false); + //let us scavenge the node containing the Consumer. + insert(new ConsumerNode(), false); } //advance the next node reference in the 'prevNode' to scavenge - //the newly 'deleted' node for the Subscription. + //the newly 'deleted' node for the Consumer. prevNode.findNext(); nodeMarkerCleanup(node); @@ -189,9 +189,9 @@ public class SubscriptionList return false; } - private void nodeMarkerCleanup(final SubscriptionNode node) + private void nodeMarkerCleanup(final ConsumerNode node) { - SubscriptionNode markedNode = _subNodeMarker.get(); + ConsumerNode markedNode = _subNodeMarker.get(); if(node == markedNode) { //if the marked node is the one we are removing, then @@ -200,7 +200,7 @@ public class SubscriptionList //into the list and find the next node to use. //Because we inserted a dummy if node was the //tail, markedNode.nextNode() can never be null. - SubscriptionNode dummy = new SubscriptionNode(); + ConsumerNode dummy = new ConsumerNode(); dummy.setNext(markedNode.nextNode()); //if the CAS fails the marked node has changed, thus @@ -219,53 +219,53 @@ public class SubscriptionList } } - public boolean updateMarkedNode(final SubscriptionNode expected, final SubscriptionNode nextNode) + public boolean updateMarkedNode(final ConsumerNode expected, final ConsumerNode nextNode) { return _subNodeMarker.compareAndSet(expected, nextNode); } /** - * Get the current marked SubscriptionNode. This should only be used only to index into the list and find the next node + * Get the current marked ConsumerNode. This should only be used only to index into the list and find the next node * after the mark, since if the previously marked node was subsequently deleted the item returned may be a dummy node * with reference to the next node. * * @return the previously marked node (or a dummy if it was subsequently deleted) */ - public SubscriptionNode getMarkedNode() + public ConsumerNode getMarkedNode() { return _subNodeMarker.get(); } - public static class SubscriptionNodeIterator + public static class ConsumerNodeIterator { - private SubscriptionNode _lastNode; + private ConsumerNode _lastNode; - SubscriptionNodeIterator(SubscriptionNode startNode) + ConsumerNodeIterator(ConsumerNode startNode) { _lastNode = startNode; } - public SubscriptionNode getNode() + public ConsumerNode getNode() { return _lastNode; } public boolean advance() { - SubscriptionNode nextNode = _lastNode.findNext(); + ConsumerNode nextNode = _lastNode.findNext(); _lastNode = nextNode; return _lastNode != null; } } - public SubscriptionNodeIterator iterator() + public ConsumerNodeIterator iterator() { - return new SubscriptionNodeIterator(_head); + return new ConsumerNodeIterator(_head); } - public SubscriptionNode getHead() + public ConsumerNode getHead() { return _head; } diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueContext.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueContext.java index 79279b44c7..861bd3dea1 100755 --- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueContext.java +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueContext.java @@ -23,7 +23,7 @@ package org.apache.qpid.server.queue; import java.util.concurrent.atomic.AtomicReferenceFieldUpdater; -final class QueueContext implements AMQQueue.Context +final class QueueContext { private volatile QueueEntry _lastSeenEntry; private volatile QueueEntry _releasedEntry; diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueEntry.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueEntry.java index 2aa1d1f473..6a42088c47 100644 --- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueEntry.java +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueEntry.java @@ -20,207 +20,20 @@ */ package org.apache.qpid.server.queue; -import org.apache.qpid.AMQException; -import org.apache.qpid.server.filter.Filterable; import org.apache.qpid.server.message.MessageInstance; -import org.apache.qpid.server.subscription.Subscription; -import org.apache.qpid.server.txn.ServerTransaction; -public interface QueueEntry extends MessageInstance, Comparable<QueueEntry> +public interface QueueEntry extends MessageInstance<QueueConsumer>, Comparable<QueueEntry> { - - - public static enum State - { - AVAILABLE, - ACQUIRED, - EXPIRED, - DEQUEUED, - DELETED; - - - } - - public static interface StateChangeListener - { - public void stateChanged(QueueEntry entry, State oldSate, State newState); - } - - public abstract class EntryState - { - private EntryState() - { - } - - public abstract State getState(); - - /** - * Returns true if state is either DEQUEUED or DELETED. - * - * @return true if state is either DEQUEUED or DELETED. - */ - public boolean isDispensed() - { - State currentState = getState(); - return currentState == State.DEQUEUED || currentState == State.DELETED; - } - } - - - public final class AvailableState extends EntryState - { - - public State getState() - { - return State.AVAILABLE; - } - - public String toString() - { - return getState().name(); - } - } - - - public final class DequeuedState extends EntryState - { - - public State getState() - { - return State.DEQUEUED; - } - - public String toString() - { - return getState().name(); - } - } - - - public final class DeletedState extends EntryState - { - - public State getState() - { - return State.DELETED; - } - - public String toString() - { - return getState().name(); - } - } - - public final class ExpiredState extends EntryState - { - - public State getState() - { - return State.EXPIRED; - } - - public String toString() - { - return getState().name(); - } - } - - - public final class NonSubscriptionAcquiredState extends EntryState - { - public State getState() - { - return State.ACQUIRED; - } - - public String toString() - { - return getState().name(); - } - } - - public final class SubscriptionAcquiredState extends EntryState - { - private final Subscription _subscription; - - public SubscriptionAcquiredState(Subscription subscription) - { - _subscription = subscription; - } - - - public State getState() - { - return State.ACQUIRED; - } - - public Subscription getSubscription() - { - return _subscription; - } - - public String toString() - { - return "{" + getState().name() + " : " + _subscription +"}"; - } - } - - - final static EntryState AVAILABLE_STATE = new AvailableState(); - final static EntryState DELETED_STATE = new DeletedState(); - final static EntryState DEQUEUED_STATE = new DequeuedState(); - final static EntryState NON_SUBSCRIPTION_ACQUIRED_STATE = new NonSubscriptionAcquiredState(); - - - - - AMQQueue getQueue(); + AMQQueue<QueueConsumer> getQueue(); long getSize(); - boolean getDeliveredToConsumer(); - - boolean expired() throws AMQException; - - boolean acquire(Subscription sub); - - boolean acquiredBySubscription(); - boolean isAcquiredBy(Subscription subscription); - - void setRedelivered(); - - boolean isRedelivered(); - - Subscription getDeliveredSubscription(); - - void reject(); - - boolean isRejectedBy(long subscriptionId); - - int routeToAlternate(final BaseQueue.PostEnqueueAction action, ServerTransaction txn); - boolean isQueueDeleted(); QueueEntry getNextNode(); QueueEntry getNextValidEntry(); - void addStateChangeListener(StateChangeListener listener); - boolean removeStateChangeListener(StateChangeListener listener); - - - /** - * Number of times this queue entry has been delivered. - * - * @return delivery count - */ - int getDeliveryCount(); - - void incrementDeliveryCount(); - - void decrementDeliveryCount(); - - Filterable asFilterable(); } diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueEntryImpl.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueEntryImpl.java index 461d493437..8b81a87903 100644 --- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueEntryImpl.java +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueEntryImpl.java @@ -26,11 +26,15 @@ import org.apache.qpid.AMQException; import org.apache.qpid.server.exchange.Exchange; import org.apache.qpid.server.filter.Filterable; import org.apache.qpid.server.message.InstanceProperties; +import org.apache.qpid.server.message.MessageInstance; import org.apache.qpid.server.message.MessageReference; import org.apache.qpid.server.message.ServerMessage; -import org.apache.qpid.server.subscription.Subscription; +import org.apache.qpid.server.store.TransactionLogResource; +import org.apache.qpid.server.consumer.Consumer; import org.apache.qpid.server.txn.LocalTransaction; import org.apache.qpid.server.txn.ServerTransaction; +import org.apache.qpid.server.util.Action; +import org.apache.qpid.server.util.StateChangeListener; import java.util.EnumMap; import java.util.HashSet; @@ -46,7 +50,7 @@ public abstract class QueueEntryImpl implements QueueEntry private final QueueEntryList _queueEntryList; - private MessageReference _message; + private final MessageReference _message; private Set<Long> _rejectedBy = null; @@ -59,7 +63,7 @@ public abstract class QueueEntryImpl implements QueueEntry (QueueEntryImpl.class, EntryState.class, "_state"); - private volatile Set<StateChangeListener> _stateChangeListeners; + private volatile Set<StateChangeListener<MessageInstance<QueueConsumer>, State>> _stateChangeListeners; private static final AtomicReferenceFieldUpdater<QueueEntryImpl, Set> @@ -134,7 +138,7 @@ public abstract class QueueEntryImpl implements QueueEntry return _entryId; } - public AMQQueue getQueue() + public AMQQueue<QueueConsumer> getQueue() { return _queueEntryList.getQueue(); } @@ -183,7 +187,7 @@ public abstract class QueueEntryImpl implements QueueEntry public boolean acquire() { - return acquire(NON_SUBSCRIPTION_ACQUIRED_STATE); + return acquire(NON_CONSUMER_ACQUIRED_STATE); } private boolean acquire(final EntryState state) @@ -198,7 +202,7 @@ public abstract class QueueEntryImpl implements QueueEntry return acquired; } - public boolean acquire(Subscription sub) + public boolean acquire(QueueConsumer sub) { final boolean acquired = acquire(sub.getOwningState()); if(acquired) @@ -208,17 +212,17 @@ public abstract class QueueEntryImpl implements QueueEntry return acquired; } - public boolean acquiredBySubscription() + public boolean acquiredByConsumer() { - return (_state instanceof SubscriptionAcquiredState); + return (_state instanceof ConsumerAcquiredState); } - public boolean isAcquiredBy(Subscription subscription) + public boolean isAcquiredBy(QueueConsumer consumer) { EntryState state = _state; - return state instanceof SubscriptionAcquiredState - && ((SubscriptionAcquiredState)state).getSubscription() == subscription; + return state instanceof ConsumerAcquiredState + && ((ConsumerAcquiredState)state).getConsumer() == consumer; } public void release() @@ -228,14 +232,9 @@ public abstract class QueueEntryImpl implements QueueEntry if((state.getState() == State.ACQUIRED) &&_stateUpdater.compareAndSet(this, state, AVAILABLE_STATE)) { - if(state instanceof SubscriptionAcquiredState) + if(state instanceof ConsumerAcquiredState) { getQueue().decrementUnackedMsgCount(this); - Subscription subscription = ((SubscriptionAcquiredState)state).getSubscription(); - if (subscription != null) - { - subscription.releaseQueueEntry(this); - } } if(!getQueue().isDeleted()) @@ -265,12 +264,12 @@ public abstract class QueueEntryImpl implements QueueEntry return Boolean.TRUE.equals(_instanceProperties.getProperty(InstanceProperties.Property.REDELIVERED)); } - public Subscription getDeliveredSubscription() + public QueueConsumer getDeliveredConsumer() { EntryState state = _state; - if (state instanceof SubscriptionAcquiredState) + if (state instanceof ConsumerAcquiredState) { - return ((SubscriptionAcquiredState) state).getSubscription(); + return (QueueConsumer) ((ConsumerAcquiredState) state).getConsumer(); } else { @@ -280,16 +279,16 @@ public abstract class QueueEntryImpl implements QueueEntry public void reject() { - Subscription subscription = getDeliveredSubscription(); + QueueConsumer consumer = getDeliveredConsumer(); - if (subscription != null) + if (consumer != null) { if (_rejectedBy == null) { _rejectedBy = new HashSet<Long>(); } - _rejectedBy.add(subscription.getSubscriptionID()); + _rejectedBy.add(consumer.getId()); } else { @@ -297,12 +296,12 @@ public abstract class QueueEntryImpl implements QueueEntry } } - public boolean isRejectedBy(long subscriptionId) + public boolean isRejectedBy(QueueConsumer consumer) { - if (_rejectedBy != null) // We have subscriptions that rejected this message + if (_rejectedBy != null) // We have consumers that rejected this message { - return _rejectedBy.contains(subscriptionId); + return _rejectedBy.contains(consumer.getId()); } else // This message hasn't been rejected yet. { @@ -316,12 +315,10 @@ public abstract class QueueEntryImpl implements QueueEntry if((state.getState() == State.ACQUIRED) &&_stateUpdater.compareAndSet(this, state, DEQUEUED_STATE)) { - Subscription s = null; - if (state instanceof SubscriptionAcquiredState) + Consumer s = null; + if (state instanceof ConsumerAcquiredState) { getQueue().decrementUnackedMsgCount(this); - s = ((SubscriptionAcquiredState) state).getSubscription(); - s.onDequeue(this); } getQueue().dequeue(this,s); @@ -336,7 +333,7 @@ public abstract class QueueEntryImpl implements QueueEntry private void notifyStateChange(final State oldState, final State newState) { - for(StateChangeListener l : _stateChangeListeners) + for(StateChangeListener<MessageInstance<QueueConsumer>, State> l : _stateChangeListeners) { l.stateChanged(this, oldState, newState); } @@ -367,7 +364,7 @@ public abstract class QueueEntryImpl implements QueueEntry dispose(); } - public int routeToAlternate(final BaseQueue.PostEnqueueAction action, ServerTransaction txn) + public int routeToAlternate(final Action<MessageInstance<? extends Consumer>> action, ServerTransaction txn) { final AMQQueue currentQueue = getQueue(); Exchange alternateExchange = currentQueue.getAlternateExchange(); @@ -379,7 +376,10 @@ public abstract class QueueEntryImpl implements QueueEntry txn = new LocalTransaction(getQueue().getVirtualHost().getMessageStore()); } - int enqueues = alternateExchange.send(getMessage(), getInstanceProperties(), txn, action); + int enqueues = alternateExchange.send(getMessage(), + getInstanceProperties(), + txn, + action); txn.dequeue(currentQueue, getMessage(), new ServerTransaction.Action() { @@ -412,21 +412,21 @@ public abstract class QueueEntryImpl implements QueueEntry return getQueue().isDeleted(); } - public void addStateChangeListener(StateChangeListener listener) + public void addStateChangeListener(StateChangeListener<MessageInstance<QueueConsumer>, State> listener) { - Set<StateChangeListener> listeners = _stateChangeListeners; + Set<StateChangeListener<MessageInstance<QueueConsumer>, State>> listeners = _stateChangeListeners; if(listeners == null) { - _listenersUpdater.compareAndSet(this, null, new CopyOnWriteArraySet<StateChangeListener>()); + _listenersUpdater.compareAndSet(this, null, new CopyOnWriteArraySet<StateChangeListener<MessageInstance<QueueConsumer>, State>>()); listeners = _stateChangeListeners; } listeners.add(listener); } - public boolean removeStateChangeListener(StateChangeListener listener) + public boolean removeStateChangeListener(StateChangeListener<MessageInstance<QueueConsumer>, State> listener) { - Set<StateChangeListener> listeners = _stateChangeListeners; + Set<StateChangeListener<MessageInstance<QueueConsumer>, State>> listeners = _stateChangeListeners; if(listeners != null) { return listeners.remove(listener); @@ -461,6 +461,12 @@ public abstract class QueueEntryImpl implements QueueEntry return _deliveryCount; } + @Override + public int getMaximumDeliveryCount() + { + return getQueue().getMaximumDeliveryCount(); + } + public void incrementDeliveryCount() { _deliveryCountUpdater.incrementAndGet(this); @@ -485,6 +491,23 @@ public abstract class QueueEntryImpl implements QueueEntry '}'; } + @Override + public boolean resend() throws AMQException + { + QueueConsumer sub = getDeliveredConsumer(); + if(sub != null) + { + return sub.resend(this); + } + return false; + } + + @Override + public TransactionLogResource getOwningResource() + { + return getQueue(); + } + private static class EntryInstanceProperties implements InstanceProperties { private final EnumMap<Property, Object> _properties = new EnumMap<Property, Object>(Property.class); diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueEntryList.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueEntryList.java index 641aaa0a08..ad1f703f51 100644 --- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueEntryList.java +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueEntryList.java @@ -24,7 +24,7 @@ import org.apache.qpid.server.message.ServerMessage; public interface QueueEntryList<Q extends QueueEntry> { - AMQQueue getQueue(); + AMQQueue<QueueConsumer> getQueue(); Q add(ServerMessage message); diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueRunner.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueRunner.java index 22a2029494..005d9b66b3 100644 --- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueRunner.java +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueRunner.java @@ -32,8 +32,8 @@ import org.apache.qpid.transport.TransportException; /** * QueueRunners are Runnables used to process a queue when requiring - * asynchronous message delivery to subscriptions, which is necessary - * when straight-through delivery of a message to a subscription isn't + * asynchronous message delivery to consumers, which is necessary + * when straight-through delivery of a message to a consumer isn't * possible during the enqueue operation. */ public class QueueRunner implements Runnable diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java index 87d11a892e..f4a9794fcd 100644 --- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java @@ -18,15 +18,7 @@ */ package org.apache.qpid.server.queue; -import java.util.ArrayList; -import java.util.Collection; -import java.util.Collections; -import java.util.EnumSet; -import java.util.LinkedHashMap; -import java.util.List; -import java.util.Map; -import java.util.Set; -import java.util.UUID; +import java.util.*; import java.util.concurrent.ConcurrentSkipListSet; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.Executor; @@ -43,27 +35,33 @@ import org.apache.qpid.server.binding.Binding; import org.apache.qpid.server.configuration.BrokerProperties; import org.apache.qpid.server.configuration.QueueConfiguration; import org.apache.qpid.server.exchange.Exchange; +import org.apache.qpid.server.filter.FilterManager; import org.apache.qpid.server.logging.LogActor; import org.apache.qpid.server.logging.LogSubject; import org.apache.qpid.server.logging.actors.CurrentActor; import org.apache.qpid.server.logging.actors.QueueActor; import org.apache.qpid.server.logging.messages.QueueMessages; import org.apache.qpid.server.logging.subjects.QueueLogSubject; +import org.apache.qpid.server.message.InstanceProperties; +import org.apache.qpid.server.message.MessageDestination; +import org.apache.qpid.server.message.MessageInstance; +import org.apache.qpid.server.message.MessageReference; import org.apache.qpid.server.message.ServerMessage; import org.apache.qpid.server.model.Queue; import org.apache.qpid.server.protocol.AMQSessionModel; import org.apache.qpid.server.security.AuthorizationHolder; -import org.apache.qpid.server.subscription.AssignedSubscriptionMessageGroupManager; -import org.apache.qpid.server.subscription.DefinedGroupMessageGroupManager; -import org.apache.qpid.server.subscription.MessageGroupManager; -import org.apache.qpid.server.subscription.Subscription; -import org.apache.qpid.server.subscription.SubscriptionList; +import org.apache.qpid.server.consumer.Consumer; +import org.apache.qpid.server.consumer.ConsumerTarget; import org.apache.qpid.server.txn.AutoCommitTransaction; import org.apache.qpid.server.txn.LocalTransaction; import org.apache.qpid.server.txn.ServerTransaction; +import org.apache.qpid.server.util.Action; +import org.apache.qpid.server.util.StateChangeListener; import org.apache.qpid.server.virtualhost.VirtualHost; -public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener, MessageGroupManager.SubscriptionResetHelper +public class SimpleAMQQueue implements AMQQueue<QueueConsumer>, + StateChangeListener<QueueConsumer, QueueConsumer.State>, + MessageGroupManager.ConsumerResetHelper { private static final Logger _logger = Logger.getLogger(SimpleAMQQueue.class); @@ -98,9 +96,9 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener, Mes private final QueueEntryList<QueueEntry> _entries; - private final SubscriptionList _subscriptionList = new SubscriptionList(); + private final QueueConsumerList _consumerList = new QueueConsumerList(); - private volatile Subscription _exclusiveSubscriber; + private volatile QueueConsumer _exclusiveSubscriber; @@ -120,13 +118,7 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener, Mes private final AtomicLong _persistentMessageDequeueSize = new AtomicLong(); private final AtomicLong _persistentMessageEnqueueCount = new AtomicLong(); private final AtomicLong _persistentMessageDequeueCount = new AtomicLong(); - private final AtomicInteger _consumerCountHigh = new AtomicInteger(0); - private final AtomicLong _msgTxnEnqueues = new AtomicLong(0); - private final AtomicLong _byteTxnEnqueues = new AtomicLong(0); - private final AtomicLong _msgTxnDequeues = new AtomicLong(0); - private final AtomicLong _byteTxnDequeues = new AtomicLong(0); private final AtomicLong _unackedMsgCount = new AtomicLong(0); - private final AtomicLong _unackedMsgCountHigh = new AtomicLong(0); private final AtomicLong _unackedMsgBytes = new AtomicLong(); private final AtomicInteger _bindingCountHigh = new AtomicInteger(); @@ -165,7 +157,7 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener, Mes private final Set<AMQSessionModel> _blockedChannels = new ConcurrentSkipListSet<AMQSessionModel>(); private final AtomicBoolean _deleted = new AtomicBoolean(false); - private final List<Task> _deleteTaskList = new CopyOnWriteArrayList<Task>(); + private final List<Action<AMQQueue>> _deleteTaskList = new CopyOnWriteArrayList<Action<AMQQueue>>(); private LogSubject _logSubject; @@ -187,8 +179,8 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener, Mes private int _maximumDeliveryCount; private final MessageGroupManager _messageGroupManager; - private final Collection<SubscriptionRegistrationListener> _subscriptionListeners = - new ArrayList<SubscriptionRegistrationListener>(); + private final Collection<ConsumerRegistrationListener> _consumerListeners = + new ArrayList<ConsumerRegistrationListener>(); private AMQQueue.NotificationListener _notificationListener; private final long[] _lastNotificationTimes = new long[NotificationCheck.values().length]; @@ -257,7 +249,7 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener, Mes } else { - _messageGroupManager = new AssignedSubscriptionMessageGroupManager(String.valueOf(arguments.get( + _messageGroupManager = new AssignedConsumerMessageGroupManager(String.valueOf(arguments.get( Queue.MESSAGE_GROUP_KEY)), DEFAULT_MAX_GROUPS); } } @@ -388,11 +380,17 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener, Mes return _name; } - // ------ Manage Subscriptions + // ------ Manage Consumers - public synchronized void registerSubscription(final Subscription subscription, final boolean exclusive) - throws AMQSecurityException, ExistingExclusiveSubscription, ExistingSubscriptionPreventsExclusive + + @Override + public synchronized QueueConsumer addConsumer(final ConsumerTarget target, + final FilterManager filters, + final Class<? extends ServerMessage> messageClass, + final String consumerName, + EnumSet<Consumer.Option> optionSet) throws AMQException { + // Access control if (!getVirtualHost().getSecurityManager().authoriseConsume(this)) { @@ -400,58 +398,61 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener, Mes } - if (hasExclusiveSubscriber()) + if (hasExclusiveConsumer()) { - throw new ExistingExclusiveSubscription(); + throw new ExistingExclusiveConsumer(); } - if (exclusive && !subscription.isTransient()) + + boolean exclusive = optionSet.contains(Consumer.Option.EXCLUSIVE); + boolean isTransient = optionSet.contains(Consumer.Option.TRANSIENT); + + if (exclusive && !isTransient && getConsumerCount() != 0) { - if (getConsumerCount() != 0) - { - throw new ExistingSubscriptionPreventsExclusive(); - } - else - { - _exclusiveSubscriber = subscription; - } + throw new ExistingConsumerPreventsExclusive(); + } + + QueueConsumer consumer = new QueueConsumer(filters, messageClass, + optionSet.contains(Consumer.Option.ACQUIRES), + optionSet.contains(Consumer.Option.SEES_REQUEUES), + consumerName, optionSet.contains(Consumer.Option.TRANSIENT), target); + target.consumerAdded(consumer); + + + if (exclusive && !isTransient) + { + _exclusiveSubscriber = consumer; } - if(subscription.isActive()) + if(consumer.isActive()) { _activeSubscriberCount.incrementAndGet(); } - subscription.setStateListener(this); - subscription.setQueueContext(new QueueContext(_entries.getHead())); + + consumer.setStateListener(this); + consumer.setQueueContext(new QueueContext(_entries.getHead())); if (!isDeleted()) { - subscription.setQueue(this, exclusive); + consumer.setQueue(this, exclusive); if(_nolocal) { - subscription.setNoLocal(_nolocal); + consumer.setNoLocal(_nolocal); } - synchronized (_subscriptionListeners) + synchronized (_consumerListeners) { - for(SubscriptionRegistrationListener listener : _subscriptionListeners) + for(ConsumerRegistrationListener listener : _consumerListeners) { - listener.subscriptionRegistered(this, subscription); + listener.consumerAdded(this, consumer); } } - _subscriptionList.add(subscription); - - //Increment consumerCountHigh if necessary. (un)registerSubscription are both - //synchronized methods so we don't need additional synchronization here - if(_consumerCountHigh.get() < getConsumerCount()) - { - _consumerCountHigh.incrementAndGet(); - } + _consumerList.add(consumer); if (isDeleted()) { - subscription.queueDeleted(this); + consumer.queueDeleted(); } } else @@ -459,42 +460,49 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener, Mes // TODO } - deliverAsync(subscription); + deliverAsync(consumer); + + return consumer; } - public synchronized void unregisterSubscription(final Subscription subscription) throws AMQException + synchronized void unregisterConsumer(final QueueConsumer consumer) throws AMQException { - if (subscription == null) + if (consumer == null) { - throw new NullPointerException("subscription argument is null"); + throw new NullPointerException("consumer argument is null"); } - boolean removed = _subscriptionList.remove(subscription); + boolean removed = _consumerList.remove(consumer); if (removed) { - subscription.close(); + consumer.close(); // No longer can the queue have an exclusive consumer setExclusiveSubscriber(null); - subscription.setQueueContext(null); + consumer.setQueueContext(null); + + if(!isDeleted() && isExclusive() && getConsumerCount() == 0) + { + setAuthorizationHolder(null); + } if(_messageGroupManager != null) { - resetSubPointersForGroups(subscription, true); + resetSubPointersForGroups(consumer, true); } - synchronized (_subscriptionListeners) + synchronized (_consumerListeners) { - for(SubscriptionRegistrationListener listener : _subscriptionListeners) + for(ConsumerRegistrationListener listener : _consumerListeners) { - listener.subscriptionUnregistered(this, subscription); + listener.consumerRemoved(this, consumer); } } // auto-delete queues must be deleted if there are no remaining subscribers - if (_autoDelete && getDeleteOnNoConsumers() && !subscription.isTransient() && getConsumerCount() == 0 ) + if (_autoDelete && getDeleteOnNoConsumers() && !consumer.isTransient() && getConsumerCount() == 0 ) { if (_logger.isInfoEnabled()) { @@ -503,57 +511,57 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener, Mes getVirtualHost().removeQueue(this); - // we need to manually fire the event to the removed subscription (which was the last one left for this - // queue. This is because the delete method uses the subscription set which has just been cleared - subscription.queueDeleted(this); + // we need to manually fire the event to the removed consumer (which was the last one left for this + // queue. This is because the delete method uses the consumer set which has just been cleared + consumer.queueDeleted(); } } } - public Collection<Subscription> getConsumers() + public Collection<QueueConsumer> getConsumers() { - List<Subscription> consumers = new ArrayList<Subscription>(); - SubscriptionList.SubscriptionNodeIterator iter = _subscriptionList.iterator(); + List<QueueConsumer> consumers = new ArrayList<QueueConsumer>(); + QueueConsumerList.ConsumerNodeIterator iter = _consumerList.iterator(); while(iter.advance()) { - consumers.add(iter.getNode().getSubscription()); + consumers.add(iter.getNode().getConsumer()); } return consumers; } - public void addSubscriptionRegistrationListener(final SubscriptionRegistrationListener listener) + public void addConsumerRegistrationListener(final ConsumerRegistrationListener listener) { - synchronized (_subscriptionListeners) + synchronized (_consumerListeners) { - _subscriptionListeners.add(listener); + _consumerListeners.add(listener); } } - public void removeSubscriptionRegistrationListener(final SubscriptionRegistrationListener listener) + public void removeConsumerRegistrationListener(final ConsumerRegistrationListener listener) { - synchronized (_subscriptionListeners) + synchronized (_consumerListeners) { - _subscriptionListeners.remove(listener); + _consumerListeners.remove(listener); } } - public void resetSubPointersForGroups(Subscription subscription, boolean clearAssignments) + public void resetSubPointersForGroups(QueueConsumer consumer, boolean clearAssignments) { - QueueEntry entry = _messageGroupManager.findEarliestAssignedAvailableEntry(subscription); + QueueEntry entry = _messageGroupManager.findEarliestAssignedAvailableEntry(consumer); if(clearAssignments) { - _messageGroupManager.clearAssignments(subscription); + _messageGroupManager.clearAssignments(consumer); } if(entry != null) { - SubscriptionList.SubscriptionNodeIterator subscriberIter = _subscriptionList.iterator(); + QueueConsumerList.ConsumerNodeIterator subscriberIter = _consumerList.iterator(); // iterate over all the subscribers, and if they are in advance of this queue entry then move them backwards while (subscriberIter.advance()) { - Subscription sub = subscriberIter.getNode().getSubscription(); + QueueConsumer sub = subscriberIter.getNode().getConsumer(); // we don't make browsers send the same stuff twice if (sub.seesRequeues()) @@ -617,23 +625,9 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener, Mes } // ------ Enqueue / Dequeue - public void enqueue(ServerMessage message) throws AMQException - { - enqueue(message, null); - } - - public void enqueue(ServerMessage message, PostEnqueueAction action) throws AMQException - { - enqueue(message, false, action); - } - public void enqueue(ServerMessage message, boolean transactional, PostEnqueueAction action) throws AMQException + public void enqueue(ServerMessage message, Action<MessageInstance<? extends Consumer>> action) throws AMQException { - - if(transactional) - { - incrementTxnEnqueueStats(message); - } incrementQueueCount(); incrementQueueSize(message); @@ -641,35 +635,35 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener, Mes QueueEntry entry; - final Subscription exclusiveSub = _exclusiveSubscriber; + final QueueConsumer exclusiveSub = _exclusiveSubscriber; entry = _entries.add(message); if(action != null || (exclusiveSub == null && _queueRunner.isIdle())) { /* - iterate over subscriptions and if any is at the end of the queue and can deliver this message, then deliver the message + iterate over consumers and if any is at the end of the queue and can deliver this message, then deliver the message */ - SubscriptionList.SubscriptionNode node = _subscriptionList.getMarkedNode(); - SubscriptionList.SubscriptionNode nextNode = node.findNext(); + QueueConsumerList.ConsumerNode node = _consumerList.getMarkedNode(); + QueueConsumerList.ConsumerNode nextNode = node.findNext(); if (nextNode == null) { - nextNode = _subscriptionList.getHead().findNext(); + nextNode = _consumerList.getHead().findNext(); } while (nextNode != null) { - if (_subscriptionList.updateMarkedNode(node, nextNode)) + if (_consumerList.updateMarkedNode(node, nextNode)) { break; } else { - node = _subscriptionList.getMarkedNode(); + node = _consumerList.getMarkedNode(); nextNode = node.findNext(); if (nextNode == null) { - nextNode = _subscriptionList.getHead().findNext(); + nextNode = _consumerList.getHead().findNext(); } } } @@ -683,13 +677,13 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener, Mes if (nextNode == null) { loops--; - nextNode = _subscriptionList.getHead(); + nextNode = _consumerList.getHead(); } else { - // if subscription at end, and active, offer - Subscription sub = nextNode.getSubscription(); - deliverToSubscription(sub, entry); + // if consumer at end, and active, offer + QueueConsumer sub = nextNode.getConsumer(); + deliverToConsumer(sub, entry); } nextNode = nextNode.findNext(); @@ -699,7 +693,7 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener, Mes if (entry.isAvailable()) { - checkSubscriptionsNotAheadOfDelivery(entry); + checkConsumersNotAheadOfDelivery(entry); if (exclusiveSub != null) { @@ -715,12 +709,12 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener, Mes if(action != null) { - action.onEnqueue(entry); + action.performAction(entry); } } - private void deliverToSubscription(final Subscription sub, final QueueEntry entry) + private void deliverToConsumer(final QueueConsumer sub, final QueueEntry entry) throws AMQException { @@ -729,14 +723,14 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener, Mes try { if (!sub.isSuspended() - && subscriptionReadyAndHasInterest(sub, entry) + && consumerReadyAndHasInterest(sub, entry) && mightAssign(sub, entry) && !sub.wouldSuspend(entry)) { if (sub.acquires() && !assign(sub, entry)) { // restore credit here that would have been taken away by wouldSuspend since we didn't manage - // to acquire the entry for this subscription + // to acquire the entry for this consumer sub.restoreCredit(entry); } else @@ -752,7 +746,7 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener, Mes } } - private boolean assign(final Subscription sub, final QueueEntry entry) + private boolean assign(final QueueConsumer sub, final QueueEntry entry) { if(_messageGroupManager == null) { @@ -766,17 +760,17 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener, Mes } } - private boolean mightAssign(final Subscription sub, final QueueEntry entry) + private boolean mightAssign(final QueueConsumer sub, final QueueEntry entry) { if(_messageGroupManager == null || !sub.acquires()) { return true; } - Subscription assigned = _messageGroupManager.getAssignedSubscription(entry); + QueueConsumer assigned = _messageGroupManager.getAssignedConsumer(entry); return (assigned == null) || (assigned == sub); } - protected void checkSubscriptionsNotAheadOfDelivery(final QueueEntry entry) + protected void checkConsumersNotAheadOfDelivery(final QueueEntry entry) { // This method is only required for queues which mess with ordering // Simple Queues don't :-) @@ -810,19 +804,7 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener, Mes getAtomicQueueCount().incrementAndGet(); } - private void incrementTxnEnqueueStats(final ServerMessage message) - { - _msgTxnEnqueues.incrementAndGet(); - _byteTxnEnqueues.addAndGet(message.getSize()); - } - - private void incrementTxnDequeueStats(QueueEntry entry) - { - _msgTxnDequeues.incrementAndGet(); - _byteTxnDequeues.addAndGet(entry.getSize()); - } - - private void deliverMessage(final Subscription sub, final QueueEntry entry, boolean batch) + private void deliverMessage(final QueueConsumer sub, final QueueEntry entry, boolean batch) throws AMQException { setLastSeenEntry(sub, entry); @@ -833,15 +815,15 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener, Mes sub.send(entry, batch); } - private boolean subscriptionReadyAndHasInterest(final Subscription sub, final QueueEntry entry) throws AMQException + private boolean consumerReadyAndHasInterest(final QueueConsumer sub, final QueueEntry entry) throws AMQException { return sub.hasInterest(entry) && (getNextAvailableEntry(sub) == entry); } - private void setLastSeenEntry(final Subscription sub, final QueueEntry entry) + private void setLastSeenEntry(final QueueConsumer sub, final QueueEntry entry) { - QueueContext subContext = (QueueContext) sub.getQueueContext(); + QueueContext subContext = sub.getQueueContext(); if (subContext != null) { QueueEntry releasedEntry = subContext.getReleasedEntry(); @@ -854,10 +836,10 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener, Mes } } - private void updateSubRequeueEntry(final Subscription sub, final QueueEntry entry) + private void updateSubRequeueEntry(final QueueConsumer sub, final QueueEntry entry) { - QueueContext subContext = (QueueContext) sub.getQueueContext(); + QueueContext subContext = sub.getQueueContext(); if(subContext != null) { QueueEntry oldEntry; @@ -874,11 +856,11 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener, Mes public void requeue(QueueEntry entry) { - SubscriptionList.SubscriptionNodeIterator subscriberIter = _subscriptionList.iterator(); + QueueConsumerList.ConsumerNodeIterator subscriberIter = _consumerList.iterator(); // iterate over all the subscribers, and if they are in advance of this queue entry then move them backwards while (subscriberIter.advance() && entry.isAvailable()) { - Subscription sub = subscriberIter.getNode().getSubscription(); + QueueConsumer sub = subscriberIter.getNode().getConsumer(); // we don't make browsers send the same stuff twice if (sub.seesRequeues()) @@ -891,20 +873,15 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener, Mes } - public void dequeue(QueueEntry entry, Subscription sub) + public void dequeue(QueueEntry entry, Consumer sub) { decrementQueueCount(); decrementQueueSize(entry); - if (entry.acquiredBySubscription()) + if (entry.acquiredByConsumer()) { _deliveredMessages.decrementAndGet(); } - if(sub != null && sub.isSessionTransactional()) - { - incrementTxnDequeueStats(entry); - } - checkCapacity(); } @@ -928,17 +905,17 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener, Mes _dequeueCount.incrementAndGet(); } - public boolean resend(final QueueEntry entry, final Subscription subscription) throws AMQException + public boolean resend(final QueueEntry entry, final Consumer consumer) throws AMQException { - /* TODO : This is wrong as the subscription may be suspended, we should instead change the state of the message - entry to resend and move back the subscription pointer. */ + /* TODO : This is wrong as the consumer may be suspended, we should instead change the state of the message + entry to resend and move back the consumer pointer. */ - subscription.getSendLock(); + consumer.getSendLock(); try { - if (!subscription.isClosed()) + if (!consumer.isClosed()) { - deliverMessage(subscription, entry, false); + deliverMessage((QueueConsumer) consumer, entry, false); return true; } else @@ -948,7 +925,7 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener, Mes } finally { - subscription.releaseSendLock(); + consumer.releaseSendLock(); } } @@ -956,12 +933,7 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener, Mes public int getConsumerCount() { - return _subscriptionList.size(); - } - - public int getConsumerCountHigh() - { - return _consumerCountHigh.get(); + return _consumerList.size(); } public int getActiveConsumerCount() @@ -1039,16 +1011,16 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener, Mes } - public void stateChange(Subscription sub, Subscription.State oldState, Subscription.State newState) + public void stateChanged(QueueConsumer sub, QueueConsumer.State oldState, QueueConsumer.State newState) { - if (oldState == Subscription.State.ACTIVE && newState != Subscription.State.ACTIVE) + if (oldState == QueueConsumer.State.ACTIVE && newState != QueueConsumer.State.ACTIVE) { _activeSubscriberCount.decrementAndGet(); } - else if (newState == Subscription.State.ACTIVE) + else if (newState == QueueConsumer.State.ACTIVE) { - if (oldState != Subscription.State.ACTIVE) + if (oldState != QueueConsumer.State.ACTIVE) { _activeSubscriberCount.incrementAndGet(); @@ -1072,12 +1044,12 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener, Mes return _atomicQueueSize; } - public boolean hasExclusiveSubscriber() + public boolean hasExclusiveConsumer() { return _exclusiveSubscriber != null; } - private void setExclusiveSubscriber(Subscription exclusiveSubscriber) + private void setExclusiveSubscriber(QueueConsumer exclusiveSubscriber) { _exclusiveSubscriber = exclusiveSubscriber; } @@ -1093,9 +1065,9 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener, Mes return _entries; } - protected SubscriptionList getSubscriptionList() + protected QueueConsumerList getConsumerList() { - return _subscriptionList; + return _consumerList; } @@ -1300,12 +1272,12 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener, Mes }); } - public void addQueueDeleteTask(final Task task) + public void addQueueDeleteTask(final Action<AMQQueue> task) { _deleteTaskList.add(task); } - public void removeQueueDeleteTask(final Task task) + public void removeQueueDeleteTask(final Action<AMQQueue> task) { _deleteTaskList.remove(task); } @@ -1322,19 +1294,21 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener, Mes if (!_deleted.getAndSet(true)) { - for (Binding b : _bindings) + final ArrayList<Binding> bindingCopy = new ArrayList<Binding>(_bindings); + + for (Binding b : bindingCopy) { b.getExchange().removeBinding(b); } - SubscriptionList.SubscriptionNodeIterator subscriptionIter = _subscriptionList.iterator(); + QueueConsumerList.ConsumerNodeIterator consumerNodeIterator = _consumerList.iterator(); - while (subscriptionIter.advance()) + while (consumerNodeIterator.advance()) { - Subscription s = subscriptionIter.getNode().getSubscription(); + QueueConsumer s = consumerNodeIterator.getNode().getConsumer(); if (s != null) { - s.queueDeleted(this); + s.queueDeleted(); } } @@ -1375,9 +1349,9 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener, Mes } - for (Task task : _deleteTaskList) + for (Action<AMQQueue> task : _deleteTaskList) { - task.doTask(this); + task.performAction(this); } _deleteTaskList.clear(); @@ -1461,7 +1435,7 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener, Mes } - public void deliverAsync(Subscription sub) + public void deliverAsync(QueueConsumer sub) { if(_exclusiveSubscriber == null) { @@ -1469,28 +1443,23 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener, Mes } else { - SubFlushRunner flusher = (SubFlushRunner) sub.get(SUB_FLUSH_RUNNER); - if(flusher == null) - { - flusher = new SubFlushRunner(sub); - sub.set(SUB_FLUSH_RUNNER, flusher); - } + SubFlushRunner flusher = sub.getRunner(); flusher.execute(_asyncDelivery); } } - public void flushSubscription(Subscription sub) throws AMQException + void flushConsumer(QueueConsumer sub) throws AMQException { // Access control if (!getVirtualHost().getSecurityManager().authoriseConsume(this)) { throw new AMQSecurityException("Permission denied: " + getName()); } - flushSubscription(sub, Long.MAX_VALUE); + flushConsumer(sub, Long.MAX_VALUE); } - public boolean flushSubscription(Subscription sub, long iterations) throws AMQException + boolean flushConsumer(QueueConsumer sub, long iterations) throws AMQException { boolean atTail = false; final boolean keepSendLockHeld = iterations <= SimpleAMQQueue.MAX_ASYNC_DELIVERIES; @@ -1511,8 +1480,8 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener, Mes sub.getSendLock(); } - atTail = attemptDelivery(sub, true); - if (atTail && getNextAvailableEntry(sub) == null) + atTail = attemptDelivery((QueueConsumer)sub, true); + if (atTail && getNextAvailableEntry((QueueConsumer)sub) == null) { queueEmpty = true; } @@ -1546,21 +1515,21 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener, Mes } - // if there's (potentially) more than one subscription the others will potentially not have been advanced to the + // if there's (potentially) more than one consumer the others will potentially not have been advanced to the // next entry they are interested in yet. This would lead to holding on to references to expired messages, etc // which would give us memory "leak". - if (!hasExclusiveSubscriber()) + if (!hasExclusiveConsumer()) { - advanceAllSubscriptions(); + advanceAllConsumers(); } return atTail; } /** - * Attempt delivery for the given subscription. + * Attempt delivery for the given consumer. * - * Looks up the next node for the subscription and attempts to deliver it. + * Looks up the next node for the consumer and attempts to deliver it. * * * @param sub @@ -1568,7 +1537,7 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener, Mes * @return true if we have completed all possible deliveries for this sub. * @throws AMQException */ - private boolean attemptDelivery(Subscription sub, boolean batch) throws AMQException + private boolean attemptDelivery(QueueConsumer sub, boolean batch) throws AMQException { boolean atTail = false; @@ -1587,7 +1556,7 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener, Mes if (sub.acquires() && !assign(sub, node)) { // restore credit here that would have been taken away by wouldSuspend since we didn't manage - // to acquire the entry for this subscription + // to acquire the entry for this consumer sub.restoreCredit(node); } else @@ -1598,7 +1567,7 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener, Mes } else // Not enough Credit for message and wouldSuspend { - //QPID-1187 - Treat the subscription as suspended for this message + //QPID-1187 - Treat the consumer as suspended for this message // and wait for the message to be removed to continue delivery. subActive = false; node.addStateChangeListener(new QueueEntryListener(sub)); @@ -1611,13 +1580,13 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener, Mes return atTail || !subActive; } - protected void advanceAllSubscriptions() throws AMQException + protected void advanceAllConsumers() throws AMQException { - SubscriptionList.SubscriptionNodeIterator subscriberIter = _subscriptionList.iterator(); - while (subscriberIter.advance()) + QueueConsumerList.ConsumerNodeIterator consumerNodeIterator = _consumerList.iterator(); + while (consumerNodeIterator.advance()) { - SubscriptionList.SubscriptionNode subNode = subscriberIter.getNode(); - Subscription sub = subNode.getSubscription(); + QueueConsumerList.ConsumerNode subNode = consumerNodeIterator.getNode(); + QueueConsumer sub = subNode.getConsumer(); if(sub.acquires()) { getNextAvailableEntry(sub); @@ -1629,10 +1598,10 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener, Mes } } - private QueueEntry getNextAvailableEntry(final Subscription sub) + private QueueEntry getNextAvailableEntry(final QueueConsumer sub) throws AMQException { - QueueContext context = (QueueContext) sub.getQueueContext(); + QueueContext context = sub.getQueueContext(); if(context != null) { QueueEntry lastSeen = context.getLastSeenEntry(); @@ -1670,9 +1639,9 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener, Mes } } - public boolean isEntryAheadOfSubscription(QueueEntry entry, Subscription sub) + public boolean isEntryAheadOfConsumer(QueueEntry entry, QueueConsumer sub) { - QueueContext context = (QueueContext) sub.getQueueContext(); + QueueContext context = sub.getQueueContext(); if(context != null) { QueueEntry releasedNode = context.getReleasedEntry(); @@ -1689,14 +1658,14 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener, Mes * * A queue Runner is started whenever a state change occurs, e.g when a new * message arrives on the queue and cannot be immediately delivered to a - * subscription (i.e. asynchronous delivery is required). Unless there are - * SubFlushRunners operating (due to subscriptions unsuspending) which are + * consumer (i.e. asynchronous delivery is required). Unless there are + * SubFlushRunners operating (due to consumers unsuspending) which are * capable of accepting/delivering all messages then these messages would * otherwise remain on the queue. * * processQueue should be running while there are messages on the queue AND - * there are subscriptions that can deliver them. If there are no - * subscriptions capable of delivering the remaining messages on the queue + * there are consumers that can deliver them. If there are no + * consumers capable of delivering the remaining messages on the queue * then processQueue should stop to prevent spinning. * * Since processQueue is runs in a fixed size Executor, it should not run @@ -1720,7 +1689,7 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener, Mes boolean lastLoop = false; int iterations = MAX_ASYNC_DELIVERIES; - final int numSubs = _subscriptionList.size(); + final int numSubs = _consumerList.size(); final int perSub = Math.max(iterations / Math.max(numSubs,1), 1); @@ -1731,8 +1700,8 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener, Mes // So whilst delivery/rejection is going on a processQueue thread will be running while (iterations != 0 && ((previousStateChangeCount != (stateChangeCount = _stateChangeCount.get())) || deliveryIncomplete)) { - // we want to have one extra loop after every subscription has reached the point where it cannot move - // further, just in case the advance of one subscription in the last loop allows a different subscription to + // we want to have one extra loop after every consumer has reached the point where it cannot move + // further, just in case the advance of one consumer in the last loop allows a different consumer to // move forward in the next iteration if (previousStateChangeCount != stateChangeCount) @@ -1744,14 +1713,14 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener, Mes } previousStateChangeCount = stateChangeCount; - boolean allSubscriptionsDone = true; - boolean subscriptionDone; + boolean allConsumersDone = true; + boolean consumerDone; - SubscriptionList.SubscriptionNodeIterator subscriptionIter = _subscriptionList.iterator(); + QueueConsumerList.ConsumerNodeIterator consumerNodeIterator = _consumerList.iterator(); //iterate over the subscribers and try to advance their pointer - while (subscriptionIter.advance()) + while (consumerNodeIterator.advance()) { - Subscription sub = subscriptionIter.getNode().getSubscription(); + QueueConsumer sub = consumerNodeIterator.getNode().getConsumer(); sub.getSendLock(); try @@ -1759,8 +1728,8 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener, Mes for(int i = 0 ; i < perSub; i++) { //attempt delivery. returns true if no further delivery currently possible to this sub - subscriptionDone = attemptDelivery(sub, true); - if (subscriptionDone) + consumerDone = attemptDelivery(sub, true); + if (consumerDone) { sub.flushBatched(); if (lastLoop && !sub.isSuspended()) @@ -1771,9 +1740,9 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener, Mes } else { - //this subscription can accept additional deliveries, so we must + //this consumer can accept additional deliveries, so we must //keep going after this (if iteration slicing allows it) - allSubscriptionsDone = false; + allConsumersDone = false; lastLoop = false; if(--iterations == 0) { @@ -1792,24 +1761,24 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener, Mes } } - if(allSubscriptionsDone && lastLoop) + if(allConsumersDone && lastLoop) { //We have done an extra loop already and there are again //again no further delivery attempts possible, only //keep going if state change demands it. deliveryIncomplete = false; } - else if(allSubscriptionsDone) + else if(allConsumersDone) { - //All subscriptions reported being done, but we have to do + //All consumers reported being done, but we have to do //an extra loop if the iterations are not exhausted and //there is still any work to be done - deliveryIncomplete = _subscriptionList.size() != 0; + deliveryIncomplete = _consumerList.size() != 0; lastLoop = true; } else { - //some subscriptions can still accept more messages, + //some consumers can still accept more messages, //keep going if iteration count allows. lastLoop = false; deliveryIncomplete = true; @@ -1984,12 +1953,12 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener, Mes return _notificationChecks; } - private final class QueueEntryListener implements QueueEntry.StateChangeListener + private final class QueueEntryListener implements StateChangeListener<MessageInstance<QueueConsumer>, QueueEntry.State> { - private final Subscription _sub; + private final QueueConsumer _sub; - public QueueEntryListener(final Subscription sub) + public QueueEntryListener(final QueueConsumer sub) { _sub = sub; } @@ -2005,7 +1974,7 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener, Mes return System.identityHashCode(_sub); } - public void stateChanged(QueueEntry entry, QueueEntry.State oldSate, QueueEntry.State newState) + public void stateChanged(MessageInstance entry, QueueEntry.State oldSate, QueueEntry.State newState) { entry.removeStateChangeListener(this); deliverAsync(_sub); @@ -2076,26 +2045,6 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener, Mes return _dequeueSize.get(); } - public long getByteTxnEnqueues() - { - return _byteTxnEnqueues.get(); - } - - public long getByteTxnDequeues() - { - return _byteTxnDequeues.get(); - } - - public long getMsgTxnEnqueues() - { - return _msgTxnEnqueues.get(); - } - - public long getMsgTxnDequeues() - { - return _msgTxnDequeues.get(); - } - public long getPersistentByteEnqueues() { return _persistentMessageEnqueueSize.get(); @@ -2123,11 +2072,6 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener, Mes return getName(); } - public long getUnackedMessageCountHigh() - { - return _unackedMsgCountHigh.get(); - } - public long getUnackedMessageCount() { return _unackedMsgCount.get(); @@ -2146,17 +2090,8 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener, Mes private void incrementUnackedMsgCount(QueueEntry entry) { - long unackedMsgCount = _unackedMsgCount.incrementAndGet(); + _unackedMsgCount.incrementAndGet(); _unackedMsgBytes.addAndGet(entry.getSize()); - - long unackedMsgCountHigh; - while(unackedMsgCount > (unackedMsgCountHigh = _unackedMsgCountHigh.get())) - { - if(_unackedMsgCountHigh.compareAndSet(unackedMsgCountHigh, unackedMsgCount)) - { - break; - } - } } public LogActor getLogActor() @@ -2224,4 +2159,39 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener, Mes return (String) _arguments.get(Queue.DESCRIPTION); } + public final int send(final ServerMessage message, + final InstanceProperties instanceProperties, + final ServerTransaction txn, + final Action<MessageInstance<? extends Consumer>> postEnqueueAction) + { + txn.enqueue(this,message, new ServerTransaction.Action() + { + MessageReference _reference = message.newReference(); + + public void postCommit() + { + try + { + SimpleAMQQueue.this.enqueue(message, postEnqueueAction); + } + catch (AMQException e) + { + // TODO + throw new RuntimeException(e); + } + finally + { + _reference.release(); + } + } + + public void onRollback() + { + _reference.release(); + } + }); + return 1; + + } + } diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/SimpleQueueEntryList.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/SimpleQueueEntryList.java index b8d8ec19f4..101771c7cc 100644 --- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/SimpleQueueEntryList.java +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/SimpleQueueEntryList.java @@ -39,7 +39,7 @@ public class SimpleQueueEntryList implements QueueEntryList<SimpleQueueEntryImpl (SimpleQueueEntryList.class, SimpleQueueEntryImpl.class, "_tail"); - private final AMQQueue _queue; + private final AMQQueue<QueueConsumer> _queue; static final AtomicReferenceFieldUpdater<SimpleQueueEntryImpl, SimpleQueueEntryImpl> _nextUpdater = SimpleQueueEntryImpl._nextUpdater; @@ -49,7 +49,7 @@ public class SimpleQueueEntryList implements QueueEntryList<SimpleQueueEntryImpl private final AtomicReference<SimpleQueueEntryImpl> _unscavengedHWM = new AtomicReference<SimpleQueueEntryImpl>(); - public SimpleQueueEntryList(AMQQueue queue) + public SimpleQueueEntryList(AMQQueue<QueueConsumer> queue) { _queue = queue; _head = new SimpleQueueEntryImpl(this); @@ -71,7 +71,7 @@ public class SimpleQueueEntryList implements QueueEntryList<SimpleQueueEntryImpl } - public AMQQueue getQueue() + public AMQQueue<QueueConsumer> getQueue() { return _queue; } diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/SortedQueue.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/SortedQueue.java index b3566df0c4..cad1aa6d4f 100644 --- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/SortedQueue.java +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/SortedQueue.java @@ -20,7 +20,10 @@ package org.apache.qpid.server.queue; import org.apache.qpid.AMQException; +import org.apache.qpid.server.consumer.Consumer; +import org.apache.qpid.server.message.MessageInstance; import org.apache.qpid.server.message.ServerMessage; +import org.apache.qpid.server.util.Action; import org.apache.qpid.server.virtualhost.VirtualHost; import java.util.Map; @@ -29,7 +32,7 @@ import java.util.UUID; public class SortedQueue extends OutOfOrderQueue { //Lock object to synchronize enqueue. Used instead of the object - //monitor to prevent lock order issues with subscription sendLocks + //monitor to prevent lock order issues with consumer sendLocks //and consumer updates in the super classes private final Object _sortedQueueLock = new Object(); private final String _sortedPropertyName; @@ -48,7 +51,7 @@ public class SortedQueue extends OutOfOrderQueue return _sortedPropertyName; } - public void enqueue(ServerMessage message, PostEnqueueAction action) throws AMQException + public void enqueue(ServerMessage message, Action<MessageInstance<? extends Consumer>> action) throws AMQException { synchronized (_sortedQueueLock) { diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/SortedQueueEntryList.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/SortedQueueEntryList.java index 85559157a9..336ee566eb 100644 --- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/SortedQueueEntryList.java +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/SortedQueueEntryList.java @@ -28,7 +28,7 @@ import org.apache.qpid.server.queue.SortedQueueEntryImpl.Colour; * Uses the red/black tree algorithm specified in "Introduction to Algorithms". * ISBN-10: 0262033844 * ISBN-13: 978-0262033848 - * @see http://en.wikipedia.org/wiki/Red-black_tree + * see http://en.wikipedia.org/wiki/Red-black_tree */ public class SortedQueueEntryList implements QueueEntryList<SortedQueueEntryImpl> { @@ -36,17 +36,17 @@ public class SortedQueueEntryList implements QueueEntryList<SortedQueueEntryImpl private SortedQueueEntryImpl _root; private long _entryId = Long.MIN_VALUE; private final Object _lock = new Object(); - private final AMQQueue _queue; + private final AMQQueue<QueueConsumer> _queue; private final String _propertyName; - public SortedQueueEntryList(final AMQQueue queue, final String propertyName) + public SortedQueueEntryList(final AMQQueue<QueueConsumer> queue, final String propertyName) { _queue = queue; _head = new SortedQueueEntryImpl(this); _propertyName = propertyName; } - public AMQQueue getQueue() + public AMQQueue<QueueConsumer> getQueue() { return _queue; } diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/SubFlushRunner.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/SubFlushRunner.java index 47a7d733dd..c30a48b03a 100755 --- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/SubFlushRunner.java +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/SubFlushRunner.java @@ -25,7 +25,6 @@ import org.apache.log4j.Logger; import org.apache.qpid.AMQException; import org.apache.qpid.server.logging.actors.CurrentActor; -import org.apache.qpid.server.subscription.Subscription; import org.apache.qpid.transport.TransportException; import java.util.concurrent.Executor; @@ -38,7 +37,7 @@ class SubFlushRunner implements Runnable private static final Logger _logger = Logger.getLogger(SubFlushRunner.class); - private final Subscription _sub; + private final QueueConsumer _sub; private static int IDLE = 0; private static int SCHEDULED = 1; @@ -51,7 +50,7 @@ class SubFlushRunner implements Runnable private static final long ITERATIONS = SimpleAMQQueue.MAX_ASYNC_DELIVERIES; private final AtomicBoolean _stateChange = new AtomicBoolean(); - public SubFlushRunner(Subscription sub) + public SubFlushRunner(QueueConsumer sub) { _sub = sub; } @@ -65,7 +64,7 @@ class SubFlushRunner implements Runnable try { CurrentActor.set(_sub.getLogActor()); - complete = getQueue().flushSubscription(_sub, ITERATIONS); + complete = getQueue().flushConsumer(_sub, ITERATIONS); } catch (AMQException e) { diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/store/AbstractJDBCMessageStore.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/store/AbstractJDBCMessageStore.java index e490ac38c1..d80fa656e7 100644 --- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/store/AbstractJDBCMessageStore.java +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/store/AbstractJDBCMessageStore.java @@ -1065,7 +1065,7 @@ abstract public class AbstractJDBCMessageStore implements MessageStore, DurableC stmt.setString(4, "E"); for(Transaction.Record record : enqueues) { - stmt.setString(5, record.getQueue().getId().toString()); + stmt.setString(5, record.getResource().getId().toString()); stmt.setLong(6, record.getMessage().getMessageNumber()); stmt.executeUpdate(); } @@ -1076,7 +1076,7 @@ abstract public class AbstractJDBCMessageStore implements MessageStore, DurableC stmt.setString(4, "D"); for(Transaction.Record record : dequeues) { - stmt.setString(5, record.getQueue().getId().toString()); + stmt.setString(5, record.getResource().getId().toString()); stmt.setLong(6, record.getMessage().getMessageNumber()); stmt.executeUpdate(); } @@ -1199,7 +1199,7 @@ abstract public class AbstractJDBCMessageStore implements MessageStore, DurableC buf.position(1); buf = buf.slice(); - metaData.writeToBuffer(0, buf); + metaData.writeToBuffer(buf); ByteArrayInputStream bis = new ByteArrayInputStream(underlying); try { @@ -1371,7 +1371,7 @@ abstract public class AbstractJDBCMessageStore implements MessageStore, DurableC } @Override - public TransactionLogResource getQueue() + public TransactionLogResource getResource() { return this; } @@ -1401,10 +1401,22 @@ abstract public class AbstractJDBCMessageStore implements MessageStore, DurableC } @Override + public String getName() + { + return _queueId.toString(); + } + + @Override public UUID getId() { return _queueId; } + + @Override + public boolean isDurable() + { + return true; + } } protected void recoverXids(TransactionLogRecoveryHandler.DtxRecordRecoveryHandler dtxrh) throws SQLException diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/store/DurableConfigurationStoreHelper.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/store/DurableConfigurationStoreHelper.java index a688b493e1..de7369f5ed 100644 --- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/store/DurableConfigurationStoreHelper.java +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/store/DurableConfigurationStoreHelper.java @@ -29,6 +29,7 @@ import java.util.Map; import java.util.Set; import org.apache.qpid.AMQStoreException; +import org.apache.qpid.server.consumer.Consumer; import org.apache.qpid.server.model.Binding; import org.apache.qpid.server.model.Exchange; import org.apache.qpid.server.model.LifetimePolicy; @@ -46,7 +47,7 @@ public class DurableConfigurationStoreHelper Queue.EXCLUSIVE, Queue.ALTERNATE_EXCHANGE)); - public static void updateQueue(DurableConfigurationStore store, AMQQueue queue) throws AMQStoreException + public static void updateQueue(DurableConfigurationStore store, AMQQueue<? extends Consumer> queue) throws AMQStoreException { Map<String, Object> attributesMap = new LinkedHashMap<String, Object>(); attributesMap.put(Queue.NAME, queue.getName()); @@ -71,7 +72,7 @@ public class DurableConfigurationStoreHelper store.update(queue.getId(), QUEUE, attributesMap); } - public static void createQueue(DurableConfigurationStore store, AMQQueue queue) + public static void createQueue(DurableConfigurationStore store, AMQQueue<? extends Consumer> queue) throws AMQStoreException { Map<String, Object> attributesMap = new HashMap<String, Object>(); diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/store/StorableMessageMetaData.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/store/StorableMessageMetaData.java index 9ae6cca8e6..4fd452649d 100755 --- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/store/StorableMessageMetaData.java +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/store/StorableMessageMetaData.java @@ -29,7 +29,7 @@ public interface StorableMessageMetaData int getStorableSize(); - int writeToBuffer(int offsetInMetaData, ByteBuffer dest); + int writeToBuffer(ByteBuffer dest); int getContentSize(); diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/store/Transaction.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/store/Transaction.java index 66bcfff32b..74b91dec2d 100644 --- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/store/Transaction.java +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/store/Transaction.java @@ -70,7 +70,7 @@ public interface Transaction public static interface Record { - TransactionLogResource getQueue(); + TransactionLogResource getResource(); EnqueueableMessage getMessage(); } diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/store/TransactionLogResource.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/store/TransactionLogResource.java index 576dca847d..18b3125641 100755 --- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/store/TransactionLogResource.java +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/store/TransactionLogResource.java @@ -24,5 +24,7 @@ import java.util.UUID; public interface TransactionLogResource { + String getName(); public UUID getId(); + boolean isDurable(); } diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/subscription/ClientDeliveryMethod.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/subscription/ClientDeliveryMethod.java deleted file mode 100644 index 45a1978af1..0000000000 --- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/subscription/ClientDeliveryMethod.java +++ /dev/null @@ -1,31 +0,0 @@ -/* -* -* 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.subscription; - -import org.apache.qpid.AMQException; -import org.apache.qpid.server.message.InstanceProperties; -import org.apache.qpid.server.message.ServerMessage; - -public interface ClientDeliveryMethod -{ - void deliverToClient(final Subscription sub, final ServerMessage message, final InstanceProperties props, - final long deliveryTag) throws AMQException; -} diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/subscription/RecordDeliveryMethod.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/subscription/RecordDeliveryMethod.java deleted file mode 100644 index e2ed4104de..0000000000 --- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/subscription/RecordDeliveryMethod.java +++ /dev/null @@ -1,28 +0,0 @@ -/* -* -* 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.subscription; - -import org.apache.qpid.server.queue.QueueEntry; - -public interface RecordDeliveryMethod -{ - void recordMessageDelivery(final Subscription sub, final QueueEntry entry, final long deliveryTag); -} diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/txn/AsyncAutoCommitTransaction.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/txn/AsyncAutoCommitTransaction.java index 1f5a4907ed..57c67f54cd 100755 --- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/txn/AsyncAutoCommitTransaction.java +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/txn/AsyncAutoCommitTransaction.java @@ -25,12 +25,14 @@ import org.apache.log4j.Logger; import org.apache.qpid.AMQException; import org.apache.qpid.AMQStoreException; import org.apache.qpid.server.message.EnqueueableMessage; +import org.apache.qpid.server.message.MessageInstance; import org.apache.qpid.server.message.ServerMessage; import org.apache.qpid.server.queue.BaseQueue; import org.apache.qpid.server.queue.QueueEntry; import org.apache.qpid.server.store.MessageStore; import org.apache.qpid.server.store.StoreFuture; import org.apache.qpid.server.store.Transaction; +import org.apache.qpid.server.store.TransactionLogResource; import java.util.Collection; import java.util.List; @@ -88,7 +90,7 @@ public class AsyncAutoCommitTransaction implements ServerTransaction } - public void dequeue(BaseQueue queue, EnqueueableMessage message, Action postTransactionAction) + public void dequeue(TransactionLogResource queue, EnqueueableMessage message, Action postTransactionAction) { Transaction txn = null; try @@ -158,15 +160,15 @@ public class AsyncAutoCommitTransaction implements ServerTransaction } } - public void dequeue(Collection<QueueEntry> queueEntries, Action postTransactionAction) + public void dequeue(Collection<MessageInstance> queueEntries, Action postTransactionAction) { Transaction txn = null; try { - for(QueueEntry entry : queueEntries) + for(MessageInstance entry : queueEntries) { ServerMessage message = entry.getMessage(); - BaseQueue queue = entry.getQueue(); + TransactionLogResource queue = entry.getOwningResource(); if(message.isPersistent() && queue.isDurable()) { @@ -210,7 +212,7 @@ public class AsyncAutoCommitTransaction implements ServerTransaction } - public void enqueue(BaseQueue queue, EnqueueableMessage message, Action postTransactionAction) + public void enqueue(TransactionLogResource queue, EnqueueableMessage message, Action postTransactionAction) { Transaction txn = null; try diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/txn/AutoCommitTransaction.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/txn/AutoCommitTransaction.java index b057998456..4ea48c6a24 100755 --- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/txn/AutoCommitTransaction.java +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/txn/AutoCommitTransaction.java @@ -25,11 +25,13 @@ import org.apache.log4j.Logger; import org.apache.qpid.AMQException; import org.apache.qpid.AMQStoreException; import org.apache.qpid.server.message.EnqueueableMessage; +import org.apache.qpid.server.message.MessageInstance; import org.apache.qpid.server.message.ServerMessage; import org.apache.qpid.server.queue.BaseQueue; import org.apache.qpid.server.queue.QueueEntry; import org.apache.qpid.server.store.MessageStore; import org.apache.qpid.server.store.Transaction; +import org.apache.qpid.server.store.TransactionLogResource; import java.util.Collection; import java.util.List; @@ -73,7 +75,7 @@ public class AutoCommitTransaction implements ServerTransaction immediateAction.postCommit(); } - public void dequeue(BaseQueue queue, EnqueueableMessage message, Action postTransactionAction) + public void dequeue(TransactionLogResource queue, EnqueueableMessage message, Action postTransactionAction) { Transaction txn = null; try @@ -105,15 +107,15 @@ public class AutoCommitTransaction implements ServerTransaction } - public void dequeue(Collection<QueueEntry> queueEntries, Action postTransactionAction) + public void dequeue(Collection<MessageInstance> queueEntries, Action postTransactionAction) { Transaction txn = null; try { - for(QueueEntry entry : queueEntries) + for(MessageInstance entry : queueEntries) { ServerMessage message = entry.getMessage(); - BaseQueue queue = entry.getQueue(); + TransactionLogResource queue = entry.getOwningResource(); if(message.isPersistent() && queue.isDurable()) { @@ -152,7 +154,7 @@ public class AutoCommitTransaction implements ServerTransaction } - public void enqueue(BaseQueue queue, EnqueueableMessage message, Action postTransactionAction) + public void enqueue(TransactionLogResource queue, EnqueueableMessage message, Action postTransactionAction) { Transaction txn = null; try diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/txn/DistributedTransaction.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/txn/DistributedTransaction.java index d48b09d912..4a7c16a7cd 100644 --- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/txn/DistributedTransaction.java +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/txn/DistributedTransaction.java @@ -22,10 +22,12 @@ package org.apache.qpid.server.txn; import org.apache.qpid.server.message.EnqueueableMessage; +import org.apache.qpid.server.message.MessageInstance; import org.apache.qpid.server.protocol.AMQSessionModel; import org.apache.qpid.server.queue.BaseQueue; import org.apache.qpid.server.queue.QueueEntry; import org.apache.qpid.server.store.MessageStore; +import org.apache.qpid.server.store.TransactionLogResource; import org.apache.qpid.server.virtualhost.VirtualHost; import org.apache.qpid.transport.Xid; @@ -74,7 +76,7 @@ public class DistributedTransaction implements ServerTransaction } } - public void dequeue(BaseQueue queue, EnqueueableMessage message, Action postTransactionAction) + public void dequeue(TransactionLogResource queue, EnqueueableMessage message, Action postTransactionAction) { if(_branch != null) { @@ -87,13 +89,13 @@ public class DistributedTransaction implements ServerTransaction } } - public void dequeue(Collection<QueueEntry> messages, Action postTransactionAction) + public void dequeue(Collection<MessageInstance> messages, Action postTransactionAction) { if(_branch != null) { - for(QueueEntry entry : messages) + for(MessageInstance entry : messages) { - _branch.dequeue(entry.getQueue(), entry.getMessage()); + _branch.dequeue(entry.getOwningResource(), entry.getMessage()); } _branch.addPostTransactionAction(postTransactionAction); } @@ -103,7 +105,7 @@ public class DistributedTransaction implements ServerTransaction } } - public void enqueue(BaseQueue queue, EnqueueableMessage message, Action postTransactionAction) + public void enqueue(TransactionLogResource queue, EnqueueableMessage message, Action postTransactionAction) { if(_branch != null) { diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/txn/DtxBranch.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/txn/DtxBranch.java index 6b12862690..2505548ab8 100644 --- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/txn/DtxBranch.java +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/txn/DtxBranch.java @@ -34,6 +34,7 @@ import org.apache.qpid.server.protocol.AMQSessionModel; import org.apache.qpid.server.queue.BaseQueue; import org.apache.qpid.server.store.MessageStore; import org.apache.qpid.server.store.Transaction; +import org.apache.qpid.server.store.TransactionLogResource; import org.apache.qpid.server.virtualhost.VirtualHost; import org.apache.qpid.transport.Xid; @@ -335,7 +336,7 @@ public class DtxBranch { if(enqueue.isDurable()) { - _transaction.enqueueMessage(enqueue.getQueue(), enqueue.getMessage()); + _transaction.enqueueMessage(enqueue.getResource(), enqueue.getMessage()); } } @@ -344,7 +345,7 @@ public class DtxBranch { if(enqueue.isDurable()) { - _transaction.dequeueMessage(enqueue.getQueue(), enqueue.getMessage()); + _transaction.dequeueMessage(enqueue.getResource(), enqueue.getMessage()); } } } @@ -356,31 +357,31 @@ public class DtxBranch } - public void dequeue(BaseQueue queue, EnqueueableMessage message) + public void dequeue(TransactionLogResource resource, EnqueueableMessage message) { - _dequeueRecords.add(new Record(queue, message)); + _dequeueRecords.add(new Record(resource, message)); } - public void enqueue(BaseQueue queue, EnqueueableMessage message) + public void enqueue(TransactionLogResource queue, EnqueueableMessage message) { _enqueueRecords.add(new Record(queue, message)); } private static final class Record implements Transaction.Record { - private final BaseQueue _queue; + private final TransactionLogResource _resource; private final EnqueueableMessage _message; - public Record(BaseQueue queue, EnqueueableMessage message) + public Record(TransactionLogResource resource, EnqueueableMessage message) { - _queue = queue; + _resource = resource; _message = message; } - public BaseQueue getQueue() + public TransactionLogResource getResource() { - return _queue; + return _resource; } public EnqueueableMessage getMessage() @@ -390,7 +391,7 @@ public class DtxBranch public boolean isDurable() { - return _message.isPersistent() && _queue.isDurable(); + return _message.isPersistent() && _resource.isDurable(); } } diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/txn/LocalTransaction.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/txn/LocalTransaction.java index 81aabc6bd3..4b02d4f8ec 100755 --- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/txn/LocalTransaction.java +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/txn/LocalTransaction.java @@ -21,7 +21,9 @@ package org.apache.qpid.server.txn; import org.apache.qpid.server.message.EnqueueableMessage; +import org.apache.qpid.server.message.MessageInstance; import org.apache.qpid.server.store.StoreFuture; +import org.apache.qpid.server.store.TransactionLogResource; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -91,7 +93,7 @@ public class LocalTransaction implements ServerTransaction _postTransactionActions.add(postTransactionAction); } - public void dequeue(BaseQueue queue, EnqueueableMessage message, Action postTransactionAction) + public void dequeue(TransactionLogResource queue, EnqueueableMessage message, Action postTransactionAction) { sync(); _postTransactionActions.add(postTransactionAction); @@ -118,7 +120,7 @@ public class LocalTransaction implements ServerTransaction } } - public void dequeue(Collection<QueueEntry> queueEntries, Action postTransactionAction) + public void dequeue(Collection<MessageInstance> queueEntries, Action postTransactionAction) { sync(); _postTransactionActions.add(postTransactionAction); @@ -126,10 +128,10 @@ public class LocalTransaction implements ServerTransaction try { - for(QueueEntry entry : queueEntries) + for(MessageInstance entry : queueEntries) { ServerMessage message = entry.getMessage(); - BaseQueue queue = entry.getQueue(); + TransactionLogResource queue = entry.getOwningResource(); if(message.isPersistent() && queue.isDurable()) { @@ -195,7 +197,7 @@ public class LocalTransaction implements ServerTransaction } } - public void enqueue(BaseQueue queue, EnqueueableMessage message, Action postTransactionAction) + public void enqueue(TransactionLogResource queue, EnqueueableMessage message, Action postTransactionAction) { sync(); _postTransactionActions.add(postTransactionAction); diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/txn/ServerTransaction.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/txn/ServerTransaction.java index 240ad154ba..cae5fa73bf 100755 --- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/txn/ServerTransaction.java +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/txn/ServerTransaction.java @@ -24,8 +24,9 @@ import java.util.Collection; import java.util.List; import org.apache.qpid.server.message.EnqueueableMessage; +import org.apache.qpid.server.message.MessageInstance; import org.apache.qpid.server.queue.BaseQueue; -import org.apache.qpid.server.queue.QueueEntry; +import org.apache.qpid.server.store.TransactionLogResource; /** @@ -79,21 +80,21 @@ public interface ServerTransaction * * A store operation will result only for a persistent message on a durable queue. */ - void dequeue(BaseQueue queue, EnqueueableMessage message, Action postTransactionAction); + void dequeue(TransactionLogResource queue, EnqueueableMessage message, Action postTransactionAction); /** * Dequeue a message(s) from queue(s) registering a post transaction action. * * Store operations will result only for a persistent messages on durable queues. */ - void dequeue(Collection<QueueEntry> messages, Action postTransactionAction); + void dequeue(Collection<MessageInstance> messages, Action postTransactionAction); /** * Enqueue a message to a queue registering a post transaction action. * * A store operation will result only for a persistent message on a durable queue. */ - void enqueue(BaseQueue queue, EnqueueableMessage message, Action postTransactionAction); + void enqueue(TransactionLogResource queue, EnqueueableMessage message, Action postTransactionAction); /** * Enqueue a message(s) to queue(s) registering a post transaction action. diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/util/Action.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/util/Action.java new file mode 100644 index 0000000000..0d53b4d03b --- /dev/null +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/util/Action.java @@ -0,0 +1,26 @@ +/* + * + * 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.util; + +public interface Action<T> +{ + void performAction(T object); +} diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/util/StateChangeListener.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/util/StateChangeListener.java new file mode 100644 index 0000000000..b5dc90cfb6 --- /dev/null +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/util/StateChangeListener.java @@ -0,0 +1,26 @@ +/* + * + * 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.util; + +public interface StateChangeListener<T, E extends Enum> +{ + void stateChanged(T object, E oldState, E newState); +} diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractVirtualHost.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractVirtualHost.java index 5859ce3c68..9a23e00f90 100644 --- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractVirtualHost.java +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractVirtualHost.java @@ -47,8 +47,13 @@ import org.apache.qpid.server.exchange.ExchangeFactory; import org.apache.qpid.server.exchange.ExchangeRegistry; import org.apache.qpid.server.logging.actors.CurrentActor; import org.apache.qpid.server.logging.messages.VirtualHostMessages; +import org.apache.qpid.server.message.MessageDestination; +import org.apache.qpid.server.message.MessageNode; +import org.apache.qpid.server.message.MessageSource; import org.apache.qpid.server.model.UUIDGenerator; import org.apache.qpid.server.plugin.ExchangeType; +import org.apache.qpid.server.plugin.QpidServiceLoader; +import org.apache.qpid.server.plugin.SystemNodeCreator; import org.apache.qpid.server.protocol.AMQConnectionModel; import org.apache.qpid.server.protocol.AMQSessionModel; import org.apache.qpid.server.protocol.LinkRegistry; @@ -99,6 +104,7 @@ public abstract class AbstractVirtualHost implements VirtualHost, IConnectionReg private final DtxRegistry _dtxRegistry; private final AMQQueueFactory _queueFactory; + private final SystemNodeRegistry _systemNodeRegistry = new SystemNodeRegistry(); private volatile State _state = State.INITIALISING; @@ -107,6 +113,13 @@ public abstract class AbstractVirtualHost implements VirtualHost, IConnectionReg private final Map<String, LinkRegistry> _linkRegistry = new HashMap<String, LinkRegistry>(); private boolean _blocked; + private final Map<String, MessageDestination> _systemNodeDestinations = + Collections.synchronizedMap(new HashMap<String,MessageDestination>()); + + private final Map<String, MessageSource> _systemNodeSources = + Collections.synchronizedMap(new HashMap<String,MessageSource>()); + + public AbstractVirtualHost(VirtualHostRegistry virtualHostRegistry, StatisticsGatherer brokerStatisticsGatherer, SecurityManager parentSecurityManager, @@ -149,6 +162,8 @@ public abstract class AbstractVirtualHost implements VirtualHost, IConnectionReg _exchangeRegistry = new DefaultExchangeRegistry(this, _queueRegistry); + registerSystemNodes(); + initialiseStatistics(); initialiseStorage(hostConfig, virtualHost); @@ -157,6 +172,16 @@ public abstract class AbstractVirtualHost implements VirtualHost, IConnectionReg getMessageStore().addEventListener(this, Event.PERSISTENT_MESSAGE_SIZE_UNDERFULL); } + private void registerSystemNodes() + { + QpidServiceLoader<SystemNodeCreator> qpidServiceLoader = new QpidServiceLoader<SystemNodeCreator>(); + Iterable<SystemNodeCreator> factories = qpidServiceLoader.instancesOf(SystemNodeCreator.class); + for(SystemNodeCreator creator : factories) + { + creator.register(_systemNodeRegistry); + } + } + abstract protected void initialiseStorage(VirtualHostConfiguration hostConfig, org.apache.qpid.server.model.VirtualHost virtualHost) throws Exception; @@ -441,6 +466,13 @@ public abstract class AbstractVirtualHost implements VirtualHost, IConnectionReg } @Override + public MessageSource getMessageSource(final String name) + { + MessageSource systemSource = _systemNodeSources.get(name); + return systemSource == null ? getQueue(name) : systemSource; + } + + @Override public AMQQueue getQueue(UUID id) { return _queueRegistry.getQueue(id); @@ -524,6 +556,14 @@ public abstract class AbstractVirtualHost implements VirtualHost, IConnectionReg } + + @Override + public MessageDestination getMessageDestination(final String name) + { + MessageDestination destination = _systemNodeDestinations.get(name); + return destination == null ? getExchange(name) : destination; + } + @Override public Exchange getExchange(String name) { @@ -927,4 +967,39 @@ public abstract class AbstractVirtualHost implements VirtualHost, IConnectionReg } } } + + private class SystemNodeRegistry implements SystemNodeCreator.SystemNodeRegistry + { + @Override + public void registerSystemNode(final MessageNode node) + { + if(node instanceof MessageDestination) + { + _systemNodeDestinations.put(node.getName(), (MessageDestination) node); + } + if(node instanceof MessageSource) + { + _systemNodeSources.put(node.getName(), (MessageSource)node); + } + } + + @Override + public void removeSystemNode(final MessageNode node) + { + if(node instanceof MessageDestination) + { + _systemNodeDestinations.remove(node.getName()); + } + if(node instanceof MessageSource) + { + _systemNodeSources.remove(node.getName()); + } + } + + @Override + public VirtualHost getVirtualHost() + { + return AbstractVirtualHost.this; + } + } } diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java index 2ebbedccd4..7034311d84 100755 --- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java @@ -30,6 +30,9 @@ import org.apache.qpid.common.Closeable; import org.apache.qpid.server.configuration.VirtualHostConfiguration; import org.apache.qpid.server.connection.IConnectionRegistry; import org.apache.qpid.server.exchange.Exchange; +import org.apache.qpid.server.message.MessageDestination; +import org.apache.qpid.server.message.MessageNode; +import org.apache.qpid.server.message.MessageSource; import org.apache.qpid.server.plugin.ExchangeType; import org.apache.qpid.server.protocol.LinkRegistry; import org.apache.qpid.server.queue.AMQQueue; @@ -49,6 +52,7 @@ public interface VirtualHost extends DurableConfigurationStore.Source, Closeable String getName(); AMQQueue getQueue(String name); + MessageSource getMessageSource(String name); AMQQueue getQueue(UUID id); @@ -76,6 +80,8 @@ public interface VirtualHost extends DurableConfigurationStore.Source, Closeable void removeExchange(Exchange exchange, boolean force) throws AMQException; + MessageDestination getMessageDestination(String name); + Exchange getExchange(String name); Exchange getExchange(UUID id); diff --git a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/VirtualHostConfigRecoveryHandler.java b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/VirtualHostConfigRecoveryHandler.java index b7d3cf872b..6e36cdfa94 100755 --- a/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/VirtualHostConfigRecoveryHandler.java +++ b/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/VirtualHostConfigRecoveryHandler.java @@ -119,7 +119,7 @@ public class VirtualHostConfigRecoveryHandler implements } for(Transaction.Record record : enqueues) { - final AMQQueue queue = _virtualHost.getQueue(record.getQueue().getId()); + final AMQQueue queue = _virtualHost.getQueue(record.getResource().getId()); if(queue != null) { final long messageId = record.getMessage().getMessageNumber(); @@ -141,7 +141,7 @@ public class VirtualHostConfigRecoveryHandler implements try { - queue.enqueue(message, true, null); + queue.enqueue(message, null); ref.release(); } catch (AMQException e) @@ -173,13 +173,13 @@ public class VirtualHostConfigRecoveryHandler implements StringBuilder xidString = xidAsString(id); CurrentActor.get().message(_logSubject, TransactionLogMessages.XA_INCOMPLETE_QUEUE(xidString.toString(), - record.getQueue().getId().toString())); + record.getResource().getId().toString())); } } for(Transaction.Record record : dequeues) { - final AMQQueue queue = _virtualHost.getQueue(record.getQueue().getId()); + final AMQQueue queue = _virtualHost.getQueue(record.getResource().getId()); if(queue != null) { final long messageId = record.getMessage().getMessageNumber(); @@ -223,7 +223,7 @@ public class VirtualHostConfigRecoveryHandler implements StringBuilder xidString = xidAsString(id); CurrentActor.get().message(_logSubject, TransactionLogMessages.XA_INCOMPLETE_QUEUE(xidString.toString(), - record.getQueue().getId().toString())); + record.getResource().getId().toString())); } } @@ -292,7 +292,7 @@ public class VirtualHostConfigRecoveryHandler implements count = 0; } - queue.enqueue(message); + queue.enqueue(message,null); _queueRecoveries.put(queueName, ++count); } @@ -312,10 +312,22 @@ public class VirtualHostConfigRecoveryHandler implements new TransactionLogResource() { @Override + public String getName() + { + return "<<UNKNOWN>>"; + } + + @Override public UUID getId() { return queueId; } + + @Override + public boolean isDurable() + { + return false; + } }; txn.dequeueMessage(mockQueue, new DummyMessage(messageId)); txn.commitTranAsync(); diff --git a/qpid/java/broker-core/src/test/java/org/apache/qpid/server/subscription/MockSubscription.java b/qpid/java/broker-core/src/test/java/org/apache/qpid/server/consumer/MockConsumer.java index 8d1b27e272..a9b99503ec 100644 --- a/qpid/java/broker-core/src/test/java/org/apache/qpid/server/subscription/MockSubscription.java +++ b/qpid/java/broker-core/src/test/java/org/apache/qpid/server/consumer/MockConsumer.java @@ -19,20 +19,24 @@ * */ -package org.apache.qpid.server.subscription; +package org.apache.qpid.server.consumer; import org.apache.qpid.AMQException; import org.apache.qpid.protocol.AMQConstant; -import org.apache.qpid.server.logging.LogActor; +import org.apache.qpid.server.filter.FilterManager; +import org.apache.qpid.server.filter.Filterable; +import org.apache.qpid.server.filter.MessageFilter; +import org.apache.qpid.server.filter.SimpleFilterManager; import org.apache.qpid.server.logging.LogSubject; +import org.apache.qpid.server.message.MessageInstance; +import org.apache.qpid.server.message.ServerMessage; import org.apache.qpid.server.model.Port; import org.apache.qpid.server.model.Transport; import org.apache.qpid.server.protocol.AMQConnectionModel; import org.apache.qpid.server.protocol.AMQSessionModel; import org.apache.qpid.server.queue.AMQQueue; -import org.apache.qpid.server.queue.QueueEntry; -import org.apache.qpid.server.queue.QueueEntry.SubscriptionAcquiredState; import org.apache.qpid.server.stats.StatisticsCounter; +import org.apache.qpid.server.util.StateChangeListener; import java.util.ArrayList; import java.util.List; @@ -41,83 +45,66 @@ import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; -public class MockSubscription implements Subscription +public class MockConsumer implements ConsumerTarget { + private final List<String> _messageIds; private boolean _closed = false; private String tag = "mocktag"; private AMQQueue queue = null; - private StateListener _listener = null; - private volatile AMQQueue.Context _queueContext = null; + private StateChangeListener<ConsumerTarget, State> _listener = null; private State _state = State.ACTIVE; - private ArrayList<QueueEntry> messages = new ArrayList<QueueEntry>(); + private ArrayList<MessageInstance> messages = new ArrayList<MessageInstance>(); private final Lock _stateChangeLock = new ReentrantLock(); - private List<QueueEntry> _acceptEntries = null; - private final QueueEntry.SubscriptionAcquiredState _owningState = new QueueEntry.SubscriptionAcquiredState(this); - - private static final AtomicLong idGenerator = new AtomicLong(0); - // Create a simple ID that increments for ever new Subscription - private final long _subscriptionID = idGenerator.getAndIncrement(); private boolean _isActive = true; - public MockSubscription() + public MockConsumer() { + _messageIds = null; } - public MockSubscription(List<QueueEntry> acceptEntries) + public MockConsumer(List<String> messageIds) { - _acceptEntries = acceptEntries; + _messageIds = messageIds; } - public void close() + public boolean close() { _closed = true; if (_listener != null) { - _listener.stateChange(this, _state, State.CLOSED); + _listener.stateChanged(this, _state, State.CLOSED); } _state = State.CLOSED; + return true; } - public String getConsumerName() + public String getName() { return tag; } - public long getSubscriptionID() - { - return _subscriptionID; - } - - public AMQQueue.Context getQueueContext() - { - return _queueContext; - } - - public SubscriptionAcquiredState getOwningState() - { - return _owningState; - } - - public LogActor getLogActor() - { - return null; //To change body of implemented methods use File | Settings | File Templates. - } - - public boolean isTransient() - { - return false; - } - - public long getBytesOut() - { - return 0; // TODO - Implement - } - - public long getMessagesOut() + public FilterManager getFilters() { - return 0; // TODO - Implement + if(_messageIds != null) + { + SimpleFilterManager filters = new SimpleFilterManager(); + filters.add(new MessageFilter() + { + @Override + public boolean matches(final Filterable message) + { + final String messageId = message.getMessageHeader().getMessageId(); + return _messageIds.contains(messageId); + } + }); + return filters; + } + else + { + return null; + } } public long getUnacknowledgedBytes() @@ -140,90 +127,33 @@ public class MockSubscription implements Subscription return new MockSessionModel(); } - public boolean trySendLock() - { - return _stateChangeLock.tryLock(); - } - - - public void getSendLock() - { - _stateChangeLock.lock(); - } - - public boolean hasInterest(QueueEntry entry) - { - if(_acceptEntries != null) - { - //simulate selector behaviour, only signal - //interest in the dictated queue entries - return _acceptEntries.contains(entry); - } - - return true; - } - public boolean isActive() { return _isActive ; } - public void set(String key, Object value) - { - } - - public Object get(String key) - { - return null; - } - public boolean isAutoClose() - { - return false; - } public boolean isClosed() { return _closed; } - public boolean acquires() - { - return true; - } - - public boolean seesRequeues() - { - return true; - } public boolean isSuspended() { return false; } - public void queueDeleted(AMQQueue queue) - { - } - - public void releaseSendLock() + public void queueDeleted() { - _stateChangeLock.unlock(); } - public void onDequeue(QueueEntry queueEntry) + public void restoreCredit(ServerMessage message) { } - public void restoreCredit(QueueEntry queueEntry) - { - } - - public void releaseQueueEntry(QueueEntry queueEntry) - { - } - - public void send(QueueEntry entry, boolean batch) throws AMQException + public void send(MessageInstance entry, boolean batch) throws AMQException { if (messages.contains(entry)) { @@ -237,47 +167,52 @@ public class MockSubscription implements Subscription } - public void setQueueContext(AMQQueue.Context queueContext) + public State getState() { - _queueContext = queueContext; + return _state; } - public void setQueue(AMQQueue queue, boolean exclusive) + @Override + public void consumerAdded(final Consumer sub) { - this.queue = queue; } - public void setNoLocal(boolean noLocal) + @Override + public void consumerRemoved(final Consumer sub) { - } - public void setStateListener(StateListener listener) - { - this._listener = listener; } - public State getState() + public void setState(State state) { - return _state; + State oldState = _state; + _state = state; + if(_listener != null) + { + _listener.stateChanged(this, oldState, state); + } } - public boolean wouldSuspend(QueueEntry msg) + @Override + public void setStateListener(final StateChangeListener<ConsumerTarget, State> listener) { - return false; + _listener = listener; } - public ArrayList<QueueEntry> getMessages() + public ArrayList<MessageInstance> getMessages() { return messages; } - public boolean isSessionTransactional() + + public void queueEmpty() throws AMQException { - return false; } - public void queueEmpty() throws AMQException + @Override + public boolean allocateCredit(final ServerMessage msg) { + return true; } public void setActive(final boolean isActive) diff --git a/qpid/java/broker-core/src/test/java/org/apache/qpid/server/exchange/TopicExchangeTest.java b/qpid/java/broker-core/src/test/java/org/apache/qpid/server/exchange/TopicExchangeTest.java index 764549626a..ea9d0ac693 100644 --- a/qpid/java/broker-core/src/test/java/org/apache/qpid/server/exchange/TopicExchangeTest.java +++ b/qpid/java/broker-core/src/test/java/org/apache/qpid/server/exchange/TopicExchangeTest.java @@ -25,6 +25,7 @@ import junit.framework.Assert; import org.apache.qpid.AMQException; import org.apache.qpid.server.binding.Binding; +import org.apache.qpid.server.consumer.Consumer; import org.apache.qpid.server.message.InstanceProperties; import org.apache.qpid.server.message.MessageReference; import org.apache.qpid.server.message.ServerMessage; @@ -85,7 +86,7 @@ public class TopicExchangeTest extends QpidTestCase public void testDirectMatch() throws AMQException { - AMQQueue queue = _vhost.createQueue(UUIDGenerator.generateRandomUUID(), "ab", false, null, false, false, + AMQQueue<Consumer> queue = _vhost.createQueue(UUIDGenerator.generateRandomUUID(), "ab", false, null, false, false, false, null); _exchange.registerQueue(new Binding(null, "a.b",queue, _exchange, null)); @@ -108,7 +109,7 @@ public class TopicExchangeTest extends QpidTestCase public void testStarMatch() throws AMQException { - AMQQueue queue = _vhost.createQueue(UUIDGenerator.generateRandomUUID(), "a*", false, null, false, false, false, null); + AMQQueue<Consumer> queue = _vhost.createQueue(UUIDGenerator.generateRandomUUID(), "a*", false, null, false, false, false, null); _exchange.registerQueue(new Binding(null, "a.*",queue, _exchange, null)); @@ -139,7 +140,7 @@ public class TopicExchangeTest extends QpidTestCase public void testHashMatch() throws AMQException { - AMQQueue queue = _vhost.createQueue(UUIDGenerator.generateRandomUUID(), "a#", false, null, false, false, false, null); + AMQQueue<Consumer> queue = _vhost.createQueue(UUIDGenerator.generateRandomUUID(), "a#", false, null, false, false, false, null); _exchange.registerQueue(new Binding(null, "a.#",queue, _exchange, null)); @@ -190,7 +191,7 @@ public class TopicExchangeTest extends QpidTestCase public void testMidHash() throws AMQException { - AMQQueue queue = _vhost.createQueue(UUIDGenerator.generateRandomUUID(), "a", false, null, false, false, + AMQQueue<Consumer> queue = _vhost.createQueue(UUIDGenerator.generateRandomUUID(), "a", false, null, false, false, false, null); _exchange.registerQueue(new Binding(null, "a.*.#.b",queue, _exchange, null)); @@ -216,7 +217,7 @@ public class TopicExchangeTest extends QpidTestCase public void testMatchAfterHash() throws AMQException { - AMQQueue queue = _vhost.createQueue(UUIDGenerator.generateRandomUUID(), "a#", false, null, false, false, + AMQQueue<Consumer> queue = _vhost.createQueue(UUIDGenerator.generateRandomUUID(), "a#", false, null, false, false, false, null); _exchange.registerQueue(new Binding(null, "a.*.#.b.c",queue, _exchange, null)); @@ -255,7 +256,7 @@ public class TopicExchangeTest extends QpidTestCase public void testHashAfterHash() throws AMQException { - AMQQueue queue = _vhost.createQueue(UUIDGenerator.generateRandomUUID(), "a#", false, null, false, false, + AMQQueue<Consumer> queue = _vhost.createQueue(UUIDGenerator.generateRandomUUID(), "a#", false, null, false, false, false, null); _exchange.registerQueue(new Binding(null, "a.*.#.b.c.#.d",queue, _exchange, null)); @@ -277,7 +278,7 @@ public class TopicExchangeTest extends QpidTestCase public void testHashHash() throws AMQException { - AMQQueue queue = _vhost.createQueue(UUIDGenerator.generateRandomUUID(), "a#", false, null, false, false, + AMQQueue<Consumer> queue = _vhost.createQueue(UUIDGenerator.generateRandomUUID(), "a#", false, null, false, false, false, null); _exchange.registerQueue(new Binding(null, "a.#.*.#.d",queue, _exchange, null)); @@ -321,7 +322,7 @@ public class TopicExchangeTest extends QpidTestCase when(message.getMessageNumber()).thenReturn(messageNumber); for(BaseQueue q : queues) { - q.enqueue(message); + q.enqueue(message, null); } return queues.size(); diff --git a/qpid/java/broker-core/src/test/java/org/apache/qpid/server/logging/actors/SubscriptionActorTest.java b/qpid/java/broker-core/src/test/java/org/apache/qpid/server/logging/actors/SubscriptionActorTest.java deleted file mode 100644 index a0a2a7b648..0000000000 --- a/qpid/java/broker-core/src/test/java/org/apache/qpid/server/logging/actors/SubscriptionActorTest.java +++ /dev/null @@ -1,86 +0,0 @@ -/* - * - * 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.logging.actors; - -import org.apache.qpid.server.subscription.MockSubscription; -import org.apache.qpid.server.util.BrokerTestHelper; - -import java.util.List; - -/** - * Test : AMQPConnectionActorTest - * Validate the AMQPConnectionActor class. - * - * The test creates a new AMQPActor and then logs a message using it. - * - * The test then verifies that the logged message was the only one created and - * that the message contains the required message. - */ -public class SubscriptionActorTest extends BaseConnectionActorTestCase -{ - - @Override - public void setUp() throws Exception - { - super.setUp(); - - MockSubscription mockSubscription = new MockSubscription(); - - mockSubscription.setQueue(BrokerTestHelper.createQueue(getName(), getVirtualHost()), false); - - setAmqpActor(new SubscriptionActor(getRootLogger(), mockSubscription)); - } - - /** - * Test the AMQPActor logging as a Subscription logger. - * - * The test sends a message then verifies that it entered the logs. - * - * The log message should be fully replaced (no '{n}' values) and should - * contain subscription identification. - */ - public void testSubscription() - { - final String message = sendTestLogMessage(getAmqpActor()); - - List<Object> logs = getRawLogger().getLogMessages(); - - assertEquals("Message log size not as expected.", 1, logs.size()); - - // Verify that the logged message is present in the output - assertTrue("Message was not found in log message", - logs.get(0).toString().contains(message)); - - // Verify that all the values were presented to the MessageFormatter - // so we will not end up with '{n}' entries in the log. - assertFalse("Verify that the string does not contain any '{'.", - logs.get(0).toString().contains("{")); - - // Verify that the message has the correct type - assertTrue("Message contains the [sub: prefix", - logs.get(0).toString().contains("[sub:")); - - // Verify that the logged message does not contains the 'ch:' marker - assertFalse("Message was logged with a channel identifier." + logs.get(0), - logs.get(0).toString().contains("/ch:")); - } - -} diff --git a/qpid/java/broker-core/src/test/java/org/apache/qpid/server/logging/messages/SubscriptionMessagesTest.java b/qpid/java/broker-core/src/test/java/org/apache/qpid/server/logging/messages/ConsumerMessagesTest.java index b2bc351f8f..52a53e8d38 100644 --- a/qpid/java/broker-core/src/test/java/org/apache/qpid/server/logging/messages/SubscriptionMessagesTest.java +++ b/qpid/java/broker-core/src/test/java/org/apache/qpid/server/logging/messages/ConsumerMessagesTest.java @@ -25,7 +25,7 @@ import java.util.List; /** * Test SUB Log Messages */ -public class SubscriptionMessagesTest extends AbstractTestMessages +public class ConsumerMessagesTest extends AbstractTestMessages { public void testSubscriptionCreateALL() { diff --git a/qpid/java/broker-core/src/test/java/org/apache/qpid/server/logging/messages/VirtualHostMessagesTest.java b/qpid/java/broker-core/src/test/java/org/apache/qpid/server/logging/messages/VirtualHostMessagesTest.java index 17d68ef7c3..c1068b4a0b 100644 --- a/qpid/java/broker-core/src/test/java/org/apache/qpid/server/logging/messages/VirtualHostMessagesTest.java +++ b/qpid/java/broker-core/src/test/java/org/apache/qpid/server/logging/messages/VirtualHostMessagesTest.java @@ -38,7 +38,7 @@ public class VirtualHostMessagesTest extends AbstractTestMessages validateLogMessage(log, "VHT-1001", expected); } - public void testSubscriptionClosed() + public void testVirtualhostClosed() { _logMessage = VirtualHostMessages.CLOSED(); List<Object> log = performLog(); diff --git a/qpid/java/broker-core/src/test/java/org/apache/qpid/server/logging/subjects/SubscriptionLogSubjectTest.java b/qpid/java/broker-core/src/test/java/org/apache/qpid/server/logging/subjects/SubscriptionLogSubjectTest.java deleted file mode 100644 index 3afba28cd6..0000000000 --- a/qpid/java/broker-core/src/test/java/org/apache/qpid/server/logging/subjects/SubscriptionLogSubjectTest.java +++ /dev/null @@ -1,105 +0,0 @@ -/* - * - * 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.logging.subjects; - -import org.apache.qpid.server.queue.AMQQueue; -import org.apache.qpid.server.queue.MockAMQQueue; -import org.apache.qpid.server.subscription.Subscription; -import org.apache.qpid.server.util.BrokerTestHelper; -import org.apache.qpid.server.virtualhost.VirtualHost; - -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.when; - -/** - * Validate SubscriptionLogSubjects are logged as expected - */ -public class SubscriptionLogSubjectTest extends AbstractTestLogSubject -{ - - private static final long SUBSCRIPTION_ID = 1; - private AMQQueue _queue; - private VirtualHost _testVhost; - private Subscription _subscription; - - @Override - public void setUp() throws Exception - { - super.setUp(); - - _testVhost = BrokerTestHelper.createVirtualHost("test"); - - _queue = new MockAMQQueue("SubscriptionLogSubjectTest"); - ((MockAMQQueue) _queue).setVirtualHost(_testVhost); - - _subscription = mock(Subscription.class); - when(_subscription.getQueue()).thenReturn(_queue); - when(_subscription.getSubscriptionID()).thenReturn(SUBSCRIPTION_ID); - - _subject = new SubscriptionLogSubject(_subscription); - } - - @Override - public void tearDown() throws Exception - { - if (_testVhost != null) - { - _testVhost.close(); - } - super.tearDown(); - } - - /** - * Validate that the logged Subject message is as expected: - * MESSAGE [Blank][sub:0(vh(/test)/qu(SubscriptionLogSubjectTest))] <Log Message> - * - * @param message the message whose format needs validation - */ - @Override - protected void validateLogStatement(String message) - { - String subscriptionSlice = getSlice("sub:" - + _subscription.getSubscriptionID(), - message); - - assertNotNull("Unable to locate subscription 'sub:" + - _subscription.getSubscriptionID() + "'"); - - - - // Pull out the qu(..) section from the subscription message - // Split it into three parts - // MESSAGE [Blank][sub:0(vh(/ - // test)/ - // qu(SubscriptionLogSubjectTest))] - // Take the last bit and drop off the extra )] - String[] parts = message.split("/"); - assertEquals("Message part count wrong", 3, parts.length); - String subscription = parts[2].substring(0, parts[2].indexOf(")") + 1); - - // Adding the ')' is a bit ugly but SubscriptionLogSubject is the only - // Subject that nests () and so the simple parser of checking for the - // next ')' falls down. - verifyVirtualHost(subscriptionSlice+ ")", _queue.getVirtualHost()); - - verifyQueue(subscription, _queue); - } -} diff --git a/qpid/java/broker-core/src/test/java/org/apache/qpid/server/queue/AMQPriorityQueueTest.java b/qpid/java/broker-core/src/test/java/org/apache/qpid/server/queue/AMQPriorityQueueTest.java index a468fa072b..ced00dc578 100644 --- a/qpid/java/broker-core/src/test/java/org/apache/qpid/server/queue/AMQPriorityQueueTest.java +++ b/qpid/java/broker-core/src/test/java/org/apache/qpid/server/queue/AMQPriorityQueueTest.java @@ -25,10 +25,14 @@ import junit.framework.AssertionFailedError; import org.apache.qpid.AMQException; import org.apache.qpid.server.message.AMQMessageHeader; +import org.apache.qpid.server.message.MessageInstance; import org.apache.qpid.server.message.ServerMessage; import java.util.ArrayList; +import java.util.EnumSet; + import org.apache.qpid.server.model.Queue; +import org.apache.qpid.server.consumer.Consumer; import static org.mockito.Mockito.when; @@ -47,25 +51,25 @@ public class AMQPriorityQueueTest extends SimpleAMQQueueTest // Enqueue messages in order SimpleAMQQueue queue = getQueue(); - queue.enqueue(createMessage(1L, (byte) 10)); - queue.enqueue(createMessage(2L, (byte) 4)); - queue.enqueue(createMessage(3L, (byte) 0)); + queue.enqueue(createMessage(1L, (byte) 10), null); + queue.enqueue(createMessage(2L, (byte) 4), null); + queue.enqueue(createMessage(3L, (byte) 0), null); // Enqueue messages in reverse order - queue.enqueue(createMessage(4L, (byte) 0)); - queue.enqueue(createMessage(5L, (byte) 4)); - queue.enqueue(createMessage(6L, (byte) 10)); + queue.enqueue(createMessage(4L, (byte) 0), null); + queue.enqueue(createMessage(5L, (byte) 4), null); + queue.enqueue(createMessage(6L, (byte) 10), null); // Enqueue messages out of order - queue.enqueue(createMessage(7L, (byte) 4)); - queue.enqueue(createMessage(8L, (byte) 10)); - queue.enqueue(createMessage(9L, (byte) 0)); + queue.enqueue(createMessage(7L, (byte) 4), null); + queue.enqueue(createMessage(8L, (byte) 10), null); + queue.enqueue(createMessage(9L, (byte) 0), null); // Register subscriber - queue.registerSubscription(getSubscription(), false); + queue.addConsumer(getConsumer(), null, null, "test", EnumSet.noneOf(Consumer.Option.class)); Thread.sleep(150); - ArrayList<QueueEntry> msgs = getSubscription().getMessages(); + ArrayList<MessageInstance> msgs = getConsumer().getMessages(); try { assertEquals(1L, msgs.get(0).getMessage().getMessageNumber()); @@ -84,7 +88,7 @@ public class AMQPriorityQueueTest extends SimpleAMQQueueTest { // Show message order on failure. int index = 1; - for (QueueEntry qe : msgs) + for (MessageInstance qe : msgs) { System.err.println(index + ":" + qe.getMessage().getMessageNumber()); index++; diff --git a/qpid/java/broker-core/src/test/java/org/apache/qpid/server/queue/ConsumerListTest.java b/qpid/java/broker-core/src/test/java/org/apache/qpid/server/queue/ConsumerListTest.java new file mode 100644 index 0000000000..35508bb2c4 --- /dev/null +++ b/qpid/java/broker-core/src/test/java/org/apache/qpid/server/queue/ConsumerListTest.java @@ -0,0 +1,445 @@ +/* + * + * 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.protocol.AMQSessionModel; +import org.apache.qpid.server.consumer.Consumer; +import org.apache.qpid.server.consumer.ConsumerTarget; +import org.apache.qpid.test.utils.QpidTestCase; + +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +public class ConsumerListTest extends QpidTestCase +{ + private QueueConsumerList _subList; + private QueueConsumer _sub1; + private QueueConsumer _sub2; + private QueueConsumer _sub3; + private QueueConsumerList.ConsumerNode _node; + + protected void setUp() + { + _subList = new QueueConsumerList(); + + _sub1 = newMockConsumer(); + _sub2 = newMockConsumer(); + _sub3 = newMockConsumer(); + + _subList.add(_sub1); + _subList.add(_sub2); + _subList.add(_sub3); + + _node = _subList.getHead(); + } + + + private QueueConsumer newMockConsumer() + { + ConsumerTarget target = mock(ConsumerTarget.class); + when(target.getSessionModel()).thenReturn(mock(AMQSessionModel.class)); + return new QueueConsumer(null,null,true,true,"sub",false,target); + } + + /** + * Test that if the first (non-head) node in the list is deleted (but is still present), + * it is not returned when searching through the list for the next viable node, and the + * subsequent viable node is returned instead. + */ + public void testFindNextSkipsFirstDeletedNode() + { + assertTrue("Deleting consumer node should have succeeded", + getNodeForConsumer(_subList, _sub1).delete()); + + assertNotNull("Returned node should not be null", _node = _node.findNext()); + assertEquals("Should have returned node for 2nd consumer", _sub2, _node.getConsumer()); + + assertNotNull("Returned node should not be null", _node = _node.findNext()); + assertEquals("Should have returned node for 3rd consumer", _sub3, _node.getConsumer()); + } + + /** + * Test that if a central node in the list is deleted (but is still present), + * it is not returned when searching through the list for the next viable node, + * and the subsequent viable node is returned instead. + */ + public void testFindNextSkipsCentralDeletedNode() + { + assertNotNull("Returned node should not be null", _node = _node.findNext()); + + assertTrue("Deleting consumer node should have succeeded", + getNodeForConsumer(_subList, _sub2).delete()); + + assertNotNull("Returned node should not be null", _node = _node.findNext()); + assertEquals("Should have returned node for 3rd consumer", _sub3, _node.getConsumer()); + } + + /** + * Test that if the last node in the list is deleted (but is still present), + * it is not returned when searching through the list for the next viable node, + * and null is returned instead. + */ + public void testFindNextSkipsLastDeletedNode() + { + assertNotNull("Returned node should not be null", _node = _node.findNext()); + assertEquals("Should have returned node for 1st consumer", _sub1, _node.getConsumer()); + + assertNotNull("Returned node should not be null", _node = _node.findNext()); + assertEquals("Should have returned node for 2nd consumer", _sub2, _node.getConsumer()); + + assertTrue("Deleting consumer node should have succeeded", + getNodeForConsumer(_subList, _sub3).delete()); + + assertNull("Returned node should be null", _node = _node.findNext()); + } + + /** + * Test that if multiple nodes in the list are deleted (but still present), they + * are not returned when searching through the list for the next viable node, + * and the subsequent viable node is returned instead. + */ + public void testFindNextSkipsMultipleDeletedNode() + { + assertTrue("Deleting consumer node should have succeeded", + getNodeForConsumer(_subList, _sub1).delete()); + assertTrue("Deleting consumer node should have succeeded", + getNodeForConsumer(_subList, _sub2).delete()); + + assertNotNull("Returned node should not be null", _node = _node.findNext()); + assertEquals("Should have returned node for 3rd consumer", _sub3, _node.getConsumer()); + } + + /** + * Test that if a node in the list is marked 'deleted' it is still present in the list + * until actually removed. counter-test to verify above testing of getNext() method. + */ + public void testDeletedNodeStillPresent() + { + assertTrue("Deleting consumer node should have succeeded", + getNodeForConsumer(_subList, _sub1).delete()); + + assertNotNull("Node marked deleted should still be present", getNodeForConsumer(_subList, _sub1)); + assertEquals("All 3 nodes are still expected to be present", 3, countNodes(_subList)); + } + + /** + * Traverses the list nodes in a non-mutating fashion, returning the first node which matches the given + * Consumer, or null if none is found. + */ + private QueueConsumerList.ConsumerNode getNodeForConsumer(final QueueConsumerList list, final Consumer sub) + { + QueueConsumerList.ConsumerNode node = list.getHead(); + while (node != null && node.getConsumer() != sub) + { + node = node.nextNode(); + } + + return node; + } + + /** + * Counts the number of (non-head) nodes in the list. + */ + private int countNodes(final QueueConsumerList list) + { + QueueConsumerList.ConsumerNode node = list.getHead(); + int count; + for(count = -1; node != null; count++) + { + node = node.nextNode(); + } + + return count; + } + + /** + * Tests that the head is returned as expected, and isn't the node for the first consumer. + */ + public void testGetHead() + { + assertNotNull("List head should be non null", _node); + assertNotSame("Head should not be node for first consumer", + _node, getNodeForConsumer(_subList, _sub1)); + } + + /** + * Tests that the size is returned correctly in the face of additions and removals. + */ + public void testGetSize() + { + QueueConsumerList subList = new QueueConsumerList(); + + assertEquals("Unexpected size result", 0, subList.size()); + + QueueConsumer sub1 = newMockConsumer(); + QueueConsumer sub2 = newMockConsumer(); + QueueConsumer sub3 = newMockConsumer(); + + subList.add(sub1); + assertEquals("Unexpected size result", 1, subList.size()); + + subList.add(sub2); + assertEquals("Unexpected size result", 2, subList.size()); + + subList.add(sub3); + assertEquals("Unexpected size result", 3, subList.size()); + + assertTrue("Removing consumer from list should have succeeded", subList.remove(sub1)); + assertEquals("Unexpected size result", 2, subList.size()); + + assertTrue("Removing consumer from list should have succeeded", subList.remove(sub2)); + assertEquals("Unexpected size result", 1, subList.size()); + + assertTrue("Removing consumer from list should have succeeded", subList.remove(sub3)); + assertEquals("Unexpected size result", 0, subList.size()); + } + + /** + * Test that if the first (non-head) node in the list is removed it is no longer + * present in the node structure of the list at all. + */ + public void testRemoveFirstNode() + { + assertNotNull("Should have been a node present for the consumer", getNodeForConsumer(_subList, _sub1)); + assertTrue("Removing consumer node should have succeeded", _subList.remove(_sub1)); + assertNull("Should not have been a node present for the removed consumer", + getNodeForConsumer(_subList, _sub1)); + assertEquals("Unexpected number of nodes", 2, countNodes(_subList)); + assertNotNull("Should have been a node present for the consumer", getNodeForConsumer(_subList, _sub2)); + assertNotNull("Should have been a node present for the consumer", getNodeForConsumer(_subList, _sub3)); + } + + /** + * Test that if a central node in the list is removed it is no longer + * present in the node structure of the list at all. + */ + public void testRemoveCentralNode() + { + assertNotNull("Should have been a node present for the consumer", getNodeForConsumer(_subList, _sub2)); + assertTrue("Removing consumer node should have succeeded", _subList.remove(_sub2)); + assertNull("Should not have been a node present for the removed consumer", + getNodeForConsumer(_subList, _sub2)); + assertEquals("Unexpected number of nodes", 2, countNodes(_subList)); + assertNotNull("Should have been a node present for the consumer", getNodeForConsumer(_subList, _sub1)); + assertNotNull("Should have been a node present for the consumer", getNodeForConsumer(_subList, _sub3)); + } + + /** + * Test that if the consumer contained in the last node of the list is removed + * it is no longer present in the node structure of the list at all. However, + * as the last node in the structure can't actually be removed a dummy will instead + * be present. + */ + public void testRemoveLastNode() + { + assertNotNull("Should have been a node present for the consumer", getNodeForConsumer(_subList, _sub3)); + assertTrue("Removing consumer node should have succeeded", _subList.remove(_sub3)); + assertNull("Should not have been a node present for the removed consumer", + getNodeForConsumer(_subList, _sub3)); + + //We actually expect 3 nodes to remain this time, because the last node cant be removed for thread safety reasons, + //however a dummy final node can be used as substitute to allow removal of the consumer node. + assertEquals("Unexpected number of nodes", 2 + 1, countNodes(_subList)); + assertNotNull("Should have been a node present for the consumer", getNodeForConsumer(_subList, _sub1)); + assertNotNull("Should have been a node present for the consumer", getNodeForConsumer(_subList, _sub2)); + } + + /** + * Test that if the consumer not contained in the list is requested to be removed + * that the removal fails + */ + public void testRemoveNonexistentNode() + { + QueueConsumer sub4 = newMockConsumer(); + assertNull("Should not have been a node present for the consumer", getNodeForConsumer(_subList, sub4)); + assertFalse("Removing consumer node should not have succeeded", _subList.remove(sub4)); + assertEquals("Unexpected number of nodes", 3, countNodes(_subList)); + } + + /** + * Test that if a consumer node which occurs later in the main list than the marked node is + * removed from the list after the marked node is also removed, then the marker node doesn't + * serve to retain the subsequent nodes in the list structure (and thus memory) despite their + * removal. + */ + public void testDeletedMarkedNodeDoesntLeakSubsequentlyDeletedNodes() + { + //get the nodes out the list for the 1st and 3rd consumers + QueueConsumerList.ConsumerNode sub1Node = getNodeForConsumer(_subList, _sub1); + assertNotNull("Should have been a node present for the consumer", sub1Node); + QueueConsumerList.ConsumerNode sub3Node = getNodeForConsumer(_subList, _sub3); + assertNotNull("Should have been a node present for the consumer", sub3Node); + + //mark the first consumer node + assertTrue("should have succeeded in updating the marked node", + _subList.updateMarkedNode(_subList.getMarkedNode(), sub1Node)); + + //remove the 1st consumer from the list + assertTrue("Removing consumer node should have succeeded", _subList.remove(_sub1)); + //verify the 1st consumer is no longer the marker node (replaced by a dummy), or in the main list structure + assertNotSame("Unexpected marker node", sub1Node, _subList.getMarkedNode()); + assertNull("Should not have been a node present in the list structure for the marked-but-removed sub1 node", + getNodeForConsumer(_subList, _sub1)); + + //remove the 2nd consumer from the list + assertTrue("Removing consumer node should have succeeded", _subList.remove(_sub2)); + + //verify the marker node isn't leaking subsequently removed nodes, by ensuring the very next node + //in its list structure is now the 3rd consumer (since the 2nd was removed too) + assertEquals("Unexpected next node", sub3Node, _subList.getMarkedNode().nextNode()); + + //remove the 3rd and final/tail consumer + assertTrue("Removing consumer node should have succeeded", _subList.remove(_sub3)); + + //verify the marker node isn't leaking subsequently removed nodes, by ensuring the very next node + //in its list structure is now the dummy tail (since the 3rd consumer was removed, and a dummy + //tail was inserted) and NOT the 3rd sub node. + assertNotSame("Unexpected next node", sub3Node, _subList.getMarkedNode().nextNode()); + assertTrue("Unexpected next node", _subList.getMarkedNode().nextNode().isDeleted()); + assertNull("Next non-deleted node from the marker should now be the list end, i.e. null", _subList.getMarkedNode().findNext()); + } + + /** + * Test that the marked node 'findNext' behaviour is as expected after a consumer is added + * to the list following the tail consumer node being removed while it is the marked node. + * That is, that the new consumers node is returned by getMarkedNode().findNext(). + */ + public void testMarkedNodeFindsNewConsumerAfterRemovingTailWhilstMarked() + { + //get the node out the list for the 3rd consumer + QueueConsumerList.ConsumerNode sub3Node = getNodeForConsumer(_subList, _sub3); + assertNotNull("Should have been a node present for the consumer", sub3Node); + + //mark the 3rd consumer node + assertTrue("should have succeeded in updating the marked node", + _subList.updateMarkedNode(_subList.getMarkedNode(), sub3Node)); + + //verify calling findNext on the marked node returns null, i.e. the end of the list has been reached + assertEquals("Unexpected node after marked node", null, _subList.getMarkedNode().findNext()); + + //remove the 3rd(marked) consumer from the list + assertTrue("Removing consumer node should have succeeded", _subList.remove(_sub3)); + + //add a new 4th consumer to the list + QueueConsumer sub4 = newMockConsumer(); + _subList.add(sub4); + + //get the node out the list for the 4th consumer + QueueConsumerList.ConsumerNode sub4Node = getNodeForConsumer(_subList, sub4); + assertNotNull("Should have been a node present for the consumer", sub4Node); + + //verify the marked node (which is now a dummy substitute for the 3rd consumer) returns + //the 4th consumers node as the next non-deleted node. + assertEquals("Unexpected next node", sub4Node, _subList.getMarkedNode().findNext()); + } + + /** + * Test that setting the marked node to null doesn't cause problems during remove operations + */ + public void testRemoveWithNullMarkedNode() + { + //set the marker to null + assertTrue("should have succeeded in updating the marked node", + _subList.updateMarkedNode(_subList.getMarkedNode(), null)); + + //remove the 1st consumer from the main list + assertTrue("Removing consumer node should have succeeded", _subList.remove(_sub1)); + + //verify the 1st consumer is no longer in the main list structure + assertNull("Should not have been a node present in the main list structure for sub1", + getNodeForConsumer(_subList, _sub1)); + assertEquals("Unexpected number of nodes", 2, countNodes(_subList)); + } + + /** + * Tests that after the first (non-head) node of the list is marked deleted but has not + * yet been removed, the iterator still skips it. + */ + public void testIteratorSkipsFirstDeletedNode() + { + //'delete' but don't remove the node for the 1st consumer + assertTrue("Deleting consumer node should have succeeded", + getNodeForConsumer(_subList, _sub1).delete()); + assertNotNull("Should still have been a node present for the deleted consumer", + getNodeForConsumer(_subList, _sub1)); + + QueueConsumerList.ConsumerNodeIterator iter = _subList.iterator(); + + //verify the iterator returns the 2nd consumers node + assertTrue("Iterator should have been able to advance", iter.advance()); + assertEquals("Iterator returned unexpected ConsumerNode", _sub2, iter.getNode().getConsumer()); + + //verify the iterator returns the 3rd consumers node and not the 2nd. + assertTrue("Iterator should have been able to advance", iter.advance()); + assertEquals("Iterator returned unexpected ConsumerNode", _sub3, iter.getNode().getConsumer()); + } + + /** + * Tests that after a central node of the list is marked deleted but has not yet been removed, + * the iterator still skips it. + */ + public void testIteratorSkipsCentralDeletedNode() + { + //'delete' but don't remove the node for the 2nd consumer + assertTrue("Deleting consumer node should have succeeded", + getNodeForConsumer(_subList, _sub2).delete()); + assertNotNull("Should still have been a node present for the deleted consumer", + getNodeForConsumer(_subList, _sub2)); + + QueueConsumerList.ConsumerNodeIterator iter = _subList.iterator(); + + //verify the iterator returns the 1st consumers node + assertTrue("Iterator should have been able to advance", iter.advance()); + assertEquals("Iterator returned unexpected ConsumerNode", _sub1, iter.getNode().getConsumer()); + + //verify the iterator returns the 3rd consumers node and not the 2nd. + assertTrue("Iterator should have been able to advance", iter.advance()); + assertEquals("Iterator returned unexpected ConsumerNode", _sub3, iter.getNode().getConsumer()); + } + + /** + * Tests that after the last node of the list is marked deleted but has not yet been removed, + * the iterator still skips it. + */ + public void testIteratorSkipsDeletedFinalNode() + { + //'delete' but don't remove the node for the 3rd consumer + assertTrue("Deleting consumer node should have succeeded", + getNodeForConsumer(_subList, _sub3).delete()); + assertNotNull("Should still have been a node present for the deleted 3rd consumer", + getNodeForConsumer(_subList, _sub3)); + + QueueConsumerList.ConsumerNodeIterator iter = _subList.iterator(); + + //verify the iterator returns the 1st consumers node + assertTrue("Iterator should have been able to advance", iter.advance()); + assertEquals("Iterator returned unexpected ConsumerNode", _sub1, iter.getNode().getConsumer()); + + //verify the iterator returns the 2nd consumers node + assertTrue("Iterator should have been able to advance", iter.advance()); + assertEquals("Iterator returned unexpected ConsumerNode", _sub2, iter.getNode().getConsumer()); + + //verify the iterator can no longer advance and does not return a consumer node + assertFalse("Iterator should not have been able to advance", iter.advance()); + assertEquals("Iterator returned unexpected ConsumerNode", null, iter.getNode()); + } +} diff --git a/qpid/java/broker-core/src/test/java/org/apache/qpid/server/queue/MockAMQQueue.java b/qpid/java/broker-core/src/test/java/org/apache/qpid/server/queue/MockAMQQueue.java index ea2e29d40d..66c12717db 100644 --- a/qpid/java/broker-core/src/test/java/org/apache/qpid/server/queue/MockAMQQueue.java +++ b/qpid/java/broker-core/src/test/java/org/apache/qpid/server/queue/MockAMQQueue.java @@ -24,21 +24,28 @@ import org.apache.qpid.AMQException; import org.apache.qpid.server.binding.Binding; import org.apache.qpid.server.configuration.QueueConfiguration; import org.apache.qpid.server.exchange.Exchange; +import org.apache.qpid.server.filter.FilterManager; import org.apache.qpid.server.logging.LogSubject; +import org.apache.qpid.server.message.InstanceProperties; +import org.apache.qpid.server.message.MessageInstance; import org.apache.qpid.server.message.ServerMessage; import org.apache.qpid.server.protocol.AMQSessionModel; import org.apache.qpid.server.security.AuthorizationHolder; -import org.apache.qpid.server.subscription.Subscription; +import org.apache.qpid.server.consumer.Consumer; +import org.apache.qpid.server.consumer.ConsumerTarget; +import org.apache.qpid.server.txn.ServerTransaction; +import org.apache.qpid.server.util.Action; import org.apache.qpid.server.virtualhost.VirtualHost; import java.util.Collection; import java.util.Collections; +import java.util.EnumSet; import java.util.List; import java.util.Set; import java.util.UUID; import java.util.concurrent.CopyOnWriteArrayList; -public class MockAMQQueue implements AMQQueue +public class MockAMQQueue implements AMQQueue<QueueConsumer> { private boolean _deleted = false; private String _name; @@ -202,32 +209,44 @@ public class MockAMQQueue implements AMQQueue return _virtualhost; } - public String getName() + @Override + public QueueConsumer addConsumer(final ConsumerTarget target, + final FilterManager filters, + final Class<? extends ServerMessage> messageClass, + final String consumerName, + final EnumSet<Consumer.Option> options) throws AMQException { - return _name; + return new QueueConsumer(filters, messageClass, options.contains(Consumer.Option.ACQUIRES), + options.contains(Consumer.Option.SEES_REQUEUES), consumerName, + options.contains(Consumer.Option.TRANSIENT), target ); } - public void registerSubscription(Subscription subscription, boolean exclusive) throws AMQException + public String getName() { - + return _name; } - public void unregisterSubscription(Subscription subscription) throws AMQException + @Override + public int send(final ServerMessage message, + final InstanceProperties instanceProperties, + final ServerTransaction txn, + final Action<MessageInstance<? extends Consumer>> postEnqueueAction) { - + return 0; } - public Collection<Subscription> getConsumers() + + public Collection<QueueConsumer> getConsumers() { return Collections.emptyList(); } - public void addSubscriptionRegistrationListener(final SubscriptionRegistrationListener listener) + public void addConsumerRegistrationListener(final ConsumerRegistrationListener listener) { } - public void removeSubscriptionRegistrationListener(final SubscriptionRegistrationListener listener) + public void removeConsumerRegistrationListener(final ConsumerRegistrationListener listener) { } @@ -242,7 +261,7 @@ public class MockAMQQueue implements AMQQueue return 0; } - public boolean hasExclusiveSubscriber() + public boolean hasExclusiveConsumer() { return false; } @@ -293,41 +312,29 @@ public class MockAMQQueue implements AMQQueue return getMessageCount(); } - public void enqueue(ServerMessage message) throws AMQException - { - } - - public void enqueue(ServerMessage message, PostEnqueueAction action) throws AMQException + public void enqueue(ServerMessage message, Action<MessageInstance<? extends Consumer>> action) throws AMQException { } - public void enqueue(ServerMessage message, boolean sync, PostEnqueueAction action) throws AMQException - { - } - public void requeue(QueueEntry entry) { } - public void requeue(QueueEntryImpl storeContext, Subscription subscription) - { - } - - public void dequeue(QueueEntry entry, Subscription sub) + public void dequeue(QueueEntry entry, Consumer sub) { } - public boolean resend(QueueEntry entry, Subscription subscription) throws AMQException + public boolean resend(QueueEntry entry, Consumer consumer) throws AMQException { return false; } - public void addQueueDeleteTask(Task task) + public void addQueueDeleteTask(Action<AMQQueue> task) { } - public void removeQueueDeleteTask(final Task task) + public void removeQueueDeleteTask(final Action<AMQQueue> task) { } @@ -427,12 +434,12 @@ public class MockAMQQueue implements AMQQueue return null; } - public void flushSubscription(Subscription sub) throws AMQException + public void flushConsumer(Consumer sub) throws AMQException { } - public void deliverAsync(Subscription sub) + public void deliverAsync(Consumer sub) { } diff --git a/qpid/java/broker-core/src/test/java/org/apache/qpid/server/queue/MockQueueEntry.java b/qpid/java/broker-core/src/test/java/org/apache/qpid/server/queue/MockQueueEntry.java index d3c866f747..89b366567d 100644 --- a/qpid/java/broker-core/src/test/java/org/apache/qpid/server/queue/MockQueueEntry.java +++ b/qpid/java/broker-core/src/test/java/org/apache/qpid/server/queue/MockQueueEntry.java @@ -24,9 +24,13 @@ import org.apache.qpid.AMQException; import org.apache.qpid.server.filter.Filterable; import org.apache.qpid.server.message.AMQMessageHeader; import org.apache.qpid.server.message.InstanceProperties; +import org.apache.qpid.server.message.MessageInstance; import org.apache.qpid.server.message.ServerMessage; -import org.apache.qpid.server.subscription.Subscription; +import org.apache.qpid.server.store.TransactionLogResource; +import org.apache.qpid.server.consumer.Consumer; import org.apache.qpid.server.txn.ServerTransaction; +import org.apache.qpid.server.util.Action; +import org.apache.qpid.server.util.StateChangeListener; public class MockQueueEntry implements QueueEntry { @@ -38,22 +42,28 @@ public class MockQueueEntry implements QueueEntry return false; } - public boolean acquire(Subscription sub) + public boolean acquire(QueueConsumer sub) { return false; } - public boolean acquiredBySubscription() + @Override + public int getMaximumDeliveryCount() + { + return 0; + } + + public boolean acquiredByConsumer() { return false; } - public boolean isAcquiredBy(Subscription subscription) + public boolean isAcquiredBy(QueueConsumer consumer) { return false; } - public void addStateChangeListener(StateChangeListener listener) + public void addStateChangeListener(StateChangeListener<MessageInstance<QueueConsumer>, State> listener) { } @@ -63,7 +73,7 @@ public class MockQueueEntry implements QueueEntry } - public int routeToAlternate(final BaseQueue.PostEnqueueAction action, final ServerTransaction txn) + public int routeToAlternate(final Action<MessageInstance<? extends Consumer>> action, final ServerTransaction txn) { return 0; } @@ -78,7 +88,7 @@ public class MockQueueEntry implements QueueEntry return false; } - public Subscription getDeliveredSubscription() + public QueueConsumer getDeliveredConsumer() { return null; } @@ -93,7 +103,7 @@ public class MockQueueEntry implements QueueEntry return _message; } - public AMQQueue getQueue() + public AMQQueue<QueueConsumer> getQueue() { return null; } @@ -116,7 +126,7 @@ public class MockQueueEntry implements QueueEntry } - public boolean isRejectedBy(long subscriptionId) + public boolean isRejectedBy(QueueConsumer consumer) { return false; @@ -136,8 +146,14 @@ public class MockQueueEntry implements QueueEntry } + @Override + public boolean resend() throws AMQException + { + return false; + } + - public boolean removeStateChangeListener(StateChangeListener listener) + public boolean removeStateChangeListener(StateChangeListener<MessageInstance<QueueConsumer>, State> listener) { return false; @@ -217,4 +233,10 @@ public class MockQueueEntry implements QueueEntry { return InstanceProperties.EMPTY; } + + @Override + public TransactionLogResource getOwningResource() + { + return getQueue(); + } } diff --git a/qpid/java/broker-core/src/test/java/org/apache/qpid/server/queue/QueueEntryImplTestBase.java b/qpid/java/broker-core/src/test/java/org/apache/qpid/server/queue/QueueEntryImplTestBase.java index 2b1e7f5e1f..95139d8740 100644 --- a/qpid/java/broker-core/src/test/java/org/apache/qpid/server/queue/QueueEntryImplTestBase.java +++ b/qpid/java/broker-core/src/test/java/org/apache/qpid/server/queue/QueueEntryImplTestBase.java @@ -21,11 +21,11 @@ package org.apache.qpid.server.queue; import junit.framework.TestCase; import org.apache.qpid.AMQException; +import org.apache.qpid.server.consumer.ConsumerTarget; import org.apache.qpid.server.message.MessageReference; import org.apache.qpid.server.message.ServerMessage; -import org.apache.qpid.server.queue.QueueEntry.EntryState; -import org.apache.qpid.server.subscription.MockSubscription; -import org.apache.qpid.server.subscription.Subscription; +import org.apache.qpid.server.message.MessageInstance.EntryState; +import org.apache.qpid.server.protocol.AMQSessionModel; import java.lang.reflect.Field; @@ -113,11 +113,19 @@ public abstract class QueueEntryImplTestBase extends TestCase */ private void acquire() { - _queueEntry.acquire(new MockSubscription()); + _queueEntry.acquire(newConsumer()); assertTrue("Queue entry should be in ACQUIRED state after invoking of acquire method", _queueEntry.isAcquired()); } + private QueueConsumer newConsumer() + { + final ConsumerTarget target = mock(ConsumerTarget.class); + when(target.getSessionModel()).thenReturn(mock(AMQSessionModel.class)); + final QueueConsumer consumer = new QueueConsumer(null,null,true,true,"mock",false,target); + return consumer; + } + /** * A helper method to get entry state * @@ -140,36 +148,34 @@ public abstract class QueueEntryImplTestBase extends TestCase } /** - * Tests rejecting a queue entry records the Subscription ID - * for later verification by isRejectedBy(subscriptionId). + * Tests rejecting a queue entry records the Consumer ID + * for later verification by isRejectedBy(consumerId). */ public void testRejectAndRejectedBy() { - Subscription sub = new MockSubscription(); - long subId = sub.getSubscriptionID(); + QueueConsumer sub = newConsumer(); - assertFalse("Queue entry should not yet have been rejected by the subscription", _queueEntry.isRejectedBy(subId)); - assertFalse("Queue entry should not yet have been acquired by a subscription", _queueEntry.isAcquired()); + assertFalse("Queue entry should not yet have been rejected by the consumer", _queueEntry.isRejectedBy(sub)); + assertFalse("Queue entry should not yet have been acquired by a consumer", _queueEntry.isAcquired()); - //acquire, reject, and release the message using the subscription + //acquire, reject, and release the message using the consumer assertTrue("Queue entry should have been able to be acquired", _queueEntry.acquire(sub)); _queueEntry.reject(); _queueEntry.release(); //verify the rejection is recorded - assertTrue("Queue entry should have been rejected by the subscription", _queueEntry.isRejectedBy(subId)); + assertTrue("Queue entry should have been rejected by the consumer", _queueEntry.isRejectedBy(sub)); - //repeat rejection using a second subscription - Subscription sub2 = new MockSubscription(); - long sub2Id = sub2.getSubscriptionID(); + //repeat rejection using a second consumer + QueueConsumer sub2 = newConsumer(); - assertFalse("Queue entry should not yet have been rejected by the subscription", _queueEntry.isRejectedBy(sub2Id)); + assertFalse("Queue entry should not yet have been rejected by the consumer", _queueEntry.isRejectedBy(sub2)); assertTrue("Queue entry should have been able to be acquired", _queueEntry.acquire(sub2)); _queueEntry.reject(); - //verify it still records being rejected by both subscriptions - assertTrue("Queue entry should have been rejected by the subscription", _queueEntry.isRejectedBy(subId)); - assertTrue("Queue entry should have been rejected by the subscription", _queueEntry.isRejectedBy(sub2Id)); + //verify it still records being rejected by both consumers + assertTrue("Queue entry should have been rejected by the consumer", _queueEntry.isRejectedBy(sub)); + assertTrue("Queue entry should have been rejected by the consumer", _queueEntry.isRejectedBy(sub2)); } /** diff --git a/qpid/java/broker-core/src/test/java/org/apache/qpid/server/queue/SimpleAMQQueueTest.java b/qpid/java/broker-core/src/test/java/org/apache/qpid/server/queue/SimpleAMQQueueTest.java index b0e5a510b8..5abc97cee9 100644 --- a/qpid/java/broker-core/src/test/java/org/apache/qpid/server/queue/SimpleAMQQueueTest.java +++ b/qpid/java/broker-core/src/test/java/org/apache/qpid/server/queue/SimpleAMQQueueTest.java @@ -29,6 +29,8 @@ import static org.mockito.Matchers.contains; import static org.mockito.Matchers.eq; import static org.mockito.Mockito.when; +import java.util.Arrays; +import java.util.EnumSet; import java.util.Map; import org.apache.log4j.Logger; @@ -38,13 +40,15 @@ import org.apache.qpid.AMQSecurityException; import org.apache.qpid.exchange.ExchangeDefaults; import org.apache.qpid.server.exchange.DirectExchange; import org.apache.qpid.server.message.AMQMessageHeader; +import org.apache.qpid.server.message.MessageInstance; import org.apache.qpid.server.message.MessageReference; import org.apache.qpid.server.message.ServerMessage; import org.apache.qpid.server.model.UUIDGenerator; -import org.apache.qpid.server.queue.BaseQueue.PostEnqueueAction; import org.apache.qpid.server.queue.SimpleAMQQueue.QueueEntryFilter; -import org.apache.qpid.server.subscription.MockSubscription; -import org.apache.qpid.server.subscription.Subscription; +import org.apache.qpid.server.consumer.ConsumerTarget; +import org.apache.qpid.server.consumer.MockConsumer; +import org.apache.qpid.server.consumer.Consumer; +import org.apache.qpid.server.util.Action; import org.apache.qpid.server.util.BrokerTestHelper; import org.apache.qpid.server.virtualhost.VirtualHost; import org.apache.qpid.test.utils.QpidTestCase; @@ -65,7 +69,8 @@ public class SimpleAMQQueueTest extends QpidTestCase private String _owner = "owner"; private String _routingKey = "routing key"; private DirectExchange _exchange; - private MockSubscription _subscription = new MockSubscription(); + private MockConsumer _consumerTarget = new MockConsumer(); + private QueueConsumer _consumer; private Map<String,Object> _arguments = null; @Override @@ -157,20 +162,21 @@ public class SimpleAMQQueueTest extends QpidTestCase } - public void testRegisterSubscriptionThenEnqueueMessage() throws AMQException + public void testRegisterConsumerThenEnqueueMessage() throws AMQException { - // Check adding a subscription adds it to the queue - _queue.registerSubscription(_subscription, false); - assertEquals("Subscription did not get queue", _queue, - _subscription.getQueue()); + ServerMessage messageA = createMessage(new Long(24)); + + // Check adding a consumer adds it to the queue + _consumer = _queue.addConsumer(_consumerTarget, null, messageA.getClass(), "test", + EnumSet.of(Consumer.Option.ACQUIRES, + Consumer.Option.SEES_REQUEUES)); assertEquals("Queue does not have consumer", 1, _queue.getConsumerCount()); assertEquals("Queue does not have active consumer", 1, - _queue.getActiveConsumerCount()); + _queue.getActiveConsumerCount()); // Check sending a message ends up with the subscriber - ServerMessage messageA = createMessage(new Long(24)); - _queue.enqueue(messageA); + _queue.enqueue(messageA, null); try { Thread.sleep(2000L); @@ -178,45 +184,51 @@ public class SimpleAMQQueueTest extends QpidTestCase catch(InterruptedException e) { } - assertEquals(messageA, _subscription.getQueueContext().getLastSeenEntry().getMessage()); - assertNull(((QueueContext)_subscription.getQueueContext()).getReleasedEntry()); + assertEquals(messageA, _consumer.getQueueContext().getLastSeenEntry().getMessage()); + assertNull(_consumer.getQueueContext().getReleasedEntry()); - // Check removing the subscription removes it's information from the queue - _queue.unregisterSubscription(_subscription); - assertTrue("Subscription still had queue", _subscription.isClosed()); + // Check removing the consumer removes it's information from the queue + _consumer.close(); + assertTrue("Consumer still had queue", _consumerTarget.isClosed()); assertFalse("Queue still has consumer", 1 == _queue.getConsumerCount()); assertFalse("Queue still has active consumer", - 1 == _queue.getActiveConsumerCount()); + 1 == _queue.getActiveConsumerCount()); ServerMessage messageB = createMessage(new Long (25)); - _queue.enqueue(messageB); - assertNull(_subscription.getQueueContext()); + _queue.enqueue(messageB, null); + assertNull(_consumer.getQueueContext()); } - public void testEnqueueMessageThenRegisterSubscription() throws AMQException, InterruptedException + public void testEnqueueMessageThenRegisterConsumer() throws AMQException, InterruptedException { ServerMessage messageA = createMessage(new Long(24)); - _queue.enqueue(messageA); - _queue.registerSubscription(_subscription, false); + _queue.enqueue(messageA, null); + _consumer = _queue.addConsumer(_consumerTarget, null, messageA.getClass(), "test", + EnumSet.of(Consumer.Option.ACQUIRES, + Consumer.Option.SEES_REQUEUES)); Thread.sleep(150); - assertEquals(messageA, _subscription.getQueueContext().getLastSeenEntry().getMessage()); - assertNull("There should be no releasedEntry after an enqueue", ((QueueContext)_subscription.getQueueContext()).getReleasedEntry()); + assertEquals(messageA, _consumer.getQueueContext().getLastSeenEntry().getMessage()); + assertNull("There should be no releasedEntry after an enqueue", + _consumer.getQueueContext().getReleasedEntry()); } /** * Tests enqueuing two messages. */ - public void testEnqueueTwoMessagesThenRegisterSubscription() throws Exception + public void testEnqueueTwoMessagesThenRegisterConsumer() throws Exception { ServerMessage messageA = createMessage(new Long(24)); ServerMessage messageB = createMessage(new Long(25)); - _queue.enqueue(messageA); - _queue.enqueue(messageB); - _queue.registerSubscription(_subscription, false); + _queue.enqueue(messageA, null); + _queue.enqueue(messageB, null); + _consumer = _queue.addConsumer(_consumerTarget, null, messageA.getClass(), "test", + EnumSet.of(Consumer.Option.ACQUIRES, + Consumer.Option.SEES_REQUEUES)); Thread.sleep(150); - assertEquals(messageB, _subscription.getQueueContext().getLastSeenEntry().getMessage()); - assertNull("There should be no releasedEntry after enqueues", ((QueueContext)_subscription.getQueueContext()).getReleasedEntry()); + assertEquals(messageB, _consumer.getQueueContext().getLastSeenEntry().getMessage()); + assertNull("There should be no releasedEntry after enqueues", + _consumer.getQueueContext().getReleasedEntry()); } /** @@ -225,21 +237,19 @@ public class SimpleAMQQueueTest extends QpidTestCase */ public void testReleasedMessageIsResentToSubscriber() throws Exception { - _queue.registerSubscription(_subscription, false); - - final ArrayList<QueueEntry> queueEntries = new ArrayList<QueueEntry>(); - PostEnqueueAction postEnqueueAction = new PostEnqueueAction() - { - public void onEnqueue(QueueEntry entry) - { - queueEntries.add(entry); - } - }; ServerMessage messageA = createMessage(new Long(24)); ServerMessage messageB = createMessage(new Long(25)); ServerMessage messageC = createMessage(new Long(26)); + + _consumer = _queue.addConsumer(_consumerTarget, null, messageA.getClass(), "test", + EnumSet.of(Consumer.Option.ACQUIRES, + Consumer.Option.SEES_REQUEUES)); + + final ArrayList<QueueEntry> queueEntries = new ArrayList<QueueEntry>(); + EntryListAddingAction postEnqueueAction = new EntryListAddingAction(queueEntries); + /* Enqueue three messages */ _queue.enqueue(messageA, postEnqueueAction); @@ -248,7 +258,9 @@ public class SimpleAMQQueueTest extends QpidTestCase Thread.sleep(150); // Work done by SubFlushRunner/QueueRunner Threads - assertEquals("Unexpected total number of messages sent to subscription", 3, _subscription.getMessages().size()); + assertEquals("Unexpected total number of messages sent to consumer", + 3, + _consumerTarget.getMessages().size()); assertFalse("Redelivery flag should not be set", queueEntries.get(0).isRedelivered()); assertFalse("Redelivery flag should not be set", queueEntries.get(1).isRedelivered()); assertFalse("Redelivery flag should not be set", queueEntries.get(2).isRedelivered()); @@ -259,11 +271,14 @@ public class SimpleAMQQueueTest extends QpidTestCase Thread.sleep(150); // Work done by SubFlushRunner/QueueRunner Threads - assertEquals("Unexpected total number of messages sent to subscription", 4, _subscription.getMessages().size()); + assertEquals("Unexpected total number of messages sent to consumer", + 4, + _consumerTarget.getMessages().size()); assertTrue("Redelivery flag should now be set", queueEntries.get(0).isRedelivered()); assertFalse("Redelivery flag should remain be unset", queueEntries.get(1).isRedelivered()); assertFalse("Redelivery flag should remain be unset",queueEntries.get(2).isRedelivered()); - assertNull("releasedEntry should be cleared after requeue processed", ((QueueContext)_subscription.getQueueContext()).getReleasedEntry()); + assertNull("releasedEntry should be cleared after requeue processed", + _consumer.getQueueContext().getReleasedEntry()); } /** @@ -273,20 +288,17 @@ public class SimpleAMQQueueTest extends QpidTestCase */ public void testReleaseMessageThatBecomesExpiredIsNotRedelivered() throws Exception { - _queue.registerSubscription(_subscription, false); + ServerMessage messageA = createMessage(new Long(24)); + + _consumer = _queue.addConsumer(_consumerTarget, null, messageA.getClass(), "test", + EnumSet.of(Consumer.Option.SEES_REQUEUES, + Consumer.Option.ACQUIRES)); final ArrayList<QueueEntry> queueEntries = new ArrayList<QueueEntry>(); - PostEnqueueAction postEnqueueAction = new PostEnqueueAction() - { - public void onEnqueue(QueueEntry entry) - { - queueEntries.add(entry); - } - }; + EntryListAddingAction postEnqueueAction = new EntryListAddingAction(queueEntries); /* Enqueue one message with expiration set for a short time in the future */ - ServerMessage messageA = createMessage(new Long(24)); int messageExpirationOffset = 200; final long expiration = System.currentTimeMillis() + messageExpirationOffset; when(messageA.getExpiration()).thenReturn(expiration); @@ -296,7 +308,9 @@ public class SimpleAMQQueueTest extends QpidTestCase int subFlushWaitTime = 150; Thread.sleep(subFlushWaitTime); // Work done by SubFlushRunner/QueueRunner Threads - assertEquals("Unexpected total number of messages sent to subscription", 1, _subscription.getMessages().size()); + assertEquals("Unexpected total number of messages sent to consumer", + 1, + _consumerTarget.getMessages().size()); assertFalse("Redelivery flag should not be set", queueEntries.get(0).isRedelivered()); /* Wait a little more to be sure that message will have expired, then release the first message only, causing it to be requeued */ @@ -306,9 +320,12 @@ public class SimpleAMQQueueTest extends QpidTestCase Thread.sleep(subFlushWaitTime); // Work done by SubFlushRunner/QueueRunner Threads assertTrue("Expecting the queue entry to be now expired", queueEntries.get(0).expired()); - assertEquals("Total number of messages sent should not have changed", 1, _subscription.getMessages().size()); + assertEquals("Total number of messages sent should not have changed", + 1, + _consumerTarget.getMessages().size()); assertFalse("Redelivery flag should not be set", queueEntries.get(0).isRedelivered()); - assertNull("releasedEntry should be cleared after requeue processed", ((QueueContext)_subscription.getQueueContext()).getReleasedEntry()); + assertNull("releasedEntry should be cleared after requeue processed", + _consumer.getQueueContext().getReleasedEntry()); } @@ -320,21 +337,18 @@ public class SimpleAMQQueueTest extends QpidTestCase */ public void testReleasedOutOfComparableOrderAreRedelivered() throws Exception { - _queue.registerSubscription(_subscription, false); - - final ArrayList<QueueEntry> queueEntries = new ArrayList<QueueEntry>(); - PostEnqueueAction postEnqueueAction = new PostEnqueueAction() - { - public void onEnqueue(QueueEntry entry) - { - queueEntries.add(entry); - } - }; ServerMessage messageA = createMessage(new Long(24)); ServerMessage messageB = createMessage(new Long(25)); ServerMessage messageC = createMessage(new Long(26)); + _consumer = _queue.addConsumer(_consumerTarget, null, messageA.getClass(), "test", + EnumSet.of(Consumer.Option.ACQUIRES, + Consumer.Option.SEES_REQUEUES)); + + final ArrayList<QueueEntry> queueEntries = new ArrayList<QueueEntry>(); + EntryListAddingAction postEnqueueAction = new EntryListAddingAction(queueEntries); + /* Enqueue three messages */ _queue.enqueue(messageA, postEnqueueAction); @@ -343,7 +357,9 @@ public class SimpleAMQQueueTest extends QpidTestCase Thread.sleep(150); // Work done by SubFlushRunner/QueueRunner Threads - assertEquals("Unexpected total number of messages sent to subscription", 3, _subscription.getMessages().size()); + assertEquals("Unexpected total number of messages sent to consumer", + 3, + _consumerTarget.getMessages().size()); assertFalse("Redelivery flag should not be set", queueEntries.get(0).isRedelivered()); assertFalse("Redelivery flag should not be set", queueEntries.get(1).isRedelivered()); assertFalse("Redelivery flag should not be set", queueEntries.get(2).isRedelivered()); @@ -355,37 +371,41 @@ public class SimpleAMQQueueTest extends QpidTestCase Thread.sleep(150); // Work done by SubFlushRunner/QueueRunner Threads - assertEquals("Unexpected total number of messages sent to subscription", 5, _subscription.getMessages().size()); + assertEquals("Unexpected total number of messages sent to consumer", + 5, + _consumerTarget.getMessages().size()); assertTrue("Redelivery flag should now be set", queueEntries.get(0).isRedelivered()); assertFalse("Redelivery flag should remain be unset", queueEntries.get(1).isRedelivered()); assertTrue("Redelivery flag should now be set",queueEntries.get(2).isRedelivered()); - assertNull("releasedEntry should be cleared after requeue processed", ((QueueContext)_subscription.getQueueContext()).getReleasedEntry()); + assertNull("releasedEntry should be cleared after requeue processed", + _consumer.getQueueContext().getReleasedEntry()); } /** - * Tests that a release requeues an entry for a queue with multiple subscriptions. Verifies that a + * Tests that a release requeues an entry for a queue with multiple consumers. Verifies that a * requeue resends a message to a <i>single</i> subscriber. */ - public void testReleaseForQueueWithMultipleSubscriptions() throws Exception + public void testReleaseForQueueWithMultipleConsumers() throws Exception { - MockSubscription subscription1 = new MockSubscription(); - MockSubscription subscription2 = new MockSubscription(); + ServerMessage messageA = createMessage(new Long(24)); + ServerMessage messageB = createMessage(new Long(25)); - _queue.registerSubscription(subscription1, false); - _queue.registerSubscription(subscription2, false); + MockConsumer target1 = new MockConsumer(); + MockConsumer target2 = new MockConsumer(); - final ArrayList<QueueEntry> queueEntries = new ArrayList<QueueEntry>(); - PostEnqueueAction postEnqueueAction = new PostEnqueueAction() - { - public void onEnqueue(QueueEntry entry) - { - queueEntries.add(entry); - } - }; - ServerMessage messageA = createMessage(new Long(24)); - ServerMessage messageB = createMessage(new Long(25)); + QueueConsumer consumer1 = _queue.addConsumer(target1, null, messageA.getClass(), "test", + EnumSet.of(Consumer.Option.ACQUIRES, + Consumer.Option.SEES_REQUEUES)); + + QueueConsumer consumer2 = _queue.addConsumer(target2, null, messageA.getClass(), "test", + EnumSet.of(Consumer.Option.ACQUIRES, + Consumer.Option.SEES_REQUEUES)); + + + final ArrayList<QueueEntry> queueEntries = new ArrayList<QueueEntry>(); + EntryListAddingAction postEnqueueAction = new EntryListAddingAction(queueEntries); /* Enqueue two messages */ @@ -394,32 +414,40 @@ public class SimpleAMQQueueTest extends QpidTestCase Thread.sleep(150); // Work done by SubFlushRunner/QueueRunner Threads - assertEquals("Unexpected total number of messages sent to both after enqueue", 2, subscription1.getMessages().size() + subscription2.getMessages().size()); + assertEquals("Unexpected total number of messages sent to both after enqueue", + 2, + target1.getMessages().size() + target2.getMessages().size()); /* Now release the first message only, causing it to be requeued */ queueEntries.get(0).release(); Thread.sleep(150); // Work done by SubFlushRunner/QueueRunner Threads - assertEquals("Unexpected total number of messages sent to both subscriptions after release", 3, subscription1.getMessages().size() + subscription2.getMessages().size()); - assertNull("releasedEntry should be cleared after requeue processed", ((QueueContext)subscription1.getQueueContext()).getReleasedEntry()); - assertNull("releasedEntry should be cleared after requeue processed", ((QueueContext)subscription2.getQueueContext()).getReleasedEntry()); + assertEquals("Unexpected total number of messages sent to both consumers after release", + 3, + target1.getMessages().size() + target2.getMessages().size()); + assertNull("releasedEntry should be cleared after requeue processed", + consumer1.getQueueContext().getReleasedEntry()); + assertNull("releasedEntry should be cleared after requeue processed", + consumer2.getQueueContext().getReleasedEntry()); } public void testExclusiveConsumer() throws AMQException { - // Check adding an exclusive subscription adds it to the queue - _queue.registerSubscription(_subscription, true); - assertEquals("Subscription did not get queue", _queue, - _subscription.getQueue()); + ServerMessage messageA = createMessage(new Long(24)); + // Check adding an exclusive consumer adds it to the queue + + _consumer = _queue.addConsumer(_consumerTarget, null, messageA.getClass(), "test", + EnumSet.of(Consumer.Option.EXCLUSIVE, Consumer.Option.ACQUIRES, + Consumer.Option.SEES_REQUEUES)); + assertEquals("Queue does not have consumer", 1, - _queue.getConsumerCount()); + _queue.getConsumerCount()); assertEquals("Queue does not have active consumer", 1, - _queue.getActiveConsumerCount()); + _queue.getActiveConsumerCount()); // Check sending a message ends up with the subscriber - ServerMessage messageA = createMessage(new Long(24)); - _queue.enqueue(messageA); + _queue.enqueue(messageA, null); try { Thread.sleep(2000L); @@ -427,14 +455,18 @@ public class SimpleAMQQueueTest extends QpidTestCase catch (InterruptedException e) { } - assertEquals(messageA, _subscription.getQueueContext().getLastSeenEntry().getMessage()); + assertEquals(messageA, _consumer.getQueueContext().getLastSeenEntry().getMessage()); // Check we cannot add a second subscriber to the queue - Subscription subB = new MockSubscription(); + MockConsumer subB = new MockConsumer(); Exception ex = null; try { - _queue.registerSubscription(subB, false); + + _queue.addConsumer(subB, null, messageA.getClass(), "test", + EnumSet.of(Consumer.Option.ACQUIRES, + Consumer.Option.SEES_REQUEUES)); + } catch (AMQException e) { @@ -443,12 +475,18 @@ public class SimpleAMQQueueTest extends QpidTestCase assertNotNull(ex); // Check we cannot add an exclusive subscriber to a queue with an - // existing subscription - _queue.unregisterSubscription(_subscription); - _queue.registerSubscription(_subscription, false); + // existing consumer + _consumer.close(); + _consumer = _queue.addConsumer(_consumerTarget, null, messageA.getClass(), "test", + EnumSet.of(Consumer.Option.ACQUIRES, + Consumer.Option.SEES_REQUEUES)); + try { - _queue.registerSubscription(subB, true); + + _consumer = _queue.addConsumer(subB, null, messageA.getClass(), "test", + EnumSet.of(Consumer.Option.EXCLUSIVE)); + } catch (AMQException e) { @@ -462,23 +500,45 @@ public class SimpleAMQQueueTest extends QpidTestCase _queue.stop(); _queue = new SimpleAMQQueue(UUIDGenerator.generateRandomUUID(), _qname, false, null, true, false, _virtualHost, Collections.EMPTY_MAP); _queue.setDeleteOnNoConsumers(true); - _queue.registerSubscription(_subscription, false); - ServerMessage message = createMessage(new Long(25)); - _queue.enqueue(message); - _queue.unregisterSubscription(_subscription); - assertTrue("Queue was not deleted when subscription was removed", + + ServerMessage message = createMessage(new Long(25)); + _consumer = _queue.addConsumer(_consumerTarget, null, message.getClass(), "test", + EnumSet.of(Consumer.Option.ACQUIRES, + Consumer.Option.SEES_REQUEUES)); + + _queue.enqueue(message, null); + _consumer.close(); + assertTrue("Queue was not deleted when consumer was removed", _queue.isDeleted()); } public void testResend() throws Exception { - _queue.registerSubscription(_subscription, false); Long id = new Long(26); ServerMessage message = createMessage(id); - _queue.enqueue(message); - QueueEntry entry = _subscription.getQueueContext().getLastSeenEntry(); - entry.setRedelivered(); - _queue.resend(entry, _subscription); + + _consumer = _queue.addConsumer(_consumerTarget, null, message.getClass(), "test", + EnumSet.of(Consumer.Option.ACQUIRES, Consumer.Option.SEES_REQUEUES)); + + _queue.enqueue(message, new Action<MessageInstance<? extends Consumer>>() + { + @Override + public void performAction(final MessageInstance<? extends Consumer> object) + { + QueueEntry entry = (QueueEntry) object; + entry.setRedelivered(); + try + { + _consumer.resend(entry); + } + catch (AMQException e) + { + fail("Exception thrown: " + e.getMessage()); + } + } + }); + + } @@ -489,7 +549,7 @@ public class SimpleAMQQueueTest extends QpidTestCase ServerMessage message = createMessage(messageId); // Put message on queue - _queue.enqueue(message); + _queue.enqueue(message, null); // Get message id Long testmsgid = _queue.getMessagesOnTheQueue(1).get(0); @@ -505,7 +565,7 @@ public class SimpleAMQQueueTest extends QpidTestCase Long messageId = new Long(i); ServerMessage message = createMessage(messageId); // Put message on queue - _queue.enqueue(message); + _queue.enqueue(message, null); } // Get message ids List<Long> msgids = _queue.getMessagesOnTheQueue(5); @@ -526,7 +586,7 @@ public class SimpleAMQQueueTest extends QpidTestCase Long messageId = new Long(i); ServerMessage message = createMessage(messageId); // Put message on queue - _queue.enqueue(message); + _queue.enqueue(message, null); } // Get message ids List<Long> msgids = _queue.getMessagesOnTheQueue(5, 5); @@ -547,7 +607,7 @@ public class SimpleAMQQueueTest extends QpidTestCase Long messageId = new Long(i); ServerMessage message = createMessage(messageId); // Put message on queue - _queue.enqueue(message); + _queue.enqueue(message, null); } // Get non-existent 0th QueueEntry & check returned list was empty @@ -605,19 +665,19 @@ public class SimpleAMQQueueTest extends QpidTestCase /** * processQueue() is used when asynchronously delivering messages to - * subscriptions which could not be delivered immediately during the + * consumers which could not be delivered immediately during the * enqueue() operation. * * A defect within the method would mean that delivery of these messages may * not occur should the Runner stop before all messages have been processed. * Such a defect was discovered when Selectors were used such that one and - * only one subscription can/will accept any given messages, but multiple - * subscriptions are present, and one of the earlier subscriptions receives + * only one consumer can/will accept any given messages, but multiple + * consumers are present, and one of the earlier consumers receives * more messages than the others. * * This test is to validate that the processQueue() method is able to * correctly deliver all of the messages present for asynchronous delivery - * to subscriptions in such a scenario. + * to consumers in such a scenario. */ public void testProcessQueueWithUniqueSelectors() throws Exception { @@ -626,10 +686,10 @@ public class SimpleAMQQueueTest extends QpidTestCase false, false, _virtualHost, factory, null) { @Override - public void deliverAsync(Subscription sub) + public void deliverAsync(QueueConsumer sub) { // do nothing, i.e prevent deliveries by the SubFlushRunner - // when registering the new subscriptions + // when registering the new consumers } }; @@ -645,25 +705,28 @@ public class SimpleAMQQueueTest extends QpidTestCase QueueEntry msg4 = list.add(createMessage(4L)); QueueEntry msg5 = list.add(createMessage(5L)); - // Create lists of the entries each subscription should be interested - // in.Bias over 50% of the messages to the first subscription so that - // the later subscriptions reject them and report being done before - // the first subscription as the processQueue method proceeds. - List<QueueEntry> msgListSub1 = createEntriesList(msg1, msg2, msg3); - List<QueueEntry> msgListSub2 = createEntriesList(msg4); - List<QueueEntry> msgListSub3 = createEntriesList(msg5); - - MockSubscription sub1 = new MockSubscription(msgListSub1); - MockSubscription sub2 = new MockSubscription(msgListSub2); - MockSubscription sub3 = new MockSubscription(msgListSub3); - - // register the subscriptions - testQueue.registerSubscription(sub1, false); - testQueue.registerSubscription(sub2, false); - testQueue.registerSubscription(sub3, false); + // Create lists of the entries each consumer should be interested + // in.Bias over 50% of the messages to the first consumer so that + // the later consumers reject them and report being done before + // the first consumer as the processQueue method proceeds. + List<String> msgListSub1 = createEntriesList(msg1, msg2, msg3); + List<String> msgListSub2 = createEntriesList(msg4); + List<String> msgListSub3 = createEntriesList(msg5); + + MockConsumer sub1 = new MockConsumer(msgListSub1); + MockConsumer sub2 = new MockConsumer(msgListSub2); + MockConsumer sub3 = new MockConsumer(msgListSub3); + + // register the consumers + testQueue.addConsumer(sub1, sub1.getFilters(), msg1.getMessage().getClass(), "test", + EnumSet.of(Consumer.Option.ACQUIRES, Consumer.Option.SEES_REQUEUES)); + testQueue.addConsumer(sub2, sub2.getFilters(), msg1.getMessage().getClass(), "test", + EnumSet.of(Consumer.Option.ACQUIRES, Consumer.Option.SEES_REQUEUES)); + testQueue.addConsumer(sub3, sub3.getFilters(), msg1.getMessage().getClass(), "test", + EnumSet.of(Consumer.Option.ACQUIRES, Consumer.Option.SEES_REQUEUES)); //check that no messages have been delivered to the - //subscriptions during registration + //consumers during registration assertEquals("No messages should have been delivered yet", 0, sub1.getMessages().size()); assertEquals("No messages should have been delivered yet", 0, sub2.getMessages().size()); assertEquals("No messages should have been delivered yet", 0, sub3.getMessages().size()); @@ -680,9 +743,9 @@ public class SimpleAMQQueueTest extends QpidTestCase }); // check expected messages delivered to correct consumers - verifyReceivedMessages(msgListSub1, sub1.getMessages()); - verifyReceivedMessages(msgListSub2, sub2.getMessages()); - verifyReceivedMessages(msgListSub3, sub3.getMessages()); + verifyReceivedMessages(Arrays.asList((MessageInstance)msg1,msg2,msg3), sub1.getMessages()); + verifyReceivedMessages(Collections.singletonList((MessageInstance)msg4), sub2.getMessages()); + verifyReceivedMessages(Collections.singletonList((MessageInstance)msg5), sub3.getMessages()); } /** @@ -850,7 +913,7 @@ public class SimpleAMQQueueTest extends QpidTestCase false, "testOwner", false, false, _virtualHost, null) { @Override - public void deliverAsync(Subscription sub) + public void deliverAsync(QueueConsumer sub) { // do nothing } @@ -865,15 +928,15 @@ public class SimpleAMQQueueTest extends QpidTestCase // latch to wait for message receipt final CountDownLatch latch = new CountDownLatch(messageNumber -1); - // create a subscription - MockSubscription subscription = new MockSubscription() + // create a consumer + MockConsumer consumer = new MockConsumer() { /** * Send a message and decrement latch * @param entry * @param batch */ - public void send(QueueEntry entry, boolean batch) throws AMQException + public void send(MessageInstance entry, boolean batch) throws AMQException { super.send(entry, batch); latch.countDown(); @@ -883,7 +946,12 @@ public class SimpleAMQQueueTest extends QpidTestCase try { // subscribe - testQueue.registerSubscription(subscription, false); + testQueue.addConsumer(consumer, + null, + entries.get(0).getMessage().getClass(), + "test", + EnumSet.of(Consumer.Option.ACQUIRES, + Consumer.Option.SEES_REQUEUES)); // process queue testQueue.processQueue(new QueueRunner(testQueue) @@ -907,12 +975,12 @@ public class SimpleAMQQueueTest extends QpidTestCase { Thread.currentThread().interrupt(); } - List<QueueEntry> expected = createEntriesList(entries.get(0), entries.get(2), entries.get(3)); - verifyReceivedMessages(expected, subscription.getMessages()); + List<MessageInstance> expected = Arrays.asList((MessageInstance)entries.get(0), entries.get(2), entries.get(3)); + verifyReceivedMessages(expected, consumer.getMessages()); } /** - * Tests that entry in dequeued state are not enqueued and not delivered to subscription + * Tests that entry in dequeued state are not enqueued and not delivered to consumer */ public void testEnqueueDequeuedEntry() { @@ -948,7 +1016,7 @@ public class SimpleAMQQueueTest extends QpidTestCase } @Override - public boolean acquire(Subscription sub) + public boolean acquire(QueueConsumer sub) { if(message.getMessageNumber() % 2 == 0) { @@ -964,24 +1032,29 @@ public class SimpleAMQQueueTest extends QpidTestCase }; } }, null); - // create a subscription - MockSubscription subscription = new MockSubscription(); + // create a consumer + MockConsumer consumer = new MockConsumer(); - // register subscription + // register consumer try { - queue.registerSubscription(subscription, false); + queue.addConsumer(consumer, + null, + createMessage(-1l).getClass(), + "test", + EnumSet.of(Consumer.Option.ACQUIRES, + Consumer.Option.SEES_REQUEUES)); } catch (AMQException e) { - fail("Failure to register subscription:" + e.getMessage()); + fail("Failure to register consumer:" + e.getMessage()); } // put test messages into a queue putGivenNumberOfMessages(queue, 4); // assert received messages - List<QueueEntry> messages = subscription.getMessages(); + List<MessageInstance> messages = consumer.getMessages(); assertEquals("Only 2 messages should be returned", 2, messages.size()); assertEquals("ID of first message should be 1", 1l, (messages.get(0).getMessage()).getMessageNumber()); @@ -994,55 +1067,64 @@ public class SimpleAMQQueueTest extends QpidTestCase final SimpleAMQQueue queue = new SimpleAMQQueue(UUIDGenerator.generateRandomUUID(), "testActiveConsumerCount", false, "testOwner", false, false, _virtualHost, new SimpleQueueEntryList.Factory(), null); - //verify adding an active subscription increases the count - final MockSubscription subscription1 = new MockSubscription(); - subscription1.setActive(true); + //verify adding an active consumer increases the count + final MockConsumer consumer1 = new MockConsumer(); + consumer1.setActive(true); + consumer1.setState(ConsumerTarget.State.ACTIVE); assertEquals("Unexpected active consumer count", 0, queue.getActiveConsumerCount()); - queue.registerSubscription(subscription1, false); + queue.addConsumer(consumer1, + null, + createMessage(-1l).getClass(), + "test", + EnumSet.of(Consumer.Option.ACQUIRES, + Consumer.Option.SEES_REQUEUES)); assertEquals("Unexpected active consumer count", 1, queue.getActiveConsumerCount()); - //verify adding an inactive subscription doesn't increase the count - final MockSubscription subscription2 = new MockSubscription(); - subscription2.setActive(false); + //verify adding an inactive consumer doesn't increase the count + final MockConsumer consumer2 = new MockConsumer(); + consumer2.setActive(false); + consumer2.setState(ConsumerTarget.State.SUSPENDED); assertEquals("Unexpected active consumer count", 1, queue.getActiveConsumerCount()); - queue.registerSubscription(subscription2, false); + queue.addConsumer(consumer2, + null, + createMessage(-1l).getClass(), + "test", + EnumSet.of(Consumer.Option.ACQUIRES, + Consumer.Option.SEES_REQUEUES)); assertEquals("Unexpected active consumer count", 1, queue.getActiveConsumerCount()); //verify behaviour in face of expected state changes: - //verify a subscription going suspended->active increases the count - queue.stateChange(subscription2, Subscription.State.SUSPENDED, Subscription.State.ACTIVE); + //verify a consumer going suspended->active increases the count + consumer2.setState(ConsumerTarget.State.ACTIVE); assertEquals("Unexpected active consumer count", 2, queue.getActiveConsumerCount()); - //verify a subscription going active->suspended decreases the count - queue.stateChange(subscription2, Subscription.State.ACTIVE, Subscription.State.SUSPENDED); + //verify a consumer going active->suspended decreases the count + consumer2.setState(ConsumerTarget.State.SUSPENDED); assertEquals("Unexpected active consumer count", 1, queue.getActiveConsumerCount()); - //verify a subscription going suspended->closed doesn't change the count - queue.stateChange(subscription2, Subscription.State.SUSPENDED, Subscription.State.CLOSED); + //verify a consumer going suspended->closed doesn't change the count + consumer2.setState(ConsumerTarget.State.CLOSED); assertEquals("Unexpected active consumer count", 1, queue.getActiveConsumerCount()); - //verify a subscription going active->closed decreases the count - queue.stateChange(subscription2, Subscription.State.ACTIVE, Subscription.State.CLOSED); - assertEquals("Unexpected active consumer count", 0, queue.getActiveConsumerCount()); + //verify a consumer going active->active doesn't change the count + consumer1.setState(ConsumerTarget.State.ACTIVE); + assertEquals("Unexpected active consumer count", 1, queue.getActiveConsumerCount()); - //verify behaviour in face of unexpected state changes: + consumer1.setState(ConsumerTarget.State.SUSPENDED); + assertEquals("Unexpected active consumer count", 0, queue.getActiveConsumerCount()); - //verify a subscription going closed->active increases the count - queue.stateChange(subscription2, Subscription.State.CLOSED, Subscription.State.ACTIVE); - assertEquals("Unexpected active consumer count", 1, queue.getActiveConsumerCount()); + //verify a consumer going suspended->suspended doesn't change the count + consumer1.setState(ConsumerTarget.State.SUSPENDED); + assertEquals("Unexpected active consumer count", 0, queue.getActiveConsumerCount()); - //verify a subscription going active->active doesn't change the count - queue.stateChange(subscription2, Subscription.State.ACTIVE, Subscription.State.ACTIVE); + consumer1.setState(ConsumerTarget.State.ACTIVE); assertEquals("Unexpected active consumer count", 1, queue.getActiveConsumerCount()); - //verify a subscription going closed->suspended doesn't change the count - queue.stateChange(subscription2, Subscription.State.CLOSED, Subscription.State.SUSPENDED); - assertEquals("Unexpected active consumer count", 1, queue.getActiveConsumerCount()); + //verify a consumer going active->closed decreases the count + consumer1.setState(ConsumerTarget.State.CLOSED); + assertEquals("Unexpected active consumer count", 0, queue.getActiveConsumerCount()); - //verify a subscription going suspended->suspended doesn't change the count - queue.stateChange(subscription2, Subscription.State.SUSPENDED, Subscription.State.SUSPENDED); - assertEquals("Unexpected active consumer count", 1, queue.getActiveConsumerCount()); } public void testNotificationFiredOnEnqueue() throws Exception @@ -1052,10 +1134,10 @@ public class SimpleAMQQueueTest extends QpidTestCase _queue.setNotificationListener(listener); _queue.setMaximumMessageCount(2); - _queue.enqueue(createMessage(new Long(24))); + _queue.enqueue(createMessage(new Long(24)), null); verifyZeroInteractions(listener); - _queue.enqueue(createMessage(new Long(25))); + _queue.enqueue(createMessage(new Long(25)), null); verify(listener, atLeastOnce()).notifyClients(eq(NotificationCheck.MESSAGE_COUNT_ALERT), eq(_queue), contains("Maximum count on queue threshold")); } @@ -1064,9 +1146,9 @@ public class SimpleAMQQueueTest extends QpidTestCase { AMQQueue.NotificationListener listener = mock(AMQQueue.NotificationListener.class); - _queue.enqueue(createMessage(new Long(24))); - _queue.enqueue(createMessage(new Long(25))); - _queue.enqueue(createMessage(new Long(26))); + _queue.enqueue(createMessage(new Long(24)), null); + _queue.enqueue(createMessage(new Long(25)), null); + _queue.enqueue(createMessage(new Long(26)), null); _queue.setNotificationListener(listener); _queue.setMaximumMessageCount(2); @@ -1132,7 +1214,7 @@ public class SimpleAMQQueueTest extends QpidTestCase // Put message on queue try { - queue.enqueue(message); + queue.enqueue(message,null); } catch (AMQException e) { @@ -1167,23 +1249,23 @@ public class SimpleAMQQueueTest extends QpidTestCase return entry; } - private List<QueueEntry> createEntriesList(QueueEntry... entries) + private List<String> createEntriesList(QueueEntry... entries) { - ArrayList<QueueEntry> entriesList = new ArrayList<QueueEntry>(); + ArrayList<String> entriesList = new ArrayList<String>(); for (QueueEntry entry : entries) { - entriesList.add(entry); + entriesList.add(entry.getMessage().getMessageHeader().getMessageId()); } return entriesList; } - private void verifyReceivedMessages(List<QueueEntry> expected, - List<QueueEntry> delivered) + private void verifyReceivedMessages(List<MessageInstance> expected, + List<MessageInstance> delivered) { assertEquals("Consumer did not receive the expected number of messages", expected.size(), delivered.size()); - for (QueueEntry msg : expected) + for (MessageInstance msg : expected) { assertTrue("Consumer did not receive msg: " + msg.getMessage().getMessageNumber(), delivered.contains(msg)); @@ -1195,9 +1277,9 @@ public class SimpleAMQQueueTest extends QpidTestCase return _queue; } - public MockSubscription getSubscription() + public MockConsumer getConsumer() { - return _subscription; + return _consumerTarget; } public Map<String,Object> getArguments() @@ -1213,20 +1295,36 @@ public class SimpleAMQQueueTest extends QpidTestCase protected ServerMessage createMessage(Long id) throws AMQException { + AMQMessageHeader header = mock(AMQMessageHeader.class); + when(header.getMessageId()).thenReturn(String.valueOf(id)); ServerMessage message = mock(ServerMessage.class); when(message.getMessageNumber()).thenReturn(id); + when(message.getMessageHeader()).thenReturn(header); MessageReference ref = mock(MessageReference.class); when(ref.getMessage()).thenReturn(message); - AMQMessageHeader hdr = mock(AMQMessageHeader.class); - when(message.getMessageHeader()).thenReturn(hdr); when(message.newReference()).thenReturn(ref); return message; } + private static class EntryListAddingAction implements Action<MessageInstance<? extends Consumer>> + { + private final ArrayList<QueueEntry> _queueEntries; + + public EntryListAddingAction(final ArrayList<QueueEntry> queueEntries) + { + _queueEntries = queueEntries; + } + + public void performAction(MessageInstance<? extends Consumer> entry) + { + _queueEntries.add((QueueEntry) entry); + } + } + class TestSimpleQueueEntryListFactory implements QueueEntryListFactory { QueueEntryList _list; diff --git a/qpid/java/broker-core/src/test/java/org/apache/qpid/server/store/AbstractDurableConfigurationStoreTestCase.java b/qpid/java/broker-core/src/test/java/org/apache/qpid/server/store/AbstractDurableConfigurationStoreTestCase.java index 3dfe057285..d3ee938586 100644 --- a/qpid/java/broker-core/src/test/java/org/apache/qpid/server/store/AbstractDurableConfigurationStoreTestCase.java +++ b/qpid/java/broker-core/src/test/java/org/apache/qpid/server/store/AbstractDurableConfigurationStoreTestCase.java @@ -464,7 +464,7 @@ public abstract class AbstractDurableConfigurationStoreTestCase extends QpidTest } @Override - public TransactionLogResource getQueue() + public TransactionLogResource getResource() { return _queue; } @@ -505,7 +505,7 @@ public abstract class AbstractDurableConfigurationStoreTestCase extends QpidTest { return false; } - if (_queue == null && other.getQueue() != null) + if (_queue == null && other.getResource() != null) { return false; } @@ -513,7 +513,7 @@ public abstract class AbstractDurableConfigurationStoreTestCase extends QpidTest { return false; } - return _queue.getId().equals(other.getQueue().getId()); + return _queue.getId().equals(other.getResource().getId()); } } diff --git a/qpid/java/broker-core/src/test/java/org/apache/qpid/server/store/MessageStoreQuotaEventsTestBase.java b/qpid/java/broker-core/src/test/java/org/apache/qpid/server/store/MessageStoreQuotaEventsTestBase.java index 121c380736..7a4f92f0ca 100644 --- a/qpid/java/broker-core/src/test/java/org/apache/qpid/server/store/MessageStoreQuotaEventsTestBase.java +++ b/qpid/java/broker-core/src/test/java/org/apache/qpid/server/store/MessageStoreQuotaEventsTestBase.java @@ -154,6 +154,12 @@ public abstract class MessageStoreQuotaEventsTestBase extends QpidTestCase imple return _transactionResource; } + @Override + public boolean isDurable() + { + return true; + } + private static class TestMessage implements EnqueueableMessage { private final StoredMessage<?> _handle; diff --git a/qpid/java/broker-core/src/test/java/org/apache/qpid/server/store/TestMessageMetaData.java b/qpid/java/broker-core/src/test/java/org/apache/qpid/server/store/TestMessageMetaData.java index a52b9f8d14..e14b41b221 100644 --- a/qpid/java/broker-core/src/test/java/org/apache/qpid/server/store/TestMessageMetaData.java +++ b/qpid/java/broker-core/src/test/java/org/apache/qpid/server/store/TestMessageMetaData.java @@ -26,7 +26,6 @@ import java.nio.ByteBuffer; import org.apache.qpid.framing.EncodingUtils; import org.apache.qpid.server.plugin.MessageMetaDataType; -import org.apache.qpid.server.store.StorableMessageMetaData; import org.apache.qpid.server.util.ByteBufferOutputStream; public class TestMessageMetaData implements StorableMessageMetaData @@ -72,7 +71,7 @@ public class TestMessageMetaData implements StorableMessageMetaData } @Override - public int writeToBuffer(int offsetInMetaData, ByteBuffer dest) + public int writeToBuffer(ByteBuffer dest) { int oldPosition = dest.position(); try diff --git a/qpid/java/broker-core/src/test/java/org/apache/qpid/server/subscription/SubscriptionListTest.java b/qpid/java/broker-core/src/test/java/org/apache/qpid/server/subscription/SubscriptionListTest.java deleted file mode 100644 index cd5b178464..0000000000 --- a/qpid/java/broker-core/src/test/java/org/apache/qpid/server/subscription/SubscriptionListTest.java +++ /dev/null @@ -1,429 +0,0 @@ -/* - * - * 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.subscription; - -import org.apache.qpid.server.subscription.SubscriptionList.SubscriptionNode; -import org.apache.qpid.server.subscription.SubscriptionList.SubscriptionNodeIterator; -import org.apache.qpid.test.utils.QpidTestCase; - -public class SubscriptionListTest extends QpidTestCase -{ - private SubscriptionList _subList; - private MockSubscription _sub1; - private MockSubscription _sub2; - private MockSubscription _sub3; - private SubscriptionNode _node; - - protected void setUp() - { - _subList = new SubscriptionList(); - - _sub1 = new MockSubscription(); - _sub2 = new MockSubscription(); - _sub3 = new MockSubscription(); - - _subList.add(_sub1); - _subList.add(_sub2); - _subList.add(_sub3); - - _node = _subList.getHead(); - } - - /** - * Test that if the first (non-head) node in the list is deleted (but is still present), - * it is not returned when searching through the list for the next viable node, and the - * subsequent viable node is returned instead. - */ - public void testFindNextSkipsFirstDeletedNode() - { - assertTrue("Deleting subscription node should have succeeded", - getNodeForSubscription(_subList, _sub1).delete()); - - assertNotNull("Returned node should not be null", _node = _node.findNext()); - assertEquals("Should have returned node for 2nd subscription", _sub2, _node.getSubscription()); - - assertNotNull("Returned node should not be null", _node = _node.findNext()); - assertEquals("Should have returned node for 3rd subscription", _sub3, _node.getSubscription()); - } - - /** - * Test that if a central node in the list is deleted (but is still present), - * it is not returned when searching through the list for the next viable node, - * and the subsequent viable node is returned instead. - */ - public void testFindNextSkipsCentralDeletedNode() - { - assertNotNull("Returned node should not be null", _node = _node.findNext()); - - assertTrue("Deleting subscription node should have succeeded", - getNodeForSubscription(_subList, _sub2).delete()); - - assertNotNull("Returned node should not be null", _node = _node.findNext()); - assertEquals("Should have returned node for 3rd subscription", _sub3, _node.getSubscription()); - } - - /** - * Test that if the last node in the list is deleted (but is still present), - * it is not returned when searching through the list for the next viable node, - * and null is returned instead. - */ - public void testFindNextSkipsLastDeletedNode() - { - assertNotNull("Returned node should not be null", _node = _node.findNext()); - assertEquals("Should have returned node for 1st subscription", _sub1, _node.getSubscription()); - - assertNotNull("Returned node should not be null", _node = _node.findNext()); - assertEquals("Should have returned node for 2nd subscription", _sub2, _node.getSubscription()); - - assertTrue("Deleting subscription node should have succeeded", - getNodeForSubscription(_subList, _sub3).delete()); - - assertNull("Returned node should be null", _node = _node.findNext()); - } - - /** - * Test that if multiple nodes in the list are deleted (but still present), they - * are not returned when searching through the list for the next viable node, - * and the subsequent viable node is returned instead. - */ - public void testFindNextSkipsMultipleDeletedNode() - { - assertTrue("Deleting subscription node should have succeeded", - getNodeForSubscription(_subList, _sub1).delete()); - assertTrue("Deleting subscription node should have succeeded", - getNodeForSubscription(_subList, _sub2).delete()); - - assertNotNull("Returned node should not be null", _node = _node.findNext()); - assertEquals("Should have returned node for 3rd subscription", _sub3, _node.getSubscription()); - } - - /** - * Test that if a node in the list is marked 'deleted' it is still present in the list - * until actually removed. counter-test to verify above testing of getNext() method. - */ - public void testDeletedNodeStillPresent() - { - assertTrue("Deleting subscription node should have succeeded", - getNodeForSubscription(_subList, _sub1).delete()); - - assertNotNull("Node marked deleted should still be present", getNodeForSubscription(_subList, _sub1)); - assertEquals("All 3 nodes are still expected to be present", 3, countNodes(_subList)); - } - - /** - * Traverses the list nodes in a non-mutating fashion, returning the first node which matches the given - * Subscription, or null if none is found. - */ - private SubscriptionNode getNodeForSubscription(final SubscriptionList list, final Subscription sub) - { - SubscriptionNode node = list.getHead(); - while (node != null && node.getSubscription() != sub) - { - node = node.nextNode(); - } - - return node; - } - - /** - * Counts the number of (non-head) nodes in the list. - */ - private int countNodes(final SubscriptionList list) - { - SubscriptionNode node = list.getHead(); - int count; - for(count = -1; node != null; count++) - { - node = node.nextNode(); - } - - return count; - } - - /** - * Tests that the head is returned as expected, and isn't the node for the first subscription. - */ - public void testGetHead() - { - assertNotNull("List head should be non null", _node); - assertNotSame("Head should not be node for first subscription", - _node, getNodeForSubscription(_subList, _sub1)); - } - - /** - * Tests that the size is returned correctly in the face of additions and removals. - */ - public void testGetSize() - { - SubscriptionList subList = new SubscriptionList(); - - assertEquals("Unexpected size result", 0, subList.size()); - - Subscription sub1 = new MockSubscription(); - Subscription sub2 = new MockSubscription(); - Subscription sub3 = new MockSubscription(); - - subList.add(sub1); - assertEquals("Unexpected size result", 1, subList.size()); - - subList.add(sub2); - assertEquals("Unexpected size result", 2, subList.size()); - - subList.add(sub3); - assertEquals("Unexpected size result", 3, subList.size()); - - assertTrue("Removing subscription from list should have succeeded", subList.remove(sub1)); - assertEquals("Unexpected size result", 2, subList.size()); - - assertTrue("Removing subscription from list should have succeeded", subList.remove(sub2)); - assertEquals("Unexpected size result", 1, subList.size()); - - assertTrue("Removing subscription from list should have succeeded", subList.remove(sub3)); - assertEquals("Unexpected size result", 0, subList.size()); - } - - /** - * Test that if the first (non-head) node in the list is removed it is no longer - * present in the node structure of the list at all. - */ - public void testRemoveFirstNode() - { - assertNotNull("Should have been a node present for the subscription", getNodeForSubscription(_subList, _sub1)); - assertTrue("Removing subscription node should have succeeded", _subList.remove(_sub1)); - assertNull("Should not have been a node present for the removed subscription", getNodeForSubscription(_subList, _sub1)); - assertEquals("Unexpected number of nodes", 2, countNodes(_subList)); - assertNotNull("Should have been a node present for the subscription", getNodeForSubscription(_subList, _sub2)); - assertNotNull("Should have been a node present for the subscription", getNodeForSubscription(_subList, _sub3)); - } - - /** - * Test that if a central node in the list is removed it is no longer - * present in the node structure of the list at all. - */ - public void testRemoveCentralNode() - { - assertNotNull("Should have been a node present for the subscription", getNodeForSubscription(_subList, _sub2)); - assertTrue("Removing subscription node should have succeeded", _subList.remove(_sub2)); - assertNull("Should not have been a node present for the removed subscription", getNodeForSubscription(_subList, _sub2)); - assertEquals("Unexpected number of nodes", 2, countNodes(_subList)); - assertNotNull("Should have been a node present for the subscription", getNodeForSubscription(_subList, _sub1)); - assertNotNull("Should have been a node present for the subscription", getNodeForSubscription(_subList, _sub3)); - } - - /** - * Test that if the subscription contained in the last node of the list is removed - * it is no longer present in the node structure of the list at all. However, - * as the last node in the structure can't actually be removed a dummy will instead - * be present. - */ - public void testRemoveLastNode() - { - assertNotNull("Should have been a node present for the subscription", getNodeForSubscription(_subList, _sub3)); - assertTrue("Removing subscription node should have succeeded", _subList.remove(_sub3)); - assertNull("Should not have been a node present for the removed subscription", getNodeForSubscription(_subList, _sub3)); - - //We actually expect 3 nodes to remain this time, because the last node cant be removed for thread safety reasons, - //however a dummy final node can be used as substitute to allow removal of the subscription node. - assertEquals("Unexpected number of nodes", 2 + 1, countNodes(_subList)); - assertNotNull("Should have been a node present for the subscription", getNodeForSubscription(_subList, _sub1)); - assertNotNull("Should have been a node present for the subscription", getNodeForSubscription(_subList, _sub2)); - } - - /** - * Test that if the subscription not contained in the list is requested to be removed - * that the removal fails - */ - public void testRemoveNonexistentNode() - { - Subscription sub4 = new MockSubscription(); - assertNull("Should not have been a node present for the subscription", getNodeForSubscription(_subList, sub4)); - assertFalse("Removing subscription node should not have succeeded", _subList.remove(sub4)); - assertEquals("Unexpected number of nodes", 3, countNodes(_subList)); - } - - /** - * Test that if a subscription node which occurs later in the main list than the marked node is - * removed from the list after the marked node is also removed, then the marker node doesn't - * serve to retain the subsequent nodes in the list structure (and thus memory) despite their - * removal. - */ - public void testDeletedMarkedNodeDoesntLeakSubsequentlyDeletedNodes() - { - //get the nodes out the list for the 1st and 3rd subscriptions - SubscriptionNode sub1Node = getNodeForSubscription(_subList, _sub1); - assertNotNull("Should have been a node present for the subscription", sub1Node); - SubscriptionNode sub3Node = getNodeForSubscription(_subList, _sub3); - assertNotNull("Should have been a node present for the subscription", sub3Node); - - //mark the first subscription node - assertTrue("should have succeeded in updating the marked node", - _subList.updateMarkedNode(_subList.getMarkedNode(), sub1Node)); - - //remove the 1st subscription from the list - assertTrue("Removing subscription node should have succeeded", _subList.remove(_sub1)); - //verify the 1st subscription is no longer the marker node (replaced by a dummy), or in the main list structure - assertNotSame("Unexpected marker node", sub1Node, _subList.getMarkedNode()); - assertNull("Should not have been a node present in the list structure for the marked-but-removed sub1 node", - getNodeForSubscription(_subList, _sub1)); - - //remove the 2nd subscription from the list - assertTrue("Removing subscription node should have succeeded", _subList.remove(_sub2)); - - //verify the marker node isn't leaking subsequently removed nodes, by ensuring the very next node - //in its list structure is now the 3rd subscription (since the 2nd was removed too) - assertEquals("Unexpected next node", sub3Node, _subList.getMarkedNode().nextNode()); - - //remove the 3rd and final/tail subscription - assertTrue("Removing subscription node should have succeeded", _subList.remove(_sub3)); - - //verify the marker node isn't leaking subsequently removed nodes, by ensuring the very next node - //in its list structure is now the dummy tail (since the 3rd subscription was removed, and a dummy - //tail was inserted) and NOT the 3rd sub node. - assertNotSame("Unexpected next node", sub3Node, _subList.getMarkedNode().nextNode()); - assertTrue("Unexpected next node", _subList.getMarkedNode().nextNode().isDeleted()); - assertNull("Next non-deleted node from the marker should now be the list end, i.e. null", _subList.getMarkedNode().findNext()); - } - - /** - * Test that the marked node 'findNext' behaviour is as expected after a subscription is added - * to the list following the tail subscription node being removed while it is the marked node. - * That is, that the new subscriptions node is returned by getMarkedNode().findNext(). - */ - public void testMarkedNodeFindsNewSubscriptionAfterRemovingTailWhilstMarked() - { - //get the node out the list for the 3rd subscription - SubscriptionNode sub3Node = getNodeForSubscription(_subList, _sub3); - assertNotNull("Should have been a node present for the subscription", sub3Node); - - //mark the 3rd subscription node - assertTrue("should have succeeded in updating the marked node", - _subList.updateMarkedNode(_subList.getMarkedNode(), sub3Node)); - - //verify calling findNext on the marked node returns null, i.e. the end of the list has been reached - assertEquals("Unexpected node after marked node", null, _subList.getMarkedNode().findNext()); - - //remove the 3rd(marked) subscription from the list - assertTrue("Removing subscription node should have succeeded", _subList.remove(_sub3)); - - //add a new 4th subscription to the list - Subscription sub4 = new MockSubscription(); - _subList.add(sub4); - - //get the node out the list for the 4th subscription - SubscriptionNode sub4Node = getNodeForSubscription(_subList, sub4); - assertNotNull("Should have been a node present for the subscription", sub4Node); - - //verify the marked node (which is now a dummy substitute for the 3rd subscription) returns - //the 4th subscriptions node as the next non-deleted node. - assertEquals("Unexpected next node", sub4Node, _subList.getMarkedNode().findNext()); - } - - /** - * Test that setting the marked node to null doesn't cause problems during remove operations - */ - public void testRemoveWithNullMarkedNode() - { - //set the marker to null - assertTrue("should have succeeded in updating the marked node", - _subList.updateMarkedNode(_subList.getMarkedNode(), null)); - - //remove the 1st subscription from the main list - assertTrue("Removing subscription node should have succeeded", _subList.remove(_sub1)); - - //verify the 1st subscription is no longer in the main list structure - assertNull("Should not have been a node present in the main list structure for sub1", - getNodeForSubscription(_subList, _sub1)); - assertEquals("Unexpected number of nodes", 2, countNodes(_subList)); - } - - /** - * Tests that after the first (non-head) node of the list is marked deleted but has not - * yet been removed, the iterator still skips it. - */ - public void testIteratorSkipsFirstDeletedNode() - { - //'delete' but don't remove the node for the 1st subscription - assertTrue("Deleting subscription node should have succeeded", - getNodeForSubscription(_subList, _sub1).delete()); - assertNotNull("Should still have been a node present for the deleted subscription", - getNodeForSubscription(_subList, _sub1)); - - SubscriptionNodeIterator iter = _subList.iterator(); - - //verify the iterator returns the 2nd subscriptions node - assertTrue("Iterator should have been able to advance", iter.advance()); - assertEquals("Iterator returned unexpected SubscriptionNode", _sub2, iter.getNode().getSubscription()); - - //verify the iterator returns the 3rd subscriptions node and not the 2nd. - assertTrue("Iterator should have been able to advance", iter.advance()); - assertEquals("Iterator returned unexpected SubscriptionNode", _sub3, iter.getNode().getSubscription()); - } - - /** - * Tests that after a central node of the list is marked deleted but has not yet been removed, - * the iterator still skips it. - */ - public void testIteratorSkipsCentralDeletedNode() - { - //'delete' but don't remove the node for the 2nd subscription - assertTrue("Deleting subscription node should have succeeded", - getNodeForSubscription(_subList, _sub2).delete()); - assertNotNull("Should still have been a node present for the deleted subscription", - getNodeForSubscription(_subList, _sub2)); - - SubscriptionNodeIterator iter = _subList.iterator(); - - //verify the iterator returns the 1st subscriptions node - assertTrue("Iterator should have been able to advance", iter.advance()); - assertEquals("Iterator returned unexpected SubscriptionNode", _sub1, iter.getNode().getSubscription()); - - //verify the iterator returns the 3rd subscriptions node and not the 2nd. - assertTrue("Iterator should have been able to advance", iter.advance()); - assertEquals("Iterator returned unexpected SubscriptionNode", _sub3, iter.getNode().getSubscription()); - } - - /** - * Tests that after the last node of the list is marked deleted but has not yet been removed, - * the iterator still skips it. - */ - public void testIteratorSkipsDeletedFinalNode() - { - //'delete' but don't remove the node for the 3rd subscription - assertTrue("Deleting subscription node should have succeeded", - getNodeForSubscription(_subList, _sub3).delete()); - assertNotNull("Should still have been a node present for the deleted 3rd subscription", - getNodeForSubscription(_subList, _sub3)); - - SubscriptionNodeIterator iter = _subList.iterator(); - - //verify the iterator returns the 1st subscriptions node - assertTrue("Iterator should have been able to advance", iter.advance()); - assertEquals("Iterator returned unexpected SubscriptionNode", _sub1, iter.getNode().getSubscription()); - - //verify the iterator returns the 2nd subscriptions node - assertTrue("Iterator should have been able to advance", iter.advance()); - assertEquals("Iterator returned unexpected SubscriptionNode", _sub2, iter.getNode().getSubscription()); - - //verify the iterator can no longer advance and does not return a subscription node - assertFalse("Iterator should not have been able to advance", iter.advance()); - assertEquals("Iterator returned unexpected SubscriptionNode", null, iter.getNode()); - } -} diff --git a/qpid/java/broker-core/src/test/java/org/apache/qpid/server/txn/AutoCommitTransactionTest.java b/qpid/java/broker-core/src/test/java/org/apache/qpid/server/txn/AutoCommitTransactionTest.java index 3c66a4c94b..11b9bbe1b4 100644 --- a/qpid/java/broker-core/src/test/java/org/apache/qpid/server/txn/AutoCommitTransactionTest.java +++ b/qpid/java/broker-core/src/test/java/org/apache/qpid/server/txn/AutoCommitTransactionTest.java @@ -20,6 +20,7 @@ */ package org.apache.qpid.server.txn; +import org.apache.qpid.server.message.MessageInstance; import org.apache.qpid.server.message.ServerMessage; import org.apache.qpid.server.queue.AMQQueue; import org.apache.qpid.server.queue.MockAMQQueue; @@ -47,7 +48,7 @@ public class AutoCommitTransactionTest extends QpidTestCase private MessageStore _transactionLog; private AMQQueue _queue; private List<AMQQueue> _queues; - private Collection<QueueEntry> _queueEntries; + private Collection<MessageInstance> _queueEntries; private ServerMessage _message; private MockAction _action; private MockStoreTransaction _storeTransaction; @@ -373,9 +374,9 @@ public class AutoCommitTransactionTest extends QpidTestCase assertFalse("Rollback action must be fired", _action.isRollbackActionFired()); } - private Collection<QueueEntry> createTestQueueEntries(boolean[] queueDurableFlags, boolean[] messagePersistentFlags) + private Collection<MessageInstance> createTestQueueEntries(boolean[] queueDurableFlags, boolean[] messagePersistentFlags) { - Collection<QueueEntry> queueEntries = new ArrayList<QueueEntry>(); + Collection<MessageInstance> queueEntries = new ArrayList<MessageInstance>(); assertTrue("Boolean arrays must be the same length", queueDurableFlags.length == messagePersistentFlags.length); diff --git a/qpid/java/broker-core/src/test/java/org/apache/qpid/server/txn/LocalTransactionTest.java b/qpid/java/broker-core/src/test/java/org/apache/qpid/server/txn/LocalTransactionTest.java index f3f5e00346..80e794e0ff 100644 --- a/qpid/java/broker-core/src/test/java/org/apache/qpid/server/txn/LocalTransactionTest.java +++ b/qpid/java/broker-core/src/test/java/org/apache/qpid/server/txn/LocalTransactionTest.java @@ -20,6 +20,7 @@ */ package org.apache.qpid.server.txn; +import org.apache.qpid.server.message.MessageInstance; import org.apache.qpid.server.message.ServerMessage; import org.apache.qpid.server.queue.AMQQueue; import org.apache.qpid.server.queue.MockAMQQueue; @@ -46,7 +47,7 @@ public class LocalTransactionTest extends QpidTestCase private AMQQueue _queue; private List<AMQQueue> _queues; - private Collection<QueueEntry> _queueEntries; + private Collection<MessageInstance> _queueEntries; private ServerMessage _message; private MockAction _action1; private MockAction _action2; @@ -597,9 +598,9 @@ public class LocalTransactionTest extends QpidTestCase assertEquals("Transaction update time should be reset after rollback", 0, _transaction.getTransactionUpdateTime()); } - private Collection<QueueEntry> createTestQueueEntries(boolean[] queueDurableFlags, boolean[] messagePersistentFlags) + private Collection<MessageInstance> createTestQueueEntries(boolean[] queueDurableFlags, boolean[] messagePersistentFlags) { - Collection<QueueEntry> queueEntries = new ArrayList<QueueEntry>(); + Collection<MessageInstance> queueEntries = new ArrayList<MessageInstance>(); assertTrue("Boolean arrays must be the same length", queueDurableFlags.length == messagePersistentFlags.length); diff --git a/qpid/java/broker-core/src/test/java/org/apache/qpid/server/virtualhost/MockVirtualHost.java b/qpid/java/broker-core/src/test/java/org/apache/qpid/server/virtualhost/MockVirtualHost.java index 1ca7ff1b65..832b89c81a 100644 --- a/qpid/java/broker-core/src/test/java/org/apache/qpid/server/virtualhost/MockVirtualHost.java +++ b/qpid/java/broker-core/src/test/java/org/apache/qpid/server/virtualhost/MockVirtualHost.java @@ -27,6 +27,8 @@ import org.apache.qpid.AMQException; import org.apache.qpid.server.configuration.VirtualHostConfiguration; import org.apache.qpid.server.connection.IConnectionRegistry; import org.apache.qpid.server.exchange.Exchange; +import org.apache.qpid.server.message.MessageDestination; +import org.apache.qpid.server.message.MessageSource; import org.apache.qpid.server.plugin.ExchangeType; import org.apache.qpid.server.protocol.LinkRegistry; import org.apache.qpid.server.queue.AMQQueue; @@ -127,6 +129,12 @@ public class MockVirtualHost implements VirtualHost } @Override + public MessageSource getMessageSource(final String name) + { + return null; + } + + @Override public AMQQueue getQueue(UUID id) { return null; @@ -174,6 +182,12 @@ public class MockVirtualHost implements VirtualHost } @Override + public MessageDestination getMessageDestination(final String name) + { + return null; + } + + @Override public Exchange getExchange(String name) { return null; |
