QPID-2243: 0-10 protocol connections do not have a matching JMX MBean to allow management.

Applied patch from Andrew MacBean <andymacbean@gmail.com> and Oleksandr Rudyy<orudyy@gmail.com>

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

6 files changed, 436 insertions, 91 deletions

diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/management/AbstractAMQManagedConnectionObject.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/management/AbstractAMQManagedConnectionObject.java
new file mode 100644
index 0000000000..68350a1632
--- /dev/null
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/management/AbstractAMQManagedConnectionObject.java
@@ -0,0 +1,71 @@
+package org.apache.qpid.server.management;
+
+import javax.management.Notification;
+
+import javax.management.JMException;
+import javax.management.MBeanNotificationInfo;
+import javax.management.NotCompliantMBeanException;
+import javax.management.ObjectName;
+import javax.management.monitor.MonitorNotification;
+import javax.management.openmbean.CompositeType;
+import javax.management.openmbean.OpenType;
+import javax.management.openmbean.SimpleType;
+import javax.management.openmbean.TabularType;
+import org.apache.qpid.management.common.mbeans.ManagedConnection;
+
+public abstract class AbstractAMQManagedConnectionObject extends AMQManagedObject implements ManagedConnection
+{
+    protected final String _name;
+
+    protected static final OpenType[] _channelAttributeTypes = { SimpleType.INTEGER, SimpleType.BOOLEAN, SimpleType.STRING, SimpleType.INTEGER, SimpleType.BOOLEAN };
+    protected static final CompositeType _channelType;
+    protected static final TabularType _channelsType;
+
+    protected static final String BROKER_MANAGEMENT_CONSOLE_HAS_CLOSED_THE_CONNECTION_STR =
+                                    "Broker Management Console has closed the connection.";
+
+    static
+    {
+        try
+        {
+            _channelType = new CompositeType("Channel", "Channel Details", COMPOSITE_ITEM_NAMES_DESC.toArray(new String[COMPOSITE_ITEM_NAMES_DESC.size()]),
+                            COMPOSITE_ITEM_NAMES_DESC.toArray(new String[COMPOSITE_ITEM_NAMES_DESC.size()]), _channelAttributeTypes);
+            _channelsType = new TabularType("Channels", "Channels", _channelType, (String[]) TABULAR_UNIQUE_INDEX.toArray(new String[TABULAR_UNIQUE_INDEX.size()]));
+        }
+        catch (JMException ex)
+        {
+            // This is not expected to ever occur.
+            throw new RuntimeException("Got JMException in static initializer.", ex);
+        }
+    }
+
+    protected AbstractAMQManagedConnectionObject(final String remoteAddress) throws NotCompliantMBeanException
+    {
+        super(ManagedConnection.class, ManagedConnection.TYPE);
+        _name = "anonymous".equals(remoteAddress) ? (remoteAddress + hashCode()) : remoteAddress;
+    }
+
+    @Override
+    public String getObjectInstanceName()
+    {
+        return ObjectName.quote(_name);
+    }
+
+    public void notifyClients(String notificationMsg)
+    {
+        final Notification n = new Notification(MonitorNotification.THRESHOLD_VALUE_EXCEEDED, this, ++_notificationSequenceNumber,
+                                                System.currentTimeMillis(), notificationMsg);
+        _broadcaster.sendNotification(n);
+    }
+
+    @Override
+    public MBeanNotificationInfo[] getNotificationInfo()
+    {
+        String[] notificationTypes = new String[] { MonitorNotification.THRESHOLD_VALUE_EXCEEDED };
+        String name = MonitorNotification.class.getName();
+        String description = "Channel count has reached threshold value";
+        MBeanNotificationInfo info1 = new MBeanNotificationInfo(notificationTypes, name, description);
+
+        return new MBeanNotificationInfo[] { info1 };
+    }
+}
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSessionMBean.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSessionMBean.java
index 16d99de492..8d39420631 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSessionMBean.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSessionMBean.java
@@ -39,89 +39,44 @@ package org.apache.qpid.server.protocol;
 
 import java.util.Date;
 import java.util.List;
-
 import javax.management.JMException;
 import javax.management.MBeanException;
-import javax.management.MBeanNotificationInfo;
 import javax.management.NotCompliantMBeanException;
-import javax.management.Notification;
-import javax.management.ObjectName;
-import javax.management.monitor.MonitorNotification;
 import javax.management.openmbean.CompositeData;
 import javax.management.openmbean.CompositeDataSupport;
-import javax.management.openmbean.CompositeType;
 import javax.management.openmbean.OpenDataException;
-import javax.management.openmbean.OpenType;
-import javax.management.openmbean.SimpleType;
 import javax.management.openmbean.TabularData;
 import javax.management.openmbean.TabularDataSupport;
-import javax.management.openmbean.TabularType;
-
 import org.apache.qpid.AMQException;
 import org.apache.qpid.framing.AMQShortString;
 import org.apache.qpid.framing.ConnectionCloseBody;
 import org.apache.qpid.framing.MethodRegistry;
-import org.apache.qpid.management.common.mbeans.ManagedConnection;
 import org.apache.qpid.management.common.mbeans.annotations.MBeanConstructor;
 import org.apache.qpid.management.common.mbeans.annotations.MBeanDescription;
 import org.apache.qpid.protocol.AMQConstant;
 import org.apache.qpid.server.AMQChannel;
 import org.apache.qpid.server.logging.actors.CurrentActor;
 import org.apache.qpid.server.logging.actors.ManagementActor;
-import org.apache.qpid.server.management.AMQManagedObject;
+import org.apache.qpid.server.management.AbstractAMQManagedConnectionObject;
 import org.apache.qpid.server.management.ManagedObject;
 
 /**
  * This MBean class implements the management interface. In order to make more attributes, operations and notifications
  * available over JMX simply augment the ManagedConnection interface and add the appropriate implementation here.
  */
-@MBeanDescription("Management Bean for an AMQ Broker Connection")
-public class AMQProtocolSessionMBean extends AMQManagedObject implements ManagedConnection
+@MBeanDescription("Management Bean for an AMQ Broker 0-9-1/0-9/0-8 Connections")
+public class AMQProtocolSessionMBean extends AbstractAMQManagedConnectionObject
 {
     private AMQProtocolSession _protocolSession = null;
-    private String _name = null;
 
-    // openmbean data types for representing the channel attributes
-
-    private static final OpenType[] _channelAttributeTypes =
-        { SimpleType.INTEGER, SimpleType.BOOLEAN, SimpleType.STRING, SimpleType.INTEGER, SimpleType.BOOLEAN };
-    private static CompositeType _channelType = null; // represents the data type for channel data
-    private static TabularType _channelsType = null; // Data type for list of channels type
     private static final AMQShortString BROKER_MANAGEMENT_CONSOLE_HAS_CLOSED_THE_CONNECTION =
-        new AMQShortString("Broker Management Console has closed the connection.");
+                                        new AMQShortString(BROKER_MANAGEMENT_CONSOLE_HAS_CLOSED_THE_CONNECTION_STR);
 
-    @MBeanConstructor("Creates an MBean exposing an AMQ Broker Connection")
+    @MBeanConstructor("Creates an MBean exposing an AMQ Broker 0-9-1/0-9/0-8 Connection")
     public AMQProtocolSessionMBean(AMQProtocolSession amqProtocolSession) throws NotCompliantMBeanException, OpenDataException
     {
-        super(ManagedConnection.class, ManagedConnection.TYPE);
+        super(amqProtocolSession.getRemoteAddress().toString());
         _protocolSession = amqProtocolSession;
-        String remote = getRemoteAddress();
-        _name = "anonymous".equals(remote) ? (remote + hashCode()) : remote;
-        init();
-    }
-
-    static
-    {
-        try
-        {
-            init();
-        }
-        catch (JMException ex)
-        {
-            // This is not expected to ever occur.
-            throw new RuntimeException("Got JMException in static initializer.", ex);
-        }
-    }
-
-    /**
-     * initialises the openmbean data types
-     */
-    private static void init() throws OpenDataException
-    {
-        _channelType =
-            new CompositeType("Channel", "Channel Details", COMPOSITE_ITEM_NAMES_DESC.toArray(new String[COMPOSITE_ITEM_NAMES_DESC.size()]),
-                    COMPOSITE_ITEM_NAMES_DESC.toArray(new String[COMPOSITE_ITEM_NAMES_DESC.size()]), _channelAttributeTypes);
-        _channelsType = new TabularType("Channels", "Channels", _channelType, TABULAR_UNIQUE_INDEX.toArray(new String[TABULAR_UNIQUE_INDEX.size()]));
     }
 
     public String getClientId()
@@ -169,16 +124,6 @@ public class AMQProtocolSessionMBean extends AMQManagedObject implements Managed
         return _protocolSession.getMaximumNumberOfChannels();
     }
 
