You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by kw...@apache.org on 2017/01/24 13:46:29 UTC
svn commit: r1780078 - in /qpid/java/trunk:
broker-core/src/main/java/org/apache/qpid/server/protocol/
broker-core/src/test/java/org/apache/qpid/server/consumer/
broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/
bro...
Author: kwall
Date: Tue Jan 24 13:46:29 2017
New Revision: 1780078
URL: http://svn.apache.org/viewvc?rev=1780078&view=rev
Log:
QPID-7633: Remove ConsumerListener
Modified:
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/protocol/AMQSessionModel.java
qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/consumer/TestConsumerTarget.java
qpid/java/trunk/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSession.java
qpid/java/trunk/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/Session_0_10.java
qpid/java/trunk/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java
qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java
Modified: qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/protocol/AMQSessionModel.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/protocol/AMQSessionModel.java?rev=1780078&r1=1780077&r2=1780078&view=diff
==============================================================================
--- qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/protocol/AMQSessionModel.java (original)
+++ qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/protocol/AMQSessionModel.java Tue Jan 24 13:46:29 2017
@@ -74,14 +74,6 @@ public interface AMQSessionModel<T exten
Collection<Consumer<?,X>> getConsumers();
- void addConsumerListener(ConsumerListener listener);
-
- void removeConsumerListener(ConsumerListener listener);
-
- void setModelObject(Session<?> session);
-
- Session<?> getModelObject();
-
/**
* Return the time the current transaction started.
*
Modified: qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/consumer/TestConsumerTarget.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/consumer/TestConsumerTarget.java?rev=1780078&r1=1780077&r2=1780078&view=diff
==============================================================================
--- qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/consumer/TestConsumerTarget.java (original)
+++ qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/consumer/TestConsumerTarget.java Tue Jan 24 13:46:29 2017
@@ -44,7 +44,6 @@ import org.apache.qpid.server.model.Cons
import org.apache.qpid.server.model.Queue;
import org.apache.qpid.server.model.Session;
import org.apache.qpid.server.protocol.AMQSessionModel;
-import org.apache.qpid.server.protocol.ConsumerListener;
import org.apache.qpid.server.message.MessageContainer;
import org.apache.qpid.server.transport.AMQPConnection;
import org.apache.qpid.server.util.Action;
@@ -359,24 +358,6 @@ public class TestConsumerTarget implemen
}
@Override
- public void addConsumerListener(final ConsumerListener listener)
- {
-
- }
-
- @Override
- public void setModelObject(final Session session)
- {
- _modelObject = session;
- }
-
- @Override
- public Session<?> getModelObject()
- {
- return _modelObject;
- }
-
- @Override
public long getTransactionStartTimeLong()
{
return 0;
@@ -389,12 +370,6 @@ public class TestConsumerTarget implemen
}
@Override
- public void removeConsumerListener(final ConsumerListener listener)
- {
-
- }
-
- @Override
public void addDeleteTask(final Action task)
{
Modified: qpid/java/trunk/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSession.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSession.java?rev=1780078&r1=1780077&r2=1780078&view=diff
==============================================================================
--- qpid/java/trunk/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSession.java (original)
+++ qpid/java/trunk/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSession.java Tue Jan 24 13:46:29 2017
@@ -62,16 +62,12 @@ import org.apache.qpid.server.message.Me
import org.apache.qpid.server.message.MessageInstance;
import org.apache.qpid.server.message.MessageInstanceConsumer;
import org.apache.qpid.server.message.RoutingResult;
-import org.apache.qpid.server.model.AbstractConfigurationChangeListener;
import org.apache.qpid.server.model.Broker;
-import org.apache.qpid.server.model.ConfigurationChangeListener;
-import org.apache.qpid.server.model.ConfiguredObject;
import org.apache.qpid.server.model.Consumer;
import org.apache.qpid.server.model.NamedAddressSpace;
import org.apache.qpid.server.model.Queue;
import org.apache.qpid.server.protocol.AMQSessionModel;
import org.apache.qpid.server.protocol.CapacityChecker;
-import org.apache.qpid.server.protocol.ConsumerListener;
import org.apache.qpid.server.store.MessageStore;
import org.apache.qpid.server.store.StoreException;
import org.apache.qpid.server.store.StoredMessage;
@@ -103,54 +99,48 @@ public class ServerSession extends Sessi
private static final int PRODUCER_CREDIT_TOPUP_THRESHOLD = 1 << 30;
private static final int UNFINISHED_COMMAND_QUEUE_THRESHOLD = 500;
- public Subject getSubject()
- {
- return _modelObject.getSubject();
- }
-
- private long _createTime = System.currentTimeMillis();
-
private final Set<Object> _blockingEntities = Collections.synchronizedSet(new HashSet<Object>());
private final AtomicBoolean _blocking = new AtomicBoolean(false);
private final AtomicInteger _outstandingCredit = new AtomicInteger(UNLIMITED_CREDIT);
private final CheckCapacityAction _checkCapacityAction = new CheckCapacityAction();
- private final CopyOnWriteArrayList<ConsumerListener> _consumerListeners = new CopyOnWriteArrayList<ConsumerListener>();
- private final ConfigurationChangeListener _consumerClosedListener = new ConsumerClosedListener();
private Session_0_10 _modelObject;
private long _blockTime;
private long _blockingTimeout;
private boolean _wireBlockingState;
- public static interface MessageDispositionChangeListener
+ public interface MessageDispositionChangeListener
{
- public void onAccept();
-
- public void onRelease(boolean setRedelivered);
+ void onAccept();
- public void onReject();
+ void onRelease(boolean setRedelivered);
- public boolean acquire();
+ void onReject();
+ boolean acquire();
+ }
+ private interface MessageDispositionAction
+ {
+ void performAction(MessageDispositionChangeListener listener);
}
private final SortedMap<Integer, MessageDispositionChangeListener> _messageDispositionListenerMap =
new ConcurrentSkipListMap<Integer, MessageDispositionChangeListener>();
private ServerTransaction _transaction;
-
private final AtomicLong _txnStarts = new AtomicLong(0);
private final AtomicLong _txnCommits = new AtomicLong(0);
private final AtomicLong _txnRejects = new AtomicLong(0);
- private final AtomicLong _txnCount = new AtomicLong(0);
+ private final AtomicLong _txnCount = new AtomicLong(0);
private Map<String, ConsumerTarget_0_10> _subscriptions = new ConcurrentHashMap<String, ConsumerTarget_0_10>();
+
private final CopyOnWriteArrayList<Consumer<?, ConsumerTarget_0_10>> _consumers = new CopyOnWriteArrayList<>();
private AtomicReference<LogMessage> _forcedCloseLogMessage = new AtomicReference<LogMessage>();
-
private volatile long _uncommittedMessageSize;
+
private final List<StoredMessage<MessageMetaData_0_10>> _uncommittedMessages = new ArrayList<>();
public ServerSession(Connection connection, SessionDelegate delegate, Binary name, long expiry)
@@ -163,6 +153,11 @@ public class ServerSession extends Sessi
_blockingTimeout = serverConnection.getBroker().getContextValue(Long.class, Broker.CHANNEL_FLOW_CONTROL_ENFORCEMENT_TIMEOUT);
}
+ public Subject getSubject()
+ {
+ return _modelObject.getSubject();
+ }
+
public AccessControlContext getAccessControllerContext()
{
return _modelObject.getAccessControllerContext();
@@ -259,6 +254,7 @@ public class ServerSession extends Sessi
_uncommittedMessages.clear();
}
+
private void incrementUncommittedMessageSize(final StoredMessage<MessageMetaData_0_10> handle)
{
if (isTransactional() && !(_transaction instanceof DistributedTransaction))
@@ -289,7 +285,6 @@ public class ServerSession extends Sessi
}
}
-
public void sendMessage(MessageTransfer xfr,
Runnable postIdSettingAction)
{
@@ -297,17 +292,12 @@ public class ServerSession extends Sessi
invoke(xfr, postIdSettingAction);
}
+
public void onMessageDispositionChange(MessageTransfer xfr, MessageDispositionChangeListener acceptListener)
{
_messageDispositionListenerMap.put(xfr.getId(), acceptListener);
}
-
- private static interface MessageDispositionAction
- {
- void performAction(MessageDispositionChangeListener listener);
- }
-
public void accept(RangeSet ranges)
{
dispositionChange(ranges, new MessageDispositionAction()
@@ -529,8 +519,6 @@ public class ServerSession extends Sessi
{
final Consumer<?,ConsumerTarget_0_10> consumer = (Consumer<?,ConsumerTarget_0_10>) messageInstanceConsumer;
_consumers.add(consumer);
- consumer.addChangeListener(_consumerClosedListener);
- consumerAdded(consumer);
}
}
@@ -1036,16 +1024,6 @@ public class ServerSession extends Sessi
return Collections.unmodifiableCollection(_consumers);
}
- public void addConsumerListener(final ConsumerListener listener)
- {
- _consumerListeners.add(listener);
- }
-
- public void removeConsumerListener(final ConsumerListener listener)
- {
- _consumerListeners.remove(listener);
- }
-
public void setModelObject(final Session_0_10 session)
{
_modelObject = session;
@@ -1082,22 +1060,6 @@ public class ServerSession extends Sessi
}
}
- private void consumerAdded(Consumer<?, ConsumerTarget_0_10> consumer)
- {
- for(ConsumerListener l : _consumerListeners)
- {
- l.consumerAdded(consumer);
- }
- }
-
- private void consumerRemoved(Consumer<?, ConsumerTarget_0_10> consumer)
- {
- for(ConsumerListener l : _consumerListeners)
- {
- l.consumerRemoved(consumer);
- }
- }
-
public void doTimeoutAction(final String reason)
{
getAMQPConnection().closeSessionAsync(_modelObject,
@@ -1132,16 +1094,4 @@ public class ServerSession extends Sessi
}
}
}
-
- private class ConsumerClosedListener extends AbstractConfigurationChangeListener
- {
- @Override
- public void stateChanged(final ConfiguredObject object, final org.apache.qpid.server.model.State oldState, final org.apache.qpid.server.model.State newState)
- {
- if(newState == org.apache.qpid.server.model.State.DELETED)
- {
- consumerRemoved((Consumer<?, ConsumerTarget_0_10>)object);
- }
- }
- }
}
Modified: qpid/java/trunk/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/Session_0_10.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/Session_0_10.java?rev=1780078&r1=1780077&r2=1780078&view=diff
==============================================================================
--- qpid/java/trunk/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/Session_0_10.java (original)
+++ qpid/java/trunk/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/Session_0_10.java Tue Jan 24 13:46:29 2017
@@ -30,9 +30,7 @@ import org.apache.qpid.server.logging.Lo
import org.apache.qpid.server.model.Connection;
import org.apache.qpid.server.model.Consumer;
import org.apache.qpid.server.model.Queue;
-import org.apache.qpid.server.model.Session;
import org.apache.qpid.server.protocol.AMQSessionModel;
-import org.apache.qpid.server.protocol.ConsumerListener;
import org.apache.qpid.server.protocol.PublishAuthorisationCache;
import org.apache.qpid.server.session.AbstractAMQPSession;
import org.apache.qpid.server.util.Action;
@@ -87,30 +85,6 @@ public class Session_0_10 extends Abstra
}
@Override
- public void addConsumerListener(final ConsumerListener listener)
- {
- _serverSession.addConsumerListener(listener);
- }
-
- @Override
- public void removeConsumerListener(final ConsumerListener listener)
- {
- _serverSession.removeConsumerListener(listener);
- }
-
- @Override
- public void setModelObject(final Session<?> session)
- {
- _serverSession.setModelObject(this);
- }
-
- @Override
- public Session<?> getModelObject()
- {
- return this;
- }
-
- @Override
public void transportStateChanged()
{
for(ConsumerTarget_0_10 consumerTarget : _serverSession.getSubscriptions())
@@ -205,17 +179,17 @@ public class Session_0_10 extends Abstra
return _subject;
}
- public AccessControlContext getAccessControllerContext()
+ AccessControlContext getAccessControllerContext()
{
return _accessControllerContext;
}
- public PublishAuthorisationCache getPublishAuthCache()
+ PublishAuthorisationCache getPublishAuthCache()
{
return _publishAuthCache;
}
- public List<Action<? super Session_0_10>> getTaskList()
+ List<Action<? super Session_0_10>> getTaskList()
{
return _taskList;
}
@@ -225,12 +199,12 @@ public class Session_0_10 extends Abstra
return _serverSession.isClosing();
}
- public ServerSession getServerSession()
+ ServerSession getServerSession()
{
return _serverSession;
}
- public long getMaxUncommittedInMemorySize()
+ long getMaxUncommittedInMemorySize()
{
return _maxUncommittedInMemorySize;
}
Modified: qpid/java/trunk/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java?rev=1780078&r1=1780077&r2=1780078&view=diff
==============================================================================
--- qpid/java/trunk/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java (original)
+++ qpid/java/trunk/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java Tue Jan 24 13:46:29 2017
@@ -80,7 +80,6 @@ import org.apache.qpid.server.message.Se
import org.apache.qpid.server.model.*;
import org.apache.qpid.server.protocol.AMQSessionModel;
import org.apache.qpid.server.protocol.CapacityChecker;
-import org.apache.qpid.server.protocol.ConsumerListener;
import org.apache.qpid.server.protocol.v0_8.UnacknowledgedMessageMap.Visitor;
import org.apache.qpid.server.queue.QueueArgumentsConverter;
import org.apache.qpid.server.session.AbstractAMQPSession;
@@ -180,14 +179,9 @@ public class AMQChannel extends Abstract
private final ClientDeliveryMethod _clientDeliveryMethod;
- private final UUID _id = UUID.randomUUID();
-
-
private final CapacityCheckAction _capacityCheckAction = new CapacityCheckAction();
private final ImmediateAction _immediateAction = new ImmediateAction();
private final CopyOnWriteArrayList<Consumer<?, ConsumerTarget_0_8>> _consumers = new CopyOnWriteArrayList<>();
- private final ConfigurationChangeListener _consumerClosedListener = new ConsumerClosedListener();
- private final CopyOnWriteArrayList<ConsumerListener> _consumerListeners = new CopyOnWriteArrayList<ConsumerListener>();
private Session<?> _modelObject;
private long _blockTime;
private long _blockingTimeout;
@@ -794,8 +788,6 @@ public class AMQChannel extends Abstract
if (sub instanceof Consumer<?, ?>)
{
final Consumer<?,ConsumerTarget_0_8> modelConsumer = (Consumer<?,ConsumerTarget_0_8>) sub;
- consumerAdded(modelConsumer);
- modelConsumer.addChangeListener(_consumerClosedListener);
_consumers.add(modelConsumer);
}
}
@@ -1855,58 +1847,6 @@ public class AMQChannel extends Abstract
return Collections.unmodifiableCollection(_consumers);
}
- private class ConsumerClosedListener extends AbstractConfigurationChangeListener
- {
- @Override
- public void stateChanged(final ConfiguredObject object, final State oldState, final State newState)
- {
- if(newState == State.DELETED)
- {
- consumerRemoved((Consumer<?,?>)object);
- }
- }
- }
-
- private void consumerAdded(final Consumer<?,?> consumer)
- {
- for(ConsumerListener l : _consumerListeners)
- {
- l.consumerAdded(consumer);
- }
- }
-
- private void consumerRemoved(final Consumer<?,?> consumer)
- {
- for(ConsumerListener l : _consumerListeners)
- {
- l.consumerRemoved(consumer);
- }
- }
-
- @Override
- public void addConsumerListener(ConsumerListener listener)
- {
- _consumerListeners.add(listener);
- }
-
- @Override
- public void removeConsumerListener(ConsumerListener listener)
- {
- _consumerListeners.remove(listener);
- }
-
- @Override
- public void setModelObject(final Session<?> session)
- {
- _modelObject = session;
- }
-
- @Override
- public Session<?> getModelObject()
- {
- return _modelObject;
- }
-
@Override
public long getTransactionStartTimeLong()
{
Modified: qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java?rev=1780078&r1=1780077&r2=1780078&view=diff
==============================================================================
--- qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java (original)
+++ qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java Tue Jan 24 13:46:29 2017
@@ -62,7 +62,6 @@ import org.apache.qpid.server.message.Me
import org.apache.qpid.server.message.MessageInstance;
import org.apache.qpid.server.message.MessageInstanceConsumer;
import org.apache.qpid.server.message.MessageSource;
-import org.apache.qpid.server.model.AbstractConfigurationChangeListener;
import org.apache.qpid.server.model.AbstractConfiguredObject;
import org.apache.qpid.server.model.Binding;
import org.apache.qpid.server.model.ConfigurationChangeListener;
@@ -77,7 +76,6 @@ import org.apache.qpid.server.model.Sess
import org.apache.qpid.server.model.State;
import org.apache.qpid.server.protocol.AMQSessionModel;
import org.apache.qpid.server.protocol.CapacityChecker;
-import org.apache.qpid.server.protocol.ConsumerListener;
import org.apache.qpid.server.protocol.LinkRegistry;
import org.apache.qpid.server.protocol.v1_0.codec.QpidByteBufferUtils;
import org.apache.qpid.server.protocol.v1_0.framing.OversizeFrameException;
@@ -125,11 +123,8 @@ import org.apache.qpid.server.txn.AutoCo
import org.apache.qpid.server.txn.ServerTransaction;
import org.apache.qpid.server.util.Action;
import org.apache.qpid.server.util.ConnectionScopedRuntimeException;
-<<<<<<< HEAD
import org.apache.qpid.server.virtualhost.QueueManagingVirtualHost;
import org.apache.qpid.transport.network.Ticker;
-=======
->>>>>>> QPID-7633: Pull up the processPendingIterator and associated methods.
public class Session_1_0 extends AbstractAMQPSession<Session_1_0, ConsumerTarget_1_0>
implements AMQSessionModel<Session_1_0, ConsumerTarget_1_0>, LogSubject
@@ -149,8 +144,6 @@ public class Session_1_0 extends Abstrac
private final CopyOnWriteArrayList<Consumer<?, ConsumerTarget_1_0>> _consumers = new CopyOnWriteArrayList<>();
- private final ConfigurationChangeListener _consumerClosedListener = new ConsumerClosedListener();
- private final CopyOnWriteArrayList<ConsumerListener> _consumerListeners = new CopyOnWriteArrayList<ConsumerListener>();
private Session<?> _modelObject = this;
private SessionState _sessionState;
@@ -1409,8 +1402,6 @@ public class Session_1_0 extends Abstrac
{
Consumer<?,ConsumerTarget_1_0> modelConsumer = (Consumer<?,ConsumerTarget_1_0>) consumer;
_consumers.add(modelConsumer);
- modelConsumer.addChangeListener(_consumerClosedListener);
- consumerAdded(modelConsumer);
}
}
@@ -1885,30 +1876,6 @@ public class Session_1_0 extends Abstrac
}
@Override
- public void addConsumerListener(final ConsumerListener listener)
- {
- _consumerListeners.add(listener);
- }
-
- @Override
- public void removeConsumerListener(final ConsumerListener listener)
- {
- _consumerListeners.remove(listener);
- }
-
- @Override
- public void setModelObject(final Session<?> session)
- {
- _modelObject = session;
- }
-
- @Override
- public Session<?> getModelObject()
- {
- return _modelObject;
- }
-
- @Override
public long getTransactionStartTimeLong()
{
return 0L;
@@ -1938,22 +1905,6 @@ public class Session_1_0 extends Abstrac
getAMQPConnection().closeSessionAsync(this, AMQPConnection.CloseReason.TRANSACTION_TIMEOUT, reason);
}
- private void consumerAdded(Consumer<?, ConsumerTarget_1_0> consumer)
- {
- for(ConsumerListener l : _consumerListeners)
- {
- l.consumerAdded(consumer);
- }
- }
-
- private void consumerRemoved(Consumer<?, ConsumerTarget_1_0> consumer)
- {
- for(ConsumerListener l : _consumerListeners)
- {
- l.consumerRemoved(consumer);
- }
- }
-
void incrementStartedTransactions()
{
_startedTransactions++;
@@ -1984,18 +1935,6 @@ public class Session_1_0 extends Abstrac
return _capacityCheckAction;
}
- private class ConsumerClosedListener extends AbstractConfigurationChangeListener
- {
- @Override
- public void stateChanged(final ConfiguredObject object, final org.apache.qpid.server.model.State oldState, final org.apache.qpid.server.model.State newState)
- {
- if(newState == org.apache.qpid.server.model.State.DELETED)
- {
- consumerRemoved((Consumer<?, ConsumerTarget_1_0>)object);
- }
- }
- }
-
@Override
public String toString()
{
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org