QPID-7633: Switch _blockedChannels to a set backed by a CHM removing need for the SkipList and Comparable
diff --git a/broker-core/src/main/java/org/apache/qpid/server/protocol/AMQSessionModel.java b/broker-core/src/main/java/org/apache/qpid/server/protocol/AMQSessionModel.java
index bb2462c..3d24785 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/protocol/AMQSessionModel.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/protocol/AMQSessionModel.java
@@ -22,23 +22,19 @@
import java.util.Collection;
import java.util.UUID;
-import java.util.concurrent.ConcurrentSkipListSet;
import org.apache.qpid.server.consumer.ConsumerTarget;
import org.apache.qpid.server.logging.LogSubject;
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.transport.AMQPConnection;
import org.apache.qpid.server.util.Deletable;
import org.apache.qpid.transport.network.Ticker;
/**
* Session model interface.
- * Extends {@link Comparable} to allow objects to be inserted into a {@link ConcurrentSkipListSet}
- * when monitoring the blocking and blocking of queues/sessions in {@link Queue}.
*/
-public interface AMQSessionModel<T extends AMQSessionModel<T,X>, X extends ConsumerTarget<X>> extends Comparable<AMQSessionModel>, Deletable<T>
+public interface AMQSessionModel<T extends AMQSessionModel<T,X>, X extends ConsumerTarget<X>> extends Deletable<T>
{
UUID getId();
diff --git a/broker-core/src/main/java/org/apache/qpid/server/queue/AbstractQueue.java b/broker-core/src/main/java/org/apache/qpid/server/queue/AbstractQueue.java
index 596bbb7..5525a25 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/queue/AbstractQueue.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/queue/AbstractQueue.java
@@ -47,7 +47,6 @@
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.ConcurrentSkipListSet;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -209,7 +208,7 @@
private AtomicBoolean _stopped = new AtomicBoolean(false);
- private final Set<AMQSessionModel<?,?>> _blockedChannels = new ConcurrentSkipListSet<>();
+ private final Set<AMQSessionModel<?,?>> _blockedChannels = Collections.newSetFromMap(new ConcurrentHashMap<AMQSessionModel<?, ?>, Boolean>());
private final AtomicBoolean _deleted = new AtomicBoolean(false);
private final SettableFuture<Integer> _deleteFuture = SettableFuture.create();
diff --git a/broker-core/src/test/java/org/apache/qpid/server/consumer/TestConsumerTarget.java b/broker-core/src/test/java/org/apache/qpid/server/consumer/TestConsumerTarget.java
index 74f7605..b755e3b 100644
--- a/broker-core/src/test/java/org/apache/qpid/server/consumer/TestConsumerTarget.java
+++ b/broker-core/src/test/java/org/apache/qpid/server/consumer/TestConsumerTarget.java
@@ -412,11 +412,6 @@
_connection.notifyWork(this);
}
- @Override
- public int compareTo(final AMQSessionModel o)
- {
- return 0;
- }
}
}
diff --git a/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSession.java b/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSession.java
index 5c6e6d4..3717211 100644
--- a/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSession.java
+++ b/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSession.java
@@ -1071,11 +1071,6 @@
return _modelObject.getMaxUncommittedInMemorySize();
}
- public int compareTo(AMQSessionModel o)
- {
- return getId().compareTo(o.getId());
- }
-
@Override
protected void sendSessionAttached(final byte[] name, final Option... options)
{
diff --git a/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/Session_0_10.java b/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/Session_0_10.java
index 351f276..df6766c 100644
--- a/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/Session_0_10.java
+++ b/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/Session_0_10.java
@@ -104,12 +104,6 @@
}
@Override
- public int compareTo(final AMQSessionModel o)
- {
- return getId().compareTo(o.getId());
- }
-
- @Override
public boolean getBlocking()
{
return _serverSession.getBlocking();
diff --git a/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java b/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java
index ed6585e..e9f9602 100644
--- a/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java
+++ b/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java
@@ -1318,12 +1318,6 @@
return message;
}
- @Override
- public int compareTo(AMQSessionModel o)
- {
- return getId().compareTo(o.getId());
- }
-
public Subject getSubject()
{
return _subject;
diff --git a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java
index a2058ad..f9f5616 100644
--- a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java
+++ b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java
@@ -53,8 +53,6 @@
import org.apache.qpid.bytebuffer.QpidByteBuffer;
import org.apache.qpid.common.AMQPFilterTypes;
import org.apache.qpid.exchange.ExchangeDefaults;
-import org.apache.qpid.server.connection.SessionPrincipal;
-import org.apache.qpid.server.consumer.ScheduledConsumerTargetSet;
import org.apache.qpid.server.logging.LogMessage;
import org.apache.qpid.server.logging.LogSubject;
import org.apache.qpid.server.logging.messages.ChannelMessages;
@@ -64,8 +62,6 @@
import org.apache.qpid.server.message.MessageSource;
import org.apache.qpid.server.model.AbstractConfiguredObject;
import org.apache.qpid.server.model.Binding;
-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.Exchange;
import org.apache.qpid.server.model.ExclusivityPolicy;
@@ -124,7 +120,6 @@
import org.apache.qpid.server.util.Action;
import org.apache.qpid.server.util.ConnectionScopedRuntimeException;
import org.apache.qpid.server.virtualhost.QueueManagingVirtualHost;
-import org.apache.qpid.transport.network.Ticker;
public class Session_1_0 extends AbstractAMQPSession<Session_1_0, ConsumerTarget_1_0>
implements AMQSessionModel<Session_1_0, ConsumerTarget_1_0>, LogSubject
@@ -1833,12 +1828,6 @@
_sendingChannel) + "] ";
}
- @Override
- public int compareTo(AMQSessionModel o)
- {
- return getId().compareTo(o.getId());
- }
-
public AMQPConnection_1_0 getConnection()
{
return _connection;