-    public void setMaximumNumberOfChannels(Long value)
-    {
-        _protocolSession.setMaximumNumberOfChannels(value);
-    }
-
-    public String getObjectInstanceName()
-    {
-        return ObjectName.quote(_name);
-    }
-
     /**
      * commits transactions for a transactional channel
      *
@@ -321,25 +266,6 @@ public class AMQProtocolSessionMBean extends AMQManagedObject implements Managed
         }
     }
 
-    @Override
-    public MBeanNotificationInfo[] getNotificationInfo()
-    {
-        String[] notificationTypes = new String[] { MonitorNotification.THRESHOLD_VALUE_EXCEEDED };
-        String name = MonitorNotification.class.getName();
-        String description = "Channel count has reached threshold value";
-        MBeanNotificationInfo info1 = new MBeanNotificationInfo(notificationTypes, name, description);
-
-        return new MBeanNotificationInfo[] { info1 };
-    }
-
-    public void notifyClients(String notificationMsg)
-    {
-        Notification n =
-            new Notification(MonitorNotification.THRESHOLD_VALUE_EXCEEDED, this, ++_notificationSequenceNumber,
-                System.currentTimeMillis(), notificationMsg);
-        _broadcaster.sendNotification(n);
-    }
-
     public void resetStatistics() throws Exception
     {
         _protocolSession.resetStatistics();
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnection.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnection.java
index e18b453db3..922531a271 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnection.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnection.java
@@ -24,6 +24,14 @@ import static org.apache.qpid.server.logging.subjects.LogSubjectFormat.CONNECTIO
 import static org.apache.qpid.server.logging.subjects.LogSubjectFormat.SOCKET_FORMAT;
 import static org.apache.qpid.server.logging.subjects.LogSubjectFormat.USER_FORMAT;
 
+import java.util.concurrent.atomic.AtomicLong;
+
+import javax.management.JMException;
+
+import org.apache.qpid.server.management.ManagedObject;
+
+import org.apache.qpid.server.management.Managable;
+
 import java.security.Principal;
 import java.text.MessageFormat;
 import java.util.ArrayList;
@@ -55,7 +63,7 @@ import org.apache.qpid.transport.Method;
 import org.apache.qpid.transport.ProtocolEvent;
 import org.apache.qpid.transport.Session;
 
-public class ServerConnection extends Connection implements AMQConnectionModel, LogSubject, AuthorizationHolder
+public class ServerConnection extends Connection implements Managable, AMQConnectionModel, LogSubject, AuthorizationHolder
 {
     private ConnectionConfig _config;
     private Runnable _onOpenTask;
@@ -67,6 +75,10 @@ public class ServerConnection extends Connection implements AMQConnectionModel,
     private boolean _statisticsEnabled = false;
     private StatisticsCounter _messagesDelivered, _dataDelivered, _messagesReceived, _dataReceived;
     private final long _connectionId;
+
+    private ServerConnectionMBean _mBean;
+    private VirtualHost _virtualHost;
+    private AtomicLong _lastIoTime = new AtomicLong();
     
     public ServerConnection(final long connectionId)
     {
@@ -133,9 +145,6 @@ public class ServerConnection extends Connection implements AMQConnectionModel,
         super.setConnectionDelegate(delegate);
     }
 
-    private VirtualHost _virtualHost;
-
-
     public VirtualHost getVirtualHost()
     {
         return _virtualHost;
@@ -144,8 +153,18 @@ public class ServerConnection extends Connection implements AMQConnectionModel,
     public void setVirtualHost(VirtualHost virtualHost)
     {
         _virtualHost = virtualHost;
-        
+
         initialiseStatistics();
+
+        try
+        {
+            _mBean = new ServerConnectionMBean(this);
+            _mBean.register();
+        }
+        catch (JMException jme)
+        {
+            log.error("Unable to create mBean for ServerConnection",jme);
+        }
     }
 
     public void setConnectionConfig(final ConnectionConfig config)
@@ -190,6 +209,7 @@ public class ServerConnection extends Connection implements AMQConnectionModel,
     @Override
     public void received(ProtocolEvent event)
     {
+        _lastIoTime.set(System.currentTimeMillis());
         if (event.isConnectionControl())
         {
             CurrentActor.set(_actor);
@@ -260,6 +280,11 @@ public class ServerConnection extends Connection implements AMQConnectionModel,
     public void close(AMQConstant cause, String message) throws AMQException
     {
         closeSubscriptions();
+        if (_mBean != null)
+        {
+            _mBean.unregister();
+            _mBean = null;
+        }
         ConnectionCloseCode replyCode = ConnectionCloseCode.NORMAL;
         try
         {
@@ -405,6 +430,11 @@ public class ServerConnection extends Connection implements AMQConnectionModel,
     public void closed()
     {
         closeSubscriptions();
+        if (_mBean != null)
+        {
+            _mBean.unregister();
+            _mBean = null;
+        }
         super.closed();
     }
 
@@ -416,4 +446,30 @@ public class ServerConnection extends Connection implements AMQConnectionModel,
         }
     }
 
+    @Override
+    public ManagedObject getManagedObject()
+    {
+        return _mBean;
+    }
+
+    @Override
+    public void send(ProtocolEvent event)
+    {
+        _lastIoTime.set(System.currentTimeMillis());
+        super.send(event);
+    }
+
+    public AtomicLong getLastIoTime()
+    {
+        return _lastIoTime;
+    }
+
+    void checkForNotification()
+    {
+        int channelsCount = getSessionModels().size();
+        if (_mBean != null && channelsCount >= getConnectionDelegate().getChannelMax())
+        {
+            _mBean.notifyClients("Channel count (" + channelsCount + ") has reached the threshold value");
+        }
+    }
 }
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionDelegate.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionDelegate.java
index 8d6e0e0d80..66ed6f1e62 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionDelegate.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionDelegate.java
@@ -28,10 +28,8 @@ import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.StringTokenizer;
-
 import javax.security.sasl.SaslException;
 import javax.security.sasl.SaslServer;
-
 import org.apache.qpid.common.ServerPropertyNames;
 import org.apache.qpid.protocol.ProtocolEngine;
 import org.apache.qpid.server.configuration.BrokerConfig;
@@ -49,6 +47,7 @@ import org.apache.qpid.transport.ConnectionClose;
 import org.apache.qpid.transport.ConnectionCloseCode;
 import org.apache.qpid.transport.ConnectionOpen;
 import org.apache.qpid.transport.ConnectionOpenOk;
+import org.apache.qpid.transport.ConnectionStartOk;
 import org.apache.qpid.transport.ConnectionTuneOk;
 import org.apache.qpid.transport.ServerDelegate;
 import org.apache.qpid.transport.Session;
@@ -62,6 +61,8 @@ public class ServerConnectionDelegate extends ServerDelegate
 {
     private final String _localFQDN;
     private final IApplicationRegistry _appRegistry;
+    private int _maxNoOfChannels;
+    private Map<String,Object> _clientProperties;
 
     public ServerConnectionDelegate(IApplicationRegistry appRegistry, String localFQDN)
     {
@@ -77,6 +78,7 @@ public class ServerConnectionDelegate extends ServerDelegate
         
         _appRegistry = appRegistry;
         _localFQDN = localFQDN;
+        _maxNoOfChannels = ApplicationRegistry.getInstance().getConfiguration().getMaxChannelCount();
     }
 
     private static Map<String, Object> createConnectionProperties(final BrokerConfig brokerConfig)
@@ -154,7 +156,7 @@ public class ServerConnectionDelegate extends ServerDelegate
     public void connectionOpen(Connection conn, ConnectionOpen open)
     {
         final ServerConnection sconn = (ServerConnection) conn;
-        
+
         VirtualHost vhost;
         String vhostName;
         if(open.hasVirtualHost())
@@ -222,7 +224,12 @@ public class ServerConnectionDelegate extends ServerDelegate
     @Override
     protected int getChannelMax()
     {
-        return ApplicationRegistry.getInstance().getConfiguration().getMaxChannelCount();
+        return _maxNoOfChannels;
+    }
+
+    protected void setChannelMax(int channelMax)
+    {
+        _maxNoOfChannels = channelMax;
     }
 
     @Override public void sessionDetach(Connection conn, SessionDetach dtc)
@@ -253,6 +260,7 @@ public class ServerConnectionDelegate extends ServerDelegate
         {
             ssn = sessionAttachImpl(conn, atc);
             conn.registerSession(ssn);
+            ((ServerConnection)conn).checkForNotification();
         }
         else
         {
@@ -279,4 +287,16 @@ public class ServerConnectionDelegate extends ServerDelegate
         }
         return true;
     }
+
+    @Override
+    public void connectionStartOk(Connection conn, ConnectionStartOk ok)
+    {
+        _clientProperties = ok.getClientProperties();
+        super.connectionStartOk(conn, ok);
+    }
+
+    public Map<String,Object> getClientProperties()
+    {
+        return _clientProperties;
+    }
 }
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionMBean.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionMBean.java
new file mode 100644
index 0000000000..17c7bed601
--- /dev/null
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionMBean.java
@@ -0,0 +1,264 @@
+/*
+ *
+ * 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.transport;
+
+import java.io.IOException;
+import java.util.Date;
+import java.util.List;
+import javax.management.JMException;
+import javax.management.NotCompliantMBeanException;
+import javax.management.openmbean.CompositeData;
+import javax.management.openmbean.CompositeDataSupport;
+import javax.management.openmbean.TabularData;
+import javax.management.openmbean.TabularDataSupport;
+
+import org.apache.qpid.common.ClientProperties;
+import org.apache.qpid.management.common.mbeans.annotations.MBeanConstructor;
+import org.apache.qpid.management.common.mbeans.annotations.MBeanDescription;
+import org.apache.qpid.server.logging.actors.CurrentActor;
+import org.apache.qpid.server.logging.actors.ManagementActor;
+import org.apache.qpid.server.management.AbstractAMQManagedConnectionObject;
+import org.apache.qpid.server.management.ManagedObject;
+import org.apache.qpid.server.protocol.AMQSessionModel;
+
+/**
+ * This MBean class implements the management interface. In order to make more attributes, operations and notifications
+ * available over JMX simply augment the ManagedConnection interface and add the appropriate implementation here.
+ */
+@MBeanDescription("Management Bean for an AMQ Broker 0-10 Connection")
+public class ServerConnectionMBean extends AbstractAMQManagedConnectionObject
+{
+    private final ServerConnection _serverConnection;
+
+    @MBeanConstructor("Creates an MBean exposing an AMQ Broker 0-10 Connection")
+    protected ServerConnectionMBean(final ServerConnection serverConnection) throws NotCompliantMBeanException
+    {
+        super(serverConnection.getConfig().getAddress());
+        _serverConnection = serverConnection;
+    }
+
+    @Override
+    public ManagedObject getParentObject()
+    {
+        return _serverConnection.getVirtualHost().getManagedObject();
+    }
+
+    @Override
+    public String getClientId()
+    {
+        return _serverConnection.getClientId();
+    }
+
+    @Override
+    public String getAuthorizedId()
+    {
+        return _serverConnection.getAuthorizedPrincipal().getName();
+    }
+
+    @Override
+    public String getVersion()
+    {
+        return String.valueOf(_serverConnection.getConnectionDelegate().getClientProperties().get(ClientProperties.version.toString()));
+    }
+
+    @Override
+    public String getRemoteAddress()
+    {
+        return _serverConnection.getConfig().getAddress();
+    }
+
+    @Override
+    public Date getLastIoTime()
+    {
+        return new Date(_serverConnection.getLastIoTime().longValue());
+    }
+
+    @Override
+    public Long getMaximumNumberOfChannels()
+    {
+        return (long) _serverConnection.getConnectionDelegate().getChannelMax();
+    }
+
+    @Override
+    public TabularData channels() throws IOException, JMException
+    {
+        final TabularDataSupport channelsList = new TabularDataSupport(_channelsType);
+        final List<AMQSessionModel> list = _serverConnection.getSessionModels();
+
+        for (final AMQSessionModel channel : list)
+        {
+            final ServerSession session = (ServerSession)channel;
+            Object[] itemValues =
+                {
+                    session.getChannel(),
+                    session.isTransactional(),
+                    null,
+                    session.getUnacknowledgedMessageCount(),
+                    session.getBlocking()
+                };
+
+            final CompositeData channelData = new CompositeDataSupport(_channelType,
+                    COMPOSITE_ITEM_NAMES_DESC.toArray(new String[COMPOSITE_ITEM_NAMES_DESC.size()]), itemValues);
+            channelsList.put(channelData);
+        }
+        return channelsList;
+    }
+
+    @Override
+    public void commitTransactions(int channelId) throws JMException
+    {
+        final ServerSession session = (ServerSession)_serverConnection.getSession(channelId);
+        if (session == null)
+        {
+            throw new JMException("The channel (channel Id = " + channelId + ") does not exist");
+        }
+        else if (session.isTransactional())
+        {
+            CurrentActor.set(new ManagementActor(_logActor.getRootMessageLogger()));
+            try
+            {
+                session.commit();
+            }
+            finally
+            {
+                CurrentActor.remove();
+            }
+        }
+    }
+
+    @Override
+    public void rollbackTransactions(int channelId) throws JMException
+    {
+        final ServerSession session = (ServerSession)_serverConnection.getSession(channelId);
+        if (session == null)
+        {
+            throw new JMException("The channel (channel Id = " + channelId + ") does not exist");
+        }
+        else if (session.isTransactional())
+        {
+            CurrentActor.set(new ManagementActor(_logActor.getRootMessageLogger()));
+            try
+            {
+                session.rollback();
+            }
+            finally
+            {
+                CurrentActor.remove();
+            }
+        }
+    }
+
+    @Override
+    public void closeConnection() throws Exception
+    {
+        _serverConnection.mgmtClose();
+    }
+
+    @Override
+    public void resetStatistics() throws Exception
+    {
+        _serverConnection.resetStatistics();
+    }
+
+    @Override
+    public double getPeakMessageDeliveryRate()
+    {
+        return _serverConnection.getMessageDeliveryStatistics().getPeak();
+    }
+
+    @Override
+    public double getPeakDataDeliveryRate()
+    {
+        return _serverConnection.getDataDeliveryStatistics().getPeak();
+    }
+
+    @Override
+    public double getMessageDeliveryRate()
+    {
+        return _serverConnection.getMessageDeliveryStatistics().getRate();
+    }
+
+    @Override
+    public double getDataDeliveryRate()
+    {
+        return _serverConnection.getDataDeliveryStatistics().getRate();
+    }
+
+    @Override
+    public long getTotalMessagesDelivered()
+    {
+        return _serverConnection.getMessageDeliveryStatistics().getTotal();
+    }
+
+    @Override
+    public long getTotalDataDelivered()
+    {
+        return _serverConnection.getDataDeliveryStatistics().getTotal();
+    }
+
+    @Override
+    public double getPeakMessageReceiptRate()
+    {
+        return _serverConnection.getMessageReceiptStatistics().getPeak();
+    }
+
+    @Override
+    public double getPeakDataReceiptRate()
+    {
+        return _serverConnection.getDataReceiptStatistics().getPeak();
+    }
+
+    @Override
+    public double getMessageReceiptRate()
+    {
+        return _serverConnection.getMessageReceiptStatistics().getRate();
+    }
+
+    @Override
+    public double getDataReceiptRate()
+    {
+        return _serverConnection.getDataReceiptStatistics().getRate();
+    }
+
+    @Override
+    public long getTotalMessagesReceived()
+    {
+        return _serverConnection.getMessageReceiptStatistics().getTotal();
+    }
+
+    @Override
+    public long getTotalDataReceived()
+    {
+        return _serverConnection.getDataReceiptStatistics().getTotal();
+    }
+
+    @Override
+    public boolean isStatisticsEnabled()
+    {
+        return _serverConnection.isStatisticsEnabled();
+    }
+
+    @Override
+    public void setStatisticsEnabled(boolean enabled)
+    {
+        _serverConnection.setStatisticsEnabled(enabled);
+    }
+}
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSession.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSession.java
index 337d1f02cc..dfad097dbc 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSession.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSession.java
@@ -37,9 +37,7 @@ import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentSkipListMap;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.atomic.AtomicLong;
-
 import javax.security.auth.Subject;
-
 import org.apache.qpid.AMQException;
 import org.apache.qpid.protocol.AMQConstant;
 import org.apache.qpid.protocol.ProtocolEngine;
@@ -698,4 +696,14 @@ public class ServerSession extends Session implements AuthorizationHolder, Sessi
             unregister(subscription_0_10);
         }
     }
+
+    public int getUnacknowledgedMessageCount()
+    {
+        return _messageDispositionListenerMap.size();
+    }
+
+    public boolean getBlocking()
+    {
+        return false; //TODO: Blocking not implemented on 0-10 yet.
+    }
 }