QPID-2002: Added testing of Channel Logging

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk/qpid@800368 13f79535-47bb-0310-9956-ffa450edef68

1 files changed, 27 insertions, 1 deletions

diff --git a/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java b/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java
index aa390d6c26..2a46ee53b4 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java
@@ -57,6 +57,12 @@ import org.apache.qpid.server.store.StoreContext;
 import org.apache.qpid.server.txn.LocalTransactionalContext;
 import org.apache.qpid.server.txn.NonTransactionalContext;
 import org.apache.qpid.server.txn.TransactionalContext;
+import org.apache.qpid.server.logging.LogActor;
+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.logging.actors.AMQPChannelActor;
+import org.apache.qpid.server.logging.actors.CurrentActor;
 
 public class AMQChannel
 {
@@ -111,13 +117,22 @@ public class AMQChannel
 
     // Why do we need this reference ? - ritchiem
     private final AMQProtocolSession _session;
-    private boolean _closing; 
+    private boolean _closing;
+
+    private LogActor _actor;
+    private LogSubject _logSubject;
 
     public AMQChannel(AMQProtocolSession session, int channelId, MessageStore messageStore)
             throws AMQException
     {
         _session = session;
         _channelId = channelId;
+
+        _actor = new AMQPChannelActor(this, session.getLogActor().getRootMessageLogger());
+        _logSubject = new ChannelLogSubject(this);
+        
+        _actor.message(ChannelMessages.CHN_1001());
+
         _storeContext = new StoreContext("Session: " + session.getClientIdentifier() + "; channel: " + channelId);
 
 
@@ -371,6 +386,8 @@ public class AMQChannel
     private void setClosing(boolean closing)
     {
         _closing = closing;
+
+        CurrentActor.get().message(_logSubject, ChannelMessages.CHN_1003());
     }
 
     private void unsubscribeAllConsumers() throws AMQException
@@ -774,6 +791,8 @@ public class AMQChannel
         boolean wasSuspended = _suspended.getAndSet(suspended);
         if (wasSuspended != suspended)
         {
+            _actor.message(_logSubject, ChannelMessages.CHN_1002(suspended ? "Stopped" : "Started"));
+
             if (wasSuspended)
             {
                 // may need to deliver queued messages
@@ -865,6 +884,8 @@ public class AMQChannel
 
     public void setCredit(final long prefetchSize, final int prefetchCount)
     {
+        //fixme
+//        _actor.message(ChannelMessages.CHN_100X(prefetchSize, prefetchCount);
         _creditManager.setCreditLimits(prefetchSize, prefetchCount);
     }
 
@@ -906,4 +927,9 @@ public class AMQChannel
     {
         return _recordDeliveryMethod;
     }
+
+    public LogActor getLogActor()
+    {
+        return _actor;
+    }
 }