QPID-7633: Remove interface AMQSessionModel
diff --git a/broker-core/src/main/java/org/apache/qpid/server/connection/SessionPrincipal.java b/broker-core/src/main/java/org/apache/qpid/server/connection/SessionPrincipal.java
index 2bdace5..bd1acfb 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/connection/SessionPrincipal.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/connection/SessionPrincipal.java
@@ -20,23 +20,23 @@
*/
package org.apache.qpid.server.connection;
-import org.apache.qpid.server.protocol.AMQSessionModel;
-
import java.io.Serializable;
import java.security.Principal;
+import org.apache.qpid.server.session.AMQPSession;
+
public class SessionPrincipal implements Principal, Serializable
{
private static final long serialVersionUID = 1L;
- private final AMQSessionModel _session;
+ private final AMQPSession<?,?> _session;
- public SessionPrincipal(final AMQSessionModel session)
+ public SessionPrincipal(final AMQPSession<?, ?> session)
{
_session = session;
}
- public AMQSessionModel getSession()
+ public AMQPSession<?, ?> getSession()
{
return _session;
}
diff --git a/broker-core/src/main/java/org/apache/qpid/server/consumer/AbstractConsumerTarget.java b/broker-core/src/main/java/org/apache/qpid/server/consumer/AbstractConsumerTarget.java
index 8706bda..b913508 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/consumer/AbstractConsumerTarget.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/consumer/AbstractConsumerTarget.java
@@ -35,9 +35,9 @@
import org.apache.qpid.server.message.MessageInstance;
import org.apache.qpid.server.message.MessageInstanceConsumer;
import org.apache.qpid.server.model.Consumer;
-import org.apache.qpid.server.protocol.AMQSessionModel;
import org.apache.qpid.server.message.MessageContainer;
import org.apache.qpid.server.queue.SuspendedConsumerLoggingTicker;
+import org.apache.qpid.server.session.AMQPSession;
import org.apache.qpid.server.transport.AMQPConnection;
public abstract class AbstractConsumerTarget<T extends AbstractConsumerTarget<T>> implements ConsumerTarget<T>
@@ -172,8 +172,7 @@
private ListenableFuture<Void> doOnIoThreadAsync(final Runnable task)
{
- AMQSessionModel<?,T> sessionModel = getSessionModel();
- return sessionModel.getAMQPConnection().doOnIOThreadAsync(task);
+ return getSessionModel().getAMQPConnection().doOnIOThreadAsync(task);
}
private void consumerRemovedInternal(final MessageInstanceConsumer sub)
diff --git a/broker-core/src/main/java/org/apache/qpid/server/consumer/ConsumerTarget.java b/broker-core/src/main/java/org/apache/qpid/server/consumer/ConsumerTarget.java
index 076f5ec..b2b6320 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/consumer/ConsumerTarget.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/consumer/ConsumerTarget.java
@@ -25,7 +25,7 @@
import org.apache.qpid.server.message.MessageInstance;
import org.apache.qpid.server.message.MessageInstanceConsumer;
import org.apache.qpid.server.message.ServerMessage;
-import org.apache.qpid.server.protocol.AMQSessionModel;
+import org.apache.qpid.server.session.AMQPSession;
public interface ConsumerTarget<T extends ConsumerTarget<T>>
{
@@ -58,7 +58,7 @@
long getUnacknowledgedMessages();
- AMQSessionModel<?,T> getSessionModel();
+ AMQPSession<?,T> getSessionModel();
void send(final MessageInstanceConsumer<T> consumer, MessageInstance entry, boolean batch);
diff --git a/broker-core/src/main/java/org/apache/qpid/server/logging/AbstractMessageLogger.java b/broker-core/src/main/java/org/apache/qpid/server/logging/AbstractMessageLogger.java
index eaee6d1..846a617 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/logging/AbstractMessageLogger.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/logging/AbstractMessageLogger.java
@@ -36,10 +36,10 @@
import org.apache.qpid.server.connection.ConnectionPrincipal;
import org.apache.qpid.server.connection.SessionPrincipal;
import org.apache.qpid.server.logging.subjects.LogSubjectFormat;
-import org.apache.qpid.server.protocol.AMQSessionModel;
import org.apache.qpid.server.security.auth.AuthenticatedPrincipal;
import org.apache.qpid.server.security.auth.ManagementConnectionPrincipal;
import org.apache.qpid.server.security.auth.TaskPrincipal;
+import org.apache.qpid.server.session.AMQPSession;
import org.apache.qpid.server.transport.AMQPConnection;
public abstract class AbstractMessageLogger implements MessageLogger
@@ -197,7 +197,7 @@
}
}
- protected String generateSessionMessage(final AMQSessionModel session)
+ protected String generateSessionMessage(final AMQPSession session)
{
AMQPConnection<?> connection = session.getAMQPConnection();
return "[" + MessageFormat.format(CHANNEL_FORMAT, connection == null ? -1L : connection.getConnectionId(),
diff --git a/broker-core/src/main/java/org/apache/qpid/server/logging/LogSubject.java b/broker-core/src/main/java/org/apache/qpid/server/logging/LogSubject.java
index 462b45d..9bb987c 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/logging/LogSubject.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/logging/LogSubject.java
@@ -32,5 +32,5 @@
*
* @return String the display representation of this LogSubject
*/
- public String toLogString();
+ String toLogString();
}
\ No newline at end of file
diff --git a/broker-core/src/main/java/org/apache/qpid/server/logging/subjects/ChannelLogSubject.java b/broker-core/src/main/java/org/apache/qpid/server/logging/subjects/ChannelLogSubject.java
index 3d8d2f5..f0d1369 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/logging/subjects/ChannelLogSubject.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/logging/subjects/ChannelLogSubject.java
@@ -22,13 +22,13 @@
import static org.apache.qpid.server.logging.subjects.LogSubjectFormat.CHANNEL_FORMAT;
-import org.apache.qpid.server.protocol.AMQSessionModel;
+import org.apache.qpid.server.session.AMQPSession;
import org.apache.qpid.server.transport.AMQPConnection;
public class ChannelLogSubject extends AbstractLogSubject
{
- private final AMQSessionModel<?,?> _sessionModel;
- public ChannelLogSubject(AMQSessionModel session)
+ private final AMQPSession<?,?> _sessionModel;
+ public ChannelLogSubject(AMQPSession<?,?> session)
{
_sessionModel = session;
updateSessionDetails();
diff --git a/broker-core/src/main/java/org/apache/qpid/server/message/MessageSource.java b/broker-core/src/main/java/org/apache/qpid/server/message/MessageSource.java
index d49f9d1..e3d380d 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/message/MessageSource.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/message/MessageSource.java
@@ -26,7 +26,7 @@
import org.apache.qpid.server.consumer.ConsumerOption;
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.session.AMQPSession;
import org.apache.qpid.server.store.TransactionLogResource;
public interface MessageSource extends TransactionLogResource, MessageNode
@@ -41,7 +41,7 @@
Collection<? extends MessageInstanceConsumer> getConsumers();
- boolean verifySessionAccess(AMQSessionModel<?,?> session);
+ boolean verifySessionAccess(AMQPSession<?,?> session);
/**
* ExistingExclusiveConsumer signals a failure to create a consumer, because an exclusive consumer
diff --git a/broker-core/src/main/java/org/apache/qpid/server/model/Consumer.java b/broker-core/src/main/java/org/apache/qpid/server/model/Consumer.java
index b69ee5e..3f43f01 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/model/Consumer.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/model/Consumer.java
@@ -24,7 +24,7 @@
import org.apache.qpid.server.consumer.ConsumerTarget;
import org.apache.qpid.server.message.MessageInstanceConsumer;
-import org.apache.qpid.server.protocol.AMQSessionModel;
+import org.apache.qpid.server.session.AMQPSession;
@ManagedObject(creatable = false, amqpName = "org.apache.qpid.Consumer")
public interface Consumer<X extends Consumer<X,T>, T extends ConsumerTarget> extends ConfiguredObject<X>, MessageInstanceConsumer<T>
@@ -80,7 +80,7 @@
long getUnacknowledgedMessages();
- AMQSessionModel getSessionModel();
+ AMQPSession<?,?> getSessionModel();
long getConsumerNumber();
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
deleted file mode 100644
index 3d24785..0000000
--- a/broker-core/src/main/java/org/apache/qpid/server/protocol/AMQSessionModel.java
+++ /dev/null
@@ -1,95 +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.protocol;
-
-import java.util.Collection;
-import java.util.UUID;
-
-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.transport.AMQPConnection;
-import org.apache.qpid.server.util.Deletable;
-import org.apache.qpid.transport.network.Ticker;
-
-/**
- * Session model interface.
- */
-public interface AMQSessionModel<T extends AMQSessionModel<T,X>, X extends ConsumerTarget<X>> extends Deletable<T>
-{
- UUID getId();
-
- AMQPConnection<?> getAMQPConnection();
-
- void close();
-
- LogSubject getLogSubject();
-
- void doTimeoutAction(String reason);
-
- void block(Queue<?> queue);
-
- void unblock(Queue<?> queue);
-
- void block();
-
- void unblock();
-
- boolean getBlocking();
-
- Object getConnectionReference();
-
- int getUnacknowledgedMessageCount();
-
- long getTxnStart();
- long getTxnCommits();
- long getTxnRejects();
-
- int getChannelId();
-
- long getConsumerCount();
-
- Collection<Consumer<?,X>> getConsumers();
-
- /**
- * Return the time the current transaction started.
- *
- * @return the time this transaction started or 0 if not in a transaction
- */
- long getTransactionStartTimeLong();
-
- /**
- * Return the time of the last activity on the current transaction.
- *
- * @return the time of the last activity or 0 if not in a transaction
- */
- long getTransactionUpdateTimeLong();
-
- void transportStateChanged();
-
- boolean processPending();
-
- void addTicker(Ticker ticker);
- void removeTicker(Ticker ticker);
-
- void notifyWork(X target);
-}
diff --git a/broker-core/src/main/java/org/apache/qpid/server/protocol/CapacityChecker.java b/broker-core/src/main/java/org/apache/qpid/server/protocol/CapacityChecker.java
index 884929f..64492f7 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/protocol/CapacityChecker.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/protocol/CapacityChecker.java
@@ -20,7 +20,9 @@
*/
package org.apache.qpid.server.protocol;
+import org.apache.qpid.server.session.AMQPSession;
+
public interface CapacityChecker
{
- void checkCapacity(AMQSessionModel<?,?> channel);
+ void checkCapacity(AMQPSession<?,?> channel);
}
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 5525a25..6b04aac 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
@@ -98,11 +98,11 @@
import org.apache.qpid.server.plugin.MessageConverter;
import org.apache.qpid.server.plugin.MessageFilterFactory;
import org.apache.qpid.server.plugin.QpidServiceLoader;
-import org.apache.qpid.server.protocol.AMQSessionModel;
import org.apache.qpid.server.protocol.MessageConverterRegistry;
import org.apache.qpid.server.security.SecurityToken;
import org.apache.qpid.server.security.access.Operation;
import org.apache.qpid.server.security.auth.AuthenticatedPrincipal;
+import org.apache.qpid.server.session.AMQPSession;
import org.apache.qpid.server.store.MessageDurability;
import org.apache.qpid.server.store.MessageEnqueueRecord;
import org.apache.qpid.server.store.StorableMessageMetaData;
@@ -208,7 +208,7 @@
private AtomicBoolean _stopped = new AtomicBoolean(false);
- private final Set<AMQSessionModel<?,?>> _blockedChannels = Collections.newSetFromMap(new ConcurrentHashMap<AMQSessionModel<?, ?>, Boolean>());
+ private final Set<AMQPSession<?,?>> _blockedChannels = Collections.newSetFromMap(new ConcurrentHashMap<AMQPSession<?,?>, Boolean>());
private final AtomicBoolean _deleted = new AtomicBoolean(false);
private final SettableFuture<Integer> _deleteFuture = SettableFuture.create();
@@ -357,7 +357,7 @@
_queueHouseKeepingTask = new AdvanceConsumersTask();
Subject activeSubject = Subject.getSubject(AccessController.getContext());
Set<SessionPrincipal> sessionPrincipals = activeSubject == null ? Collections.<SessionPrincipal>emptySet() : activeSubject.getPrincipals(SessionPrincipal.class);
- AMQSessionModel<?,?> sessionModel;
+ AMQPSession<?, ?> sessionModel;
if(sessionPrincipals.isEmpty())
{
sessionModel = null;
@@ -1734,7 +1734,7 @@
}
@Override
- public void checkCapacity(AMQSessionModel<?,?> channel)
+ public void checkCapacity(AMQPSession<?,?> channel)
{
if(_queueFlowControlSizeBytes != 0L)
{
@@ -1767,6 +1767,7 @@
}
}
+ @Override
public void checkCapacity()
{
if(getEntries() != null)
@@ -1783,7 +1784,7 @@
_queueFlowResumeSizeBytes));
}
- for (final AMQSessionModel<?,?> blockedChannel : _blockedChannels)
+ for (final AMQPSession<?,?> blockedChannel : _blockedChannels)
{
blockedChannel.unblock(this);
_blockedChannels.remove(blockedChannel);
@@ -2587,7 +2588,7 @@
}
@Override
- public boolean verifySessionAccess(final AMQSessionModel<?,?> session)
+ public boolean verifySessionAccess(final AMQPSession<?,?> session)
{
boolean allowed;
switch(_exclusive)
@@ -2682,7 +2683,7 @@
case PRINCIPAL:
case CONTAINER:
case CONNECTION:
- AMQSessionModel session = null;
+ AMQPSession<?,?> session = null;
Iterator<QueueConsumer<?,?>> queueConsumerIterator = _queueConsumerManager.getAllIterator();
while(queueConsumerIterator.hasNext())
{
@@ -2728,7 +2729,7 @@
_exclusiveOwner = con;
break;
case SESSION:
- _exclusiveOwner = _exclusiveOwner == null ? null : ((AMQSessionModel)_exclusiveOwner).getAMQPConnection();
+ _exclusiveOwner = _exclusiveOwner == null ? null : ((AMQPSession<?,?>)_exclusiveOwner).getAMQPConnection();
break;
case LINK:
_exclusiveOwner = _exclusiveSubscriber == null ? null : _exclusiveSubscriber.getSessionModel().getAMQPConnection();
@@ -2761,7 +2762,7 @@
_exclusiveOwner = _exclusiveOwner == null ? null : ((AMQPConnection)_exclusiveOwner).getRemoteContainerName();
break;
case SESSION:
- _exclusiveOwner = _exclusiveOwner == null ? null : ((AMQSessionModel)_exclusiveOwner).getAMQPConnection().getRemoteContainerName();
+ _exclusiveOwner = _exclusiveOwner == null ? null : ((AMQPSession<?,?>)_exclusiveOwner).getAMQPConnection().getRemoteContainerName();
break;
case LINK:
_exclusiveOwner = _exclusiveSubscriber == null ? null : _exclusiveSubscriber.getSessionModel().getAMQPConnection().getRemoteContainerName();
@@ -2795,7 +2796,7 @@
_exclusiveOwner = _exclusiveOwner == null ? null : ((AMQPConnection)_exclusiveOwner).getAuthorizedPrincipal();
break;
case SESSION:
- _exclusiveOwner = _exclusiveOwner == null ? null : ((AMQSessionModel)_exclusiveOwner).getAMQPConnection().getAuthorizedPrincipal();
+ _exclusiveOwner = _exclusiveOwner == null ? null : ((AMQPSession<?,?>)_exclusiveOwner).getAMQPConnection().getAuthorizedPrincipal();
break;
case LINK:
_exclusiveOwner = _exclusiveSubscriber == null ? null : _exclusiveSubscriber.getSessionModel().getAMQPConnection().getAuthorizedPrincipal();
diff --git a/broker-core/src/main/java/org/apache/qpid/server/queue/QueueConsumerImpl.java b/broker-core/src/main/java/org/apache/qpid/server/queue/QueueConsumerImpl.java
index ec3a15f..316f661 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/queue/QueueConsumerImpl.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/queue/QueueConsumerImpl.java
@@ -56,9 +56,9 @@
import org.apache.qpid.server.model.LifetimePolicy;
import org.apache.qpid.server.model.ManagedAttributeField;
import org.apache.qpid.server.model.Queue;
-import org.apache.qpid.server.protocol.AMQSessionModel;
import org.apache.qpid.server.protocol.MessageConverterRegistry;
import org.apache.qpid.server.security.access.Operation;
+import org.apache.qpid.server.session.AMQPSession;
import org.apache.qpid.server.util.StateChangeListener;
class QueueConsumerImpl<T extends ConsumerTarget>
@@ -132,7 +132,7 @@
setupLogging();
}
- private static Map<String, Object> createAttributeMap(final AMQSessionModel sessionModel,
+ private static Map<String, Object> createAttributeMap(final AMQPSession<?,?> sessionModel,
String linkName,
FilterManager filters,
EnumSet<ConsumerOption> optionSet,
@@ -215,7 +215,7 @@
}
@Override
- public AMQSessionModel getSessionModel()
+ public AMQPSession<?,?> getSessionModel()
{
return _target.getSessionModel();
}
diff --git a/broker-core/src/main/java/org/apache/qpid/server/session/AMQPSession.java b/broker-core/src/main/java/org/apache/qpid/server/session/AMQPSession.java
index 3dcf379..6274418 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/session/AMQPSession.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/session/AMQPSession.java
@@ -20,16 +20,79 @@
*/
package org.apache.qpid.server.session;
+import java.util.Collection;
+import java.util.UUID;
+
import org.apache.qpid.server.consumer.ConsumerTarget;
import org.apache.qpid.server.logging.EventLoggerProvider;
+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.protocol.AMQSessionModel;
+import org.apache.qpid.server.transport.AMQPConnection;
import org.apache.qpid.server.util.Deletable;
+import org.apache.qpid.transport.network.Ticker;
public interface AMQPSession<S extends org.apache.qpid.server.session.AMQPSession<S, X>,
X extends ConsumerTarget<X>> extends Session<S>,
Deletable<S>,
- EventLoggerProvider,
- AMQSessionModel<S, X>
+ EventLoggerProvider
{
+ Object getConnectionReference();
+
+ int getChannelId();
+
+ AMQPConnection<?> getAMQPConnection();
+
+ LogSubject getLogSubject();
+
+ void doTimeoutAction(String reason);
+
+ void block(Queue<?> queue);
+
+ void unblock(Queue<?> queue);
+
+ void block();
+
+ void unblock();
+
+ boolean getBlocking();
+
+ int getUnacknowledgedMessageCount();
+
+ long getTxnStart();
+
+ long getTxnCommits();
+
+ long getTxnRejects();
+
+ long getConsumerCount();
+
+ Collection<Consumer<?,X>> getConsumers();
+
+ /**
+ * Return the time the current transaction started.
+ *
+ * @return the time this transaction started or 0 if not in a transaction
+ */
+ long getTransactionStartTimeLong();
+
+ /**
+ * Return the time of the last activity on the current transaction.
+ *
+ * @return the time of the last activity or 0 if not in a transaction
+ */
+ long getTransactionUpdateTimeLong();
+
+ void transportStateChanged();
+
+ boolean processPending();
+
+ void addTicker(Ticker ticker);
+
+ void removeTicker(Ticker ticker);
+
+ void notifyWork(X target);
+
+ void close();
}
diff --git a/broker-core/src/main/java/org/apache/qpid/server/transport/AMQPConnection.java b/broker-core/src/main/java/org/apache/qpid/server/transport/AMQPConnection.java
index 5323a50..3592735 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/transport/AMQPConnection.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/transport/AMQPConnection.java
@@ -32,9 +32,7 @@
import org.apache.qpid.server.logging.EventLoggerProvider;
import org.apache.qpid.server.model.Broker;
import org.apache.qpid.server.model.Connection;
-import org.apache.qpid.server.model.ContextProvider;
-import org.apache.qpid.server.model.NamedAddressSpace;
-import org.apache.qpid.server.protocol.AMQSessionModel;
+import org.apache.qpid.server.session.AMQPSession;
import org.apache.qpid.server.util.Deletable;
public interface AMQPConnection<C extends AMQPConnection<C>> extends Connection<C>, Deletable<C>, EventLoggerProvider
@@ -65,7 +63,7 @@
void registerMessageDelivered(long size);
- void closeSessionAsync(AMQSessionModel<?,?> session, CloseReason reason, String message);
+ void closeSessionAsync(AMQPSession<?,?> session, CloseReason reason, String message);
SocketAddress getRemoteSocketAddress();
@@ -101,11 +99,11 @@
*
* @return list of sessions
*/
- Collection<? extends AMQSessionModel<?,?>> getSessionModels();
+ Collection<? extends AMQPSession<?,?>> getSessionModels();
void resetStatistics();
- void notifyWork(AMQSessionModel<?,?> sessionModel);
+ void notifyWork(AMQPSession<?,?> sessionModel);
boolean isTransportBlockedForWriting();
diff --git a/broker-core/src/main/java/org/apache/qpid/server/txn/DistributedTransaction.java b/broker-core/src/main/java/org/apache/qpid/server/txn/DistributedTransaction.java
index 4a220e8..0a82625 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/txn/DistributedTransaction.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/txn/DistributedTransaction.java
@@ -25,8 +25,8 @@
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.session.AMQPSession;
import org.apache.qpid.server.store.MessageEnqueueRecord;
import org.apache.qpid.server.store.TransactionLogResource;
import org.apache.qpid.transport.Xid;
@@ -37,11 +37,11 @@
private final AutoCommitTransaction _autoCommitTransaction;
private DtxBranch _branch;
- private AMQSessionModel _session;
- private DtxRegistry _dtxRegistry;
+ private final AMQPSession<?,?> _session;
+ private final DtxRegistry _dtxRegistry;
- public DistributedTransaction(AMQSessionModel session, DtxRegistry dtxRegistry)
+ public DistributedTransaction(AMQPSession<?,?> session, DtxRegistry dtxRegistry)
{
_session = session;
_dtxRegistry = dtxRegistry;
diff --git a/broker-core/src/main/java/org/apache/qpid/server/txn/DtxBranch.java b/broker-core/src/main/java/org/apache/qpid/server/txn/DtxBranch.java
index 4b7b792..9570622 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/txn/DtxBranch.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/txn/DtxBranch.java
@@ -31,7 +31,7 @@
import org.slf4j.LoggerFactory;
import org.apache.qpid.server.message.EnqueueableMessage;
-import org.apache.qpid.server.protocol.AMQSessionModel;
+import org.apache.qpid.server.session.AMQPSession;
import org.apache.qpid.server.store.MessageEnqueueRecord;
import org.apache.qpid.server.store.StoreException;
import org.apache.qpid.server.store.Transaction;
@@ -47,7 +47,7 @@
private final List<ServerTransaction.Action> _postTransactionActions = new ArrayList<ServerTransaction.Action>();
private State _state = State.ACTIVE;
private long _timeout;
- private Map<AMQSessionModel, State> _associatedSessions = new HashMap<AMQSessionModel, State>();
+ private Map<AMQPSession<?,?>, State> _associatedSessions = new HashMap<>();
private final List<EnqueueRecord> _enqueueRecords = new ArrayList<>();
private final List<DequeueRecord> _dequeueRecords = new ArrayList<>();
@@ -146,7 +146,7 @@
return _timeout != 0 && _expiration < System.currentTimeMillis();
}
- public synchronized boolean isAssociated(AMQSessionModel session)
+ public synchronized boolean isAssociated(AMQPSession<?,?> session)
{
return _associatedSessions.containsKey(session);
}
@@ -177,17 +177,17 @@
_associatedSessions.clear();
}
- synchronized boolean associateSession(AMQSessionModel associatedSession)
+ synchronized boolean associateSession(AMQPSession<?,?> associatedSession)
{
return _associatedSessions.put(associatedSession, State.ACTIVE) != null;
}
- synchronized boolean disassociateSession(AMQSessionModel associatedSession)
+ synchronized boolean disassociateSession(AMQPSession<?,?> associatedSession)
{
return _associatedSessions.remove(associatedSession) != null;
}
- public synchronized boolean resumeSession(AMQSessionModel session)
+ public synchronized boolean resumeSession(AMQPSession<?,?> session)
{
if(_associatedSessions.containsKey(session) && _associatedSessions.get(session) == State.SUSPENDED)
{
@@ -197,7 +197,7 @@
return false;
}
- public synchronized boolean suspendSession(AMQSessionModel session)
+ public synchronized boolean suspendSession(AMQPSession<?,?> session)
{
if(_associatedSessions.containsKey(session) && _associatedSessions.get(session) == State.ACTIVE)
{
diff --git a/broker-core/src/main/java/org/apache/qpid/server/txn/DtxRegistry.java b/broker-core/src/main/java/org/apache/qpid/server/txn/DtxRegistry.java
index 9dad227..5597937 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/txn/DtxRegistry.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/txn/DtxRegistry.java
@@ -27,7 +27,7 @@
import java.util.Map;
import java.util.concurrent.ScheduledFuture;
-import org.apache.qpid.server.protocol.AMQSessionModel;
+import org.apache.qpid.server.session.AMQPSession;
import org.apache.qpid.server.store.MessageStore;
import org.apache.qpid.server.store.StoreException;
import org.apache.qpid.server.virtualhost.QueueManagingVirtualHost;
@@ -328,7 +328,7 @@
return inDoubt;
}
- public synchronized void endAssociations(AMQSessionModel session)
+ public synchronized void endAssociations(AMQPSession<?,?> session)
{
for(DtxBranch branch : _branches.values())
{
diff --git a/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractSystemMessageSource.java b/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractSystemMessageSource.java
index 9c31685..df9f0fe 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractSystemMessageSource.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractSystemMessageSource.java
@@ -40,8 +40,8 @@
import org.apache.qpid.server.message.ServerMessage;
import org.apache.qpid.server.message.internal.InternalMessage;
import org.apache.qpid.server.model.NamedAddressSpace;
-import org.apache.qpid.server.protocol.AMQSessionModel;
import org.apache.qpid.server.message.MessageContainer;
+import org.apache.qpid.server.session.AMQPSession;
import org.apache.qpid.server.store.MessageDurability;
import org.apache.qpid.server.store.MessageEnqueueRecord;
import org.apache.qpid.server.store.TransactionLogResource;
@@ -104,7 +104,7 @@
}
@Override
- public boolean verifySessionAccess(final AMQSessionModel<?,?> session)
+ public boolean verifySessionAccess(final AMQPSession<?,?> session)
{
return true;
}
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 b755e3b..6319a99 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
@@ -25,29 +25,17 @@
import static org.mockito.Mockito.when;
import java.util.ArrayList;
-import java.util.Collection;
-import java.util.UUID;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
-import org.apache.qpid.server.configuration.updater.CurrentThreadTaskExecutor;
-import org.apache.qpid.server.configuration.updater.TaskExecutor;
-import org.apache.qpid.server.logging.LogSubject;
+import org.apache.qpid.server.message.MessageContainer;
import org.apache.qpid.server.message.MessageInstance;
import org.apache.qpid.server.message.MessageInstanceConsumer;
import org.apache.qpid.server.message.ServerMessage;
-import org.apache.qpid.server.model.BrokerModel;
-import org.apache.qpid.server.model.ConfiguredObjectFactory;
-import org.apache.qpid.server.model.ConfiguredObjectFactoryImpl;
-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.message.MessageContainer;
+import org.apache.qpid.server.session.AMQPSession;
import org.apache.qpid.server.transport.AMQPConnection;
-import org.apache.qpid.server.util.Action;
-import org.apache.qpid.transport.network.Ticker;
public class TestConsumerTarget implements ConsumerTarget<TestConsumerTarget>
{
@@ -56,13 +44,19 @@
private String tag = "mocktag";
private Queue<?> queue = null;
private State _state = State.OPEN;
- private ArrayList<MessageInstance> _messages = new ArrayList<MessageInstance>();
+ private ArrayList<MessageInstance> _messages = new ArrayList<>();
private boolean _isActive = true;
private MessageInstanceConsumer _consumer;
- private MockSessionModel _sessionModel = new MockSessionModel();
+ private AMQPSession _sessionModel = mock(AMQPSession.class);
private boolean _notifyDesired;
+ public TestConsumerTarget()
+ {
+ when(_sessionModel.getChannelId()).thenReturn(0);
+ when(_sessionModel.getAMQPConnection()).thenReturn(mock(AMQPConnection.class));
+ }
+
public boolean close()
{
_closed = true;
@@ -91,7 +85,7 @@
return queue;
}
- public AMQSessionModel getSessionModel()
+ public AMQPSession getSessionModel()
{
return _sessionModel;
}
@@ -236,182 +230,4 @@
_notifyDesired = isNotifyWorkDesired();
}
}
-
- private static class MockSessionModel implements AMQSessionModel<MockSessionModel, TestConsumerTarget>
- {
- private final UUID _id = UUID.randomUUID();
- private Session _modelObject;
- private AMQPConnection<?> _connection = mock(AMQPConnection.class);
-
- private MockSessionModel()
- {
- _modelObject = mock(Session.class);
- when(_modelObject.getCategoryClass()).thenReturn(Session.class);
- ConfiguredObjectFactory factory = new ConfiguredObjectFactoryImpl(BrokerModel.getInstance());
- when(_modelObject.getObjectFactory()).thenReturn(factory);
- when(_modelObject.getModel()).thenReturn(factory.getModel());
- TaskExecutor taskExecutor = CurrentThreadTaskExecutor.newStartedInstance();
- when(_modelObject.getTaskExecutor()).thenReturn(taskExecutor);
- when(_modelObject.getChildExecutor()).thenReturn(taskExecutor);
- }
-
- @Override
- public UUID getId()
- {
- return _id;
- }
-
- @Override
- public AMQPConnection<?> getAMQPConnection()
- {
- return _connection;
- }
-
- @Override
- public void close()
- {
- }
-
- @Override
- public LogSubject getLogSubject()
- {
- return null;
- }
-
- @Override
- public void doTimeoutAction(final String reason)
- {
- }
-
- @Override
- public void block(Queue<?> queue)
- {
- }
-
- @Override
- public void unblock(Queue<?> queue)
- {
- }
-
- @Override
- public void block()
- {
- }
-
- @Override
- public void unblock()
- {
- }
-
- @Override
- public boolean getBlocking()
- {
- return false;
- }
-
- @Override
- public Object getConnectionReference()
- {
- return this;
- }
-
- @Override
- public int getUnacknowledgedMessageCount()
- {
- return 0;
- }
-
- @Override
- public long getTxnStart()
- {
- return 0L;
- }
-
- @Override
- public long getTxnCommits()
- {
- return 0L;
- }
-
- @Override
- public long getTxnRejects()
- {
- return 0L;
- }
-
- @Override
- public int getChannelId()
- {
- return 0;
- }
-
- @Override
- public long getConsumerCount()
- {
- return 0;
- }
-
- @Override
- public Collection<Consumer<?,TestConsumerTarget>> getConsumers()
- {
- return null;
- }
-
- @Override
- public long getTransactionStartTimeLong()
- {
- return 0;
- }
-
- @Override
- public long getTransactionUpdateTimeLong()
- {
- return 0;
- }
-
- @Override
- public void addDeleteTask(final Action task)
- {
-
- }
-
- @Override
- public void removeDeleteTask(final Action task)
- {
-
- }
-
-
- @Override
- public void transportStateChanged()
- {
-
- }
-
- @Override
- public boolean processPending()
- {
- return false;
- }
-
- @Override
- public void addTicker(final Ticker ticker)
- {
-
- }
-
- @Override
- public void removeTicker(final Ticker ticker)
- {
-
- }
-
- @Override
- public void notifyWork(final TestConsumerTarget target)
- {
- _connection.notifyWork(this);
- }
-
- }
-
}
diff --git a/broker-core/src/test/java/org/apache/qpid/server/logging/subjects/ChannelLogSubjectTest.java b/broker-core/src/test/java/org/apache/qpid/server/logging/subjects/ChannelLogSubjectTest.java
index 5d28e65..25cbb74 100644
--- a/broker-core/src/test/java/org/apache/qpid/server/logging/subjects/ChannelLogSubjectTest.java
+++ b/broker-core/src/test/java/org/apache/qpid/server/logging/subjects/ChannelLogSubjectTest.java
@@ -20,11 +20,11 @@
*/
package org.apache.qpid.server.logging.subjects;
-import org.apache.qpid.server.protocol.AMQSessionModel;
-
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
+import org.apache.qpid.server.session.AMQPSession;
+
/**
* Validate ChannelLogSubjects are logged as expected
*/
@@ -37,7 +37,7 @@
{
super.setUp();
- AMQSessionModel session = mock(AMQSessionModel.class);
+ AMQPSession<?,?> session = mock(AMQPSession.class);
when(session.getAMQPConnection()).thenReturn(getConnection());
when(session.getChannelId()).thenReturn(_channelID);
_subject = new ChannelLogSubject(session);
diff --git a/broker-core/src/test/java/org/apache/qpid/server/model/BrokerTestHelper.java b/broker-core/src/test/java/org/apache/qpid/server/model/BrokerTestHelper.java
index 50f9d28..58f8770 100644
--- a/broker-core/src/test/java/org/apache/qpid/server/model/BrokerTestHelper.java
+++ b/broker-core/src/test/java/org/apache/qpid/server/model/BrokerTestHelper.java
@@ -49,11 +49,11 @@
import org.apache.qpid.server.configuration.updater.CurrentThreadTaskExecutor;
import org.apache.qpid.server.configuration.updater.TaskExecutor;
import org.apache.qpid.server.logging.EventLogger;
-import org.apache.qpid.server.protocol.AMQSessionModel;
import org.apache.qpid.server.security.AccessControl;
import org.apache.qpid.server.security.Result;
import org.apache.qpid.server.security.SecurityToken;
import org.apache.qpid.server.security.access.Operation;
+import org.apache.qpid.server.session.AMQPSession;
import org.apache.qpid.server.store.DurableConfigurationStore;
import org.apache.qpid.server.store.preferences.PreferenceStore;
import org.apache.qpid.server.transport.AMQPConnection;
@@ -225,22 +225,22 @@
return createVirtualHost(attributes, broker, defaultVHN, accessControl);
}
- public static AMQSessionModel<?,?> createSession(int channelId, AMQPConnection<?> connection)
+ public static AMQPSession<?,?> createSession(int channelId, AMQPConnection<?> connection)
{
@SuppressWarnings("rawtypes")
- AMQSessionModel session = mock(AMQSessionModel.class);
+ AMQPSession session = mock(AMQPSession.class);
when(session.getAMQPConnection()).thenReturn(connection);
when(session.getChannelId()).thenReturn(channelId);
return session;
}
- public static AMQSessionModel<?,?> createSession(int channelId) throws Exception
+ public static AMQPSession<?,?> createSession(int channelId) throws Exception
{
AMQPConnection<?> session = createConnection();
return createSession(channelId, session);
}
- public static AMQSessionModel<?,?> createSession() throws Exception
+ public static AMQPSession<?,?> createSession() throws Exception
{
return createSession(1);
}
diff --git a/broker-core/src/test/java/org/apache/qpid/server/queue/AbstractQueueTestBase.java b/broker-core/src/test/java/org/apache/qpid/server/queue/AbstractQueueTestBase.java
index 3598993..bc65ee3 100644
--- a/broker-core/src/test/java/org/apache/qpid/server/queue/AbstractQueueTestBase.java
+++ b/broker-core/src/test/java/org/apache/qpid/server/queue/AbstractQueueTestBase.java
@@ -79,7 +79,7 @@
private String _owner = "owner";
private String _routingKey = "routing key";
private DirectExchangeImpl _exchange;
- private TestConsumerTarget _consumerTarget = new TestConsumerTarget();
+ private TestConsumerTarget _consumerTarget = new TestConsumerTarget(); // TODO replace with minimally configured mockito mock
private QueueConsumer<?,?> _consumer;
private Map<String,Object> _arguments = Collections.emptyMap();
diff --git a/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/AMQPConnection_0_10Impl.java b/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/AMQPConnection_0_10Impl.java
index 581c50e..195bc07 100755
--- a/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/AMQPConnection_0_10Impl.java
+++ b/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/AMQPConnection_0_10Impl.java
@@ -40,8 +40,8 @@
import org.apache.qpid.server.model.Protocol;
import org.apache.qpid.server.model.Transport;
import org.apache.qpid.server.model.port.AmqpPort;
-import org.apache.qpid.server.protocol.AMQSessionModel;
import org.apache.qpid.server.security.SubjectCreator;
+import org.apache.qpid.server.session.AMQPSession;
import org.apache.qpid.server.store.StoreException;
import org.apache.qpid.server.transport.AbstractAMQPConnection;
import org.apache.qpid.server.transport.AggregateTicker;
@@ -71,8 +71,8 @@
private final AtomicReference<Action<ProtocolEngine>> _workListener = new AtomicReference<>();
private ServerDisassembler _disassembler;
- private final Set<AMQSessionModel<?,?>> _sessionsWithWork =
- Collections.newSetFromMap(new ConcurrentHashMap<AMQSessionModel<?,?>, Boolean>());
+ private final Set<AMQPSession<?,?>> _sessionsWithWork =
+ Collections.newSetFromMap(new ConcurrentHashMap<AMQPSession<?,?>, Boolean>());
public AMQPConnection_0_10Impl(final Broker<?> broker,
@@ -278,7 +278,7 @@
}
@Override
- public void notifyWork(final AMQSessionModel<?,?> sessionModel)
+ public void notifyWork(final AMQPSession<?,?> sessionModel)
{
_sessionsWithWork.add(sessionModel);
notifyWork();
@@ -309,7 +309,7 @@
}
@Override
- public void closeSessionAsync(final AMQSessionModel<?,?> session,
+ public void closeSessionAsync(final AMQPSession<?,?> session,
final CloseReason reason, final String message)
{
ServerSession s = ((Session_0_10)session).getServerSession();
diff --git a/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnection.java b/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnection.java
index 2e23c1b..fb58454 100644
--- a/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnection.java
+++ b/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnection.java
@@ -46,8 +46,8 @@
import org.apache.qpid.server.model.NamedAddressSpace;
import org.apache.qpid.server.model.Transport;
import org.apache.qpid.server.model.port.AmqpPort;
-import org.apache.qpid.server.protocol.AMQSessionModel;
import org.apache.qpid.server.protocol.ConnectionClosingTicker;
+import org.apache.qpid.server.session.AMQPSession;
import org.apache.qpid.server.transport.AMQPConnection;
import org.apache.qpid.server.transport.ServerNetworkConnection;
import org.apache.qpid.server.util.Action;
@@ -486,16 +486,16 @@
}
}
- public Iterator<Runnable> processPendingIterator(final Set<AMQSessionModel<?,?>> sessionsWithWork)
+ public Iterator<Runnable> processPendingIterator(final Set<AMQPSession<?,?>> sessionsWithWork)
{
return new ProcessPendingIterator(sessionsWithWork);
}
private class ProcessPendingIterator implements Iterator<Runnable>
{
- private final Collection<AMQSessionModel<?,?>> _sessionsWithPending;
- private Iterator<? extends AMQSessionModel<?,?>> _sessionIterator;
- private ProcessPendingIterator(final Set<AMQSessionModel<?,?>> sessionsWithWork)
+ private final Collection<AMQPSession<?,?>> _sessionsWithPending;
+ private Iterator<? extends AMQPSession<?,?>> _sessionIterator;
+ private ProcessPendingIterator(final Set<AMQPSession<?,?>> sessionsWithWork)
{
_sessionsWithPending = sessionsWithWork;
_sessionIterator = _sessionsWithPending.iterator();
@@ -547,7 +547,7 @@
{
_sessionIterator = _sessionsWithPending.iterator();
}
- final AMQSessionModel<?,?> session = _sessionIterator.next();
+ final AMQPSession<?,?> session = _sessionIterator.next();
return new Runnable()
{
@Override
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 df6766c..ae69352 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
@@ -30,13 +30,12 @@
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.protocol.AMQSessionModel;
import org.apache.qpid.server.protocol.PublishAuthorisationCache;
import org.apache.qpid.server.session.AbstractAMQPSession;
import org.apache.qpid.server.util.Action;
public class Session_0_10 extends AbstractAMQPSession<Session_0_10, ConsumerTarget_0_10>
- implements AMQSessionModel<Session_0_10, ConsumerTarget_0_10>, LogSubject
+ implements LogSubject, org.apache.qpid.server.util.Deletable<Session_0_10>
{
private final AMQPConnection_0_10 _connection;
private final ServerSession _serverSession;
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 e9f9602..e050df0 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
@@ -67,6 +67,7 @@
import org.apache.qpid.server.filter.MessageFilter;
import org.apache.qpid.server.logging.EventLoggerProvider;
import org.apache.qpid.server.logging.LogMessage;
+import org.apache.qpid.server.logging.LogSubject;
import org.apache.qpid.server.logging.messages.ChannelMessages;
import org.apache.qpid.server.logging.messages.ExchangeMessages;
import org.apache.qpid.server.message.InstanceProperties;
@@ -78,7 +79,6 @@
import org.apache.qpid.server.message.RoutingResult;
import org.apache.qpid.server.message.ServerMessage;
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.v0_8.UnacknowledgedMessageMap.Visitor;
import org.apache.qpid.server.queue.QueueArgumentsConverter;
@@ -97,12 +97,12 @@
import org.apache.qpid.server.virtualhost.ExchangeIsAlternateException;
import org.apache.qpid.server.virtualhost.RequiredExchangeException;
import org.apache.qpid.server.virtualhost.ReservedExchangeNameException;
+import org.apache.qpid.transport.network.Ticker;
public class AMQChannel extends AbstractAMQPSession<AMQChannel, ConsumerTarget_0_8>
- implements AMQSessionModel<AMQChannel, ConsumerTarget_0_8>,
- AsyncAutoCommitTransaction.FutureRecorder,
+ implements AsyncAutoCommitTransaction.FutureRecorder,
ServerChannelMethodProcessor,
- EventLoggerProvider, CreditRestorer
+ EventLoggerProvider, CreditRestorer, org.apache.qpid.server.util.Deletable<AMQChannel>
{
public static final int DEFAULT_PREFETCH = 4096;
@@ -1413,6 +1413,7 @@
{
}
+ @Override
public void performAction(MessageInstance entry)
{
TransactionLogResource queue = entry.getOwningResource();
diff --git a/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQPConnection_0_8Impl.java b/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQPConnection_0_8Impl.java
index 61401e8..84f8830 100644
--- a/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQPConnection_0_8Impl.java
+++ b/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQPConnection_0_8Impl.java
@@ -63,11 +63,11 @@
import org.apache.qpid.server.model.Protocol;
import org.apache.qpid.server.model.Transport;
import org.apache.qpid.server.model.port.AmqpPort;
-import org.apache.qpid.server.protocol.AMQSessionModel;
import org.apache.qpid.server.protocol.ConnectionClosingTicker;
import org.apache.qpid.server.security.SubjectCreator;
import org.apache.qpid.server.security.auth.SubjectAuthenticationResult;
import org.apache.qpid.server.security.auth.sasl.SaslNegotiator;
+import org.apache.qpid.server.session.AMQPSession;
import org.apache.qpid.server.store.StoreException;
import org.apache.qpid.server.transport.AbstractAMQPConnection;
import org.apache.qpid.server.transport.AggregateTicker;
@@ -158,8 +158,8 @@
private volatile boolean _transportBlockedForWriting;
private volatile SubjectAuthenticationResult _successfulAuthenticationResult;
- private final Set<AMQSessionModel<?,?>> _sessionsWithWork =
- Collections.newSetFromMap(new ConcurrentHashMap<AMQSessionModel<?,?>, Boolean>());
+ private final Set<AMQPSession<?,?>> _sessionsWithWork =
+ Collections.newSetFromMap(new ConcurrentHashMap<AMQPSession<?,?>, Boolean>());
public AMQPConnection_0_8Impl(Broker<?> broker,
@@ -739,7 +739,7 @@
return String.valueOf(getNetwork().getRemoteAddress());
}
- public void closeSessionAsync(final AMQSessionModel<?,?> session, final CloseReason reason, final String message)
+ public void closeSessionAsync(final AMQPSession<?,?> session, final CloseReason reason, final String message)
{
final int cause;
switch (reason)
@@ -1354,7 +1354,7 @@
}
@Override
- public void notifyWork(final AMQSessionModel<?,?> sessionModel)
+ public void notifyWork(final AMQPSession<?,?> sessionModel)
{
_sessionsWithWork.add(sessionModel);
notifyWork();
@@ -1384,7 +1384,7 @@
private class ProcessPendingIterator implements Iterator<Runnable>
{
- private Iterator<? extends AMQSessionModel<?,?>> _sessionIterator;
+ private Iterator<? extends AMQPSession<?,?>> _sessionIterator;
private ProcessPendingIterator()
{
@@ -1436,7 +1436,7 @@
{
_sessionIterator = _sessionsWithWork.iterator();
}
- final AMQSessionModel<?,?> session = _sessionIterator.next();
+ final AMQPSession<?,?> session = _sessionIterator.next();
return new Runnable()
{
@Override
diff --git a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/AMQPConnection_1_0Impl.java b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/AMQPConnection_1_0Impl.java
index 0b8572c..f9b85e1 100644
--- a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/AMQPConnection_1_0Impl.java
+++ b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/AMQPConnection_1_0Impl.java
@@ -57,7 +57,6 @@
import org.apache.qpid.server.model.Protocol;
import org.apache.qpid.server.model.Transport;
import org.apache.qpid.server.model.port.AmqpPort;
-import org.apache.qpid.server.protocol.AMQSessionModel;
import org.apache.qpid.server.protocol.ConnectionClosingTicker;
import org.apache.qpid.server.protocol.v1_0.codec.DescribedTypeConstructorRegistry;
import org.apache.qpid.server.protocol.v1_0.codec.FrameWriter;
@@ -103,6 +102,7 @@
import org.apache.qpid.server.security.auth.manager.AnonymousAuthenticationManager;
import org.apache.qpid.server.security.auth.manager.ExternalAuthenticationManagerImpl;
import org.apache.qpid.server.security.auth.sasl.SaslNegotiator;
+import org.apache.qpid.server.session.AMQPSession;
import org.apache.qpid.server.store.StoreException;
import org.apache.qpid.server.transport.AbstractAMQPConnection;
import org.apache.qpid.server.transport.AggregateTicker;
@@ -236,8 +236,8 @@
private boolean _closedOnOpen;
- private final Set<AMQSessionModel<?,?>> _sessionsWithWork =
- Collections.newSetFromMap(new ConcurrentHashMap<AMQSessionModel<?,?>, Boolean>());
+ private final Set<AMQPSession<?,?>> _sessionsWithWork =
+ Collections.newSetFromMap(new ConcurrentHashMap<AMQPSession<?,?>, Boolean>());
AMQPConnection_1_0Impl(final Broker<?> broker,
final ServerNetworkConnection network,
@@ -1387,7 +1387,7 @@
}
@Override
- public void notifyWork(final AMQSessionModel<?,?> sessionModel)
+ public void notifyWork(final AMQPSession<?,?> sessionModel)
{
_sessionsWithWork.add(sessionModel);
notifyWork();
@@ -1439,7 +1439,7 @@
addAsyncTask(action);
}
- public void closeSessionAsync(final AMQSessionModel<?,?> session,
+ public void closeSessionAsync(final AMQPSession<?,?> session,
final CloseReason reason, final String message)
{
final ErrorCondition cause;
@@ -1633,7 +1633,7 @@
private class ProcessPendingIterator implements Iterator<Runnable>
{
- private Iterator<? extends AMQSessionModel<?,?>> _sessionIterator;
+ private Iterator<? extends AMQPSession<?,?>> _sessionIterator;
private ProcessPendingIterator()
{
_sessionIterator = _sessionsWithWork.iterator();
@@ -1683,7 +1683,7 @@
{
_sessionIterator = _sessionsWithWork.iterator();
}
- final AMQSessionModel<?,?> session = _sessionIterator.next();
+ final AMQPSession<?,?> session = _sessionIterator.next();
return new Runnable()
{
@Override
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 3af28fc..652fa35 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
@@ -70,7 +70,6 @@
import org.apache.qpid.server.model.Queue;
import org.apache.qpid.server.model.Session;
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.LinkRegistry;
import org.apache.qpid.server.protocol.v1_0.codec.QpidByteBufferUtils;
@@ -120,9 +119,10 @@
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
+ implements LogSubject, org.apache.qpid.server.util.Deletable<Session_1_0>
{
public static final Symbol DELAYED_DELIVERY = Symbol.valueOf("DELAYED_DELIVERY");
private static final Logger _logger = LoggerFactory.getLogger(Session_1_0.class);
diff --git a/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementAddressSpace.java b/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementAddressSpace.java
index d6c3b37..38b9b4f 100644
--- a/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementAddressSpace.java
+++ b/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementAddressSpace.java
@@ -53,11 +53,11 @@
import org.apache.qpid.server.model.PublishingLink;
import org.apache.qpid.server.model.port.AmqpPort;
import org.apache.qpid.server.plugin.SystemAddressSpaceCreator;
-import org.apache.qpid.server.protocol.AMQSessionModel;
import org.apache.qpid.server.protocol.LinkModel;
import org.apache.qpid.server.protocol.LinkRegistry;
import org.apache.qpid.server.security.SecurityToken;
import org.apache.qpid.server.security.access.Operation;
+import org.apache.qpid.server.session.AMQPSession;
import org.apache.qpid.server.store.MemoryMessageStore;
import org.apache.qpid.server.store.MessageStore;
import org.apache.qpid.server.store.StorableMessageMetaData;
@@ -253,7 +253,7 @@
if (!sessionPrincipals.isEmpty())
{
final ProxyMessageSource proxyMessageSource = new ProxyMessageSource(this, attributes);
- final AMQSessionModel session = sessionPrincipals.iterator().next().getSession();
+ final AMQPSession<?,?> session = sessionPrincipals.iterator().next().getSession();
final Object connectionReference = session.getConnectionReference();
ConcurrentMap<String, ProxyMessageSource> connectionSpecificDestinations =
_connectionSpecificDestinations.get(connectionReference);
diff --git a/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementNode.java b/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementNode.java
index 4b9e013..38f01d7 100644
--- a/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementNode.java
+++ b/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementNode.java
@@ -80,10 +80,10 @@
import org.apache.qpid.server.model.OperationParameter;
import org.apache.qpid.server.model.PublishingLink;
import org.apache.qpid.server.plugin.MessageConverter;
-import org.apache.qpid.server.protocol.AMQSessionModel;
import org.apache.qpid.server.protocol.MessageConverterRegistry;
import org.apache.qpid.server.queue.BaseQueue;
import org.apache.qpid.server.security.SecurityToken;
+import org.apache.qpid.server.session.AMQPSession;
import org.apache.qpid.server.store.MessageDurability;
import org.apache.qpid.server.store.MessageEnqueueRecord;
import org.apache.qpid.server.store.StorableMessageMetaData;
@@ -1004,7 +1004,7 @@
Set<SessionPrincipal> sessionPrincipals = currentSubject.getPrincipals(SessionPrincipal.class);
if (!sessionPrincipals.isEmpty())
{
- AMQSessionModel publishingSession = sessionPrincipals.iterator().next().getSession();
+ AMQPSession<?,?> publishingSession = sessionPrincipals.iterator().next().getSession();
for (ManagementNodeConsumer candidate : _consumers)
{
if (candidate.getTarget().getTargetAddress().equals(replyTo) && candidate.getSessionModel().getConnectionReference() == publishingSession.getConnectionReference())
@@ -1496,7 +1496,7 @@
}
@Override
- public boolean verifySessionAccess(final AMQSessionModel<?,?> session)
+ public boolean verifySessionAccess(final AMQPSession<?,?> session)
{
return true;
}
diff --git a/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementNodeConsumer.java b/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementNodeConsumer.java
index a567235..1119dbc 100644
--- a/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementNodeConsumer.java
+++ b/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementNodeConsumer.java
@@ -38,10 +38,10 @@
import org.apache.qpid.server.message.internal.InternalMessage;
import org.apache.qpid.server.model.NamedAddressSpace;
import org.apache.qpid.server.model.PublishingLink;
-import org.apache.qpid.server.protocol.AMQSessionModel;
import org.apache.qpid.server.message.MessageContainer;
import org.apache.qpid.server.queue.BaseQueue;
import org.apache.qpid.server.security.SecurityToken;
+import org.apache.qpid.server.session.AMQPSession;
import org.apache.qpid.server.store.MessageDurability;
import org.apache.qpid.server.store.MessageEnqueueRecord;
import org.apache.qpid.server.store.StorableMessageMetaData;
@@ -106,7 +106,7 @@
}
}
- AMQSessionModel getSessionModel()
+ AMQPSession<?,?> getSessionModel()
{
return _target.getSessionModel();
}
diff --git a/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ProxyMessageSource.java b/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ProxyMessageSource.java
index a667eed..a0151d2 100644
--- a/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ProxyMessageSource.java
+++ b/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ProxyMessageSource.java
@@ -52,8 +52,8 @@
import org.apache.qpid.server.model.ConfiguredObject;
import org.apache.qpid.server.model.NamedAddressSpace;
import org.apache.qpid.server.model.PublishingLink;
-import org.apache.qpid.server.protocol.AMQSessionModel;
import org.apache.qpid.server.security.SecurityToken;
+import org.apache.qpid.server.session.AMQPSession;
import org.apache.qpid.server.store.MessageDurability;
import org.apache.qpid.server.store.StorableMessageMetaData;
@@ -172,7 +172,7 @@
}
@Override
- public boolean verifySessionAccess(final AMQSessionModel<?, ?> session)
+ public boolean verifySessionAccess(final AMQPSession<?,?> session)
{
return session.getConnectionReference() == _connectionReference;
}
@@ -273,7 +273,7 @@
}
@Override
- public AMQSessionModel getSessionModel()
+ public AMQPSession getSessionModel()
{
return _underlying.getSessionModel();
}
diff --git a/systests/src/test/java/org/apache/qpid/server/store/VirtualHostMessageStoreTest.java b/systests/src/test/java/org/apache/qpid/server/store/VirtualHostMessageStoreTest.java
index 7680071..07701be 100644
--- a/systests/src/test/java/org/apache/qpid/server/store/VirtualHostMessageStoreTest.java
+++ b/systests/src/test/java/org/apache/qpid/server/store/VirtualHostMessageStoreTest.java
@@ -61,12 +61,13 @@
import org.apache.qpid.server.model.UUIDGenerator;
import org.apache.qpid.server.model.VirtualHost;
import org.apache.qpid.server.model.VirtualHostNode;
-import org.apache.qpid.server.protocol.AMQSessionModel;
+import org.apache.qpid.server.protocol.v0_10.Session_0_10;
import org.apache.qpid.server.protocol.v0_8.AMQMessage;
import org.apache.qpid.server.protocol.v0_8.MessageMetaData;
import org.apache.qpid.server.queue.LastValueQueue;
import org.apache.qpid.server.queue.PriorityQueue;
import org.apache.qpid.server.queue.StandardQueue;
+import org.apache.qpid.server.session.AMQPSession;
import org.apache.qpid.server.transport.AMQPConnection;
import org.apache.qpid.server.txn.AutoCommitTransaction;
import org.apache.qpid.server.txn.ServerTransaction;
@@ -683,7 +684,7 @@
queueArguments.put(Queue.DURABLE, durable);
queueArguments.put(Queue.LIFETIME_POLICY, LifetimePolicy.PERMANENT);
queueArguments.put(Queue.EXCLUSIVE, exclusive ? ExclusivityPolicy.CONTAINER : ExclusivityPolicy.NONE);
- AMQSessionModel sessionModel = mock(AMQSessionModel.class);
+ AMQPSession sessionModel = mock(AMQPSession.class);
AMQPConnection connectionModel = mock(AMQPConnection.class);
when(sessionModel.getAMQPConnection()).thenReturn(connectionModel);
when(connectionModel.getRemoteContainerName()).thenReturn(queueOwner);
