1 files changed, 264 insertions, 0 deletions
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);
+    }
+}