You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by rg...@apache.org on 2016/11/18 20:02:13 UTC
svn commit: r1770435 -
/qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java
Author: rgodfrey
Date: Fri Nov 18 20:02:13 2016
New Revision: 1770435
URL: http://svn.apache.org/viewvc?rev=1770435&view=rev
Log:
QPID-7529 : Log flow control enforcement in AMQP 1.0
Modified:
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-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=1770435&r1=1770434&r2=1770435&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 Fri Nov 18 20:02:13 2016
@@ -53,7 +53,11 @@ import org.apache.qpid.server.connection
import org.apache.qpid.server.consumer.ConsumerImpl;
import org.apache.qpid.server.consumer.ConsumerTarget;
import org.apache.qpid.server.consumer.ScheduledConsumerTargetSet;
+import org.apache.qpid.server.logging.EventLogger;
+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.subjects.ChannelLogSubject;
import org.apache.qpid.server.message.MessageDestination;
import org.apache.qpid.server.message.MessageSource;
import org.apache.qpid.server.model.AbstractConfigurationChangeListener;
@@ -113,6 +117,7 @@ public class Session_1_0 implements AMQS
EnumSet.of(SessionState.END_RECVD, SessionState.END_PIPE, SessionState.END_SENT, SessionState.ENDED);
private final AccessControlContext _accessControllerContext;
private final SecurityToken _securityToken;
+ private final ChannelLogSubject _logSubject;
private AutoCommitTransaction _transaction;
private final LinkedHashMap<Integer, ServerTransaction> _openTransactions =
@@ -196,6 +201,7 @@ public class Session_1_0 implements AMQS
_securityToken = connection.getAddressSpace() instanceof ConfiguredObject
? ((ConfiguredObject)connection.getAddressSpace()).newToken(_subject)
: connection.getBroker().newToken(_subject);
+ _logSubject = new ChannelLogSubject(this);
}
public void setReceivingChannel(final short receivingChannel)
@@ -1306,6 +1312,8 @@ public class Session_1_0 implements AMQS
{
if(_blockingEntities.add(queue))
{
+ messageWithSubject(ChannelMessages.FLOW_ENFORCED(queue.getName()));
+
for (ReceivingLinkEndpoint endpoint : _receivingLinkMap.values())
{
ReceivingLink_1_0 link = (ReceivingLink_1_0) endpoint.getLink();
@@ -1336,6 +1344,10 @@ public class Session_1_0 implements AMQS
{
if(_blockingEntities.remove(queue) && !_blockingEntities.contains(this))
{
+ if(_blockingEntities.isEmpty())
+ {
+ messageWithSubject(ChannelMessages.FLOW_REMOVED());
+ }
for (ReceivingLinkEndpoint endpoint : _receivingLinkMap.values())
{
ReceivingLink_1_0 link = (ReceivingLink_1_0) endpoint.getLink();
@@ -1365,6 +1377,8 @@ public class Session_1_0 implements AMQS
{
if(_blockingEntities.add(this))
{
+ messageWithSubject(ChannelMessages.FLOW_ENFORCED("** All Queues **"));
+
for(LinkEndpoint endpoint : _receivingLinkMap.values())
{
endpoint.setStopped(true);
@@ -1392,6 +1406,10 @@ public class Session_1_0 implements AMQS
{
if(_blockingEntities.remove(this))
{
+ if(_blockingEntities.isEmpty())
+ {
+ messageWithSubject(ChannelMessages.FLOW_REMOVED());
+ }
for(ReceivingLinkEndpoint endpoint : _receivingLinkMap.values())
{
ReceivingLink_1_0 link = (ReceivingLink_1_0) endpoint.getLink();
@@ -1409,6 +1427,16 @@ public class Session_1_0 implements AMQS
return !_blockingEntities.isEmpty();
}
+ private void messageWithSubject(final LogMessage operationalLogMessage)
+ {
+ getEventLogger().message(_logSubject, operationalLogMessage);
+ }
+
+ public EventLogger getEventLogger()
+ {
+ return getConnection().getEventLogger();
+ }
+
@Override
public Object getConnectionReference()
{
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org