fix for: https://issues.apache.org/jira/browse/AMQ-1907

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

activemq-core/src/main/java/org/apache/activemq/broker/jmx/SubscriptionView.java

@@ -16,14 +16,22 @@
  */
 package org.apache.activemq.broker.jmx;
 
-import javax.jms.InvalidSelectorException;
+import java.io.IOException;
+import java.util.Set;
 
+import javax.jms.InvalidSelectorException;
+import javax.management.ObjectName;
+
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.broker.ConnectionContext;
 import org.apache.activemq.broker.region.Subscription;
 import org.apache.activemq.command.ActiveMQDestination;
 import org.apache.activemq.command.ActiveMQQueue;
 import org.apache.activemq.command.ActiveMQTopic;
 import org.apache.activemq.command.ConsumerInfo;
 import org.apache.activemq.filter.DestinationFilter;
+import org.apache.activemq.util.IOExceptionSupport;
+import org.apache.activemq.util.JMXSupport;
 
 /**
  *
@@ -52,6 +60,44 @@ public class SubscriptionView implements SubscriptionViewMBean {
         return clientId;
     }
 
+    /**
+     * @returns the ObjectName of the Connection that created this subscription
+     */
+    public ObjectName getConnection() {
+        ObjectName result = null;
+
+        if (clientId != null && subscription != null) {
+            ConnectionContext ctx = subscription.getContext();
+            if (ctx != null && ctx.getBroker() != null && ctx.getBroker().getBrokerService() != null) {
+                BrokerService service = ctx.getBroker().getBrokerService();
+                ManagementContext managementCtx = service.getManagementContext();
+                if (managementCtx != null) {
+
+                    try {
+                        ObjectName query = createConnectionQueury(managementCtx, service.getBrokerName());
+                        Set<ObjectName> names = managementCtx.queryNames(query, null);
+                        if (names.size() == 1) {
+                            result = names.iterator().next();
+                        }
+                    } catch (Exception e) {
+                    }
+                }
+            }
+        }
+        return result;
+    }
+
+    private ObjectName createConnectionQueury(ManagementContext ctx, String brokerName) throws IOException {
+        try {
+            return new ObjectName(ctx.getJmxDomainName() + ":" + "BrokerName="
+                                  + JMXSupport.encodeObjectNamePart(brokerName) + ","
+                                  + "Type=Connection," + "ConnectorName=*,"
+                                  + "Connection=" + JMXSupport.encodeObjectNamePart(clientId));
+        } catch (Throwable e) {
+            throw IOExceptionSupport.create(e);
+        }
+    }
+
     /**
      * @return the id of the Connection the Subscription is on
      */

activemq-core/src/main/java/org/apache/activemq/broker/jmx/SubscriptionViewMBean.java

@@ -17,6 +17,7 @@
 package org.apache.activemq.broker.jmx;
 
 import javax.jms.InvalidSelectorException;
+import javax.management.ObjectName;
 
 /**
  *
@@ -213,4 +214,14 @@ public interface SubscriptionViewMBean {
      */
     @MBeanInfo("User Name used to authorize creation of this Subscription")
     String getUserName();
+
+    /**
+     * Returns the ObjectName of the Connection that created this Subscription.
+     * This value can be null if for instance this is an off-line durable subscription.
+     *
+     * @return the name of the Connection that created this Subscription.
+     */
+    @MBeanInfo("ObjectName of the Connection that created this Subscription")
+    ObjectName getConnection();
+
 }

activemq-core/src/test/java/org/apache/activemq/broker/jmx/MBeanTest.java

@@ -894,6 +894,37 @@ public class MBeanTest extends EmbeddedBrokerTestSupport {
         assertTrue("dest has some memory usage", queue.getMemoryPercentUsage() > 0);
     }
 
+    public void testSubscriptionViewToConnectionMBean() throws Exception {
+
+        connection = connectionFactory.createConnection("admin", "admin");
+        connection.setClientID("MBeanTest");
+        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        Destination queue = session.createQueue(getDestinationString() + ".Queue");
+        @SuppressWarnings("unused")
+        MessageConsumer queueConsumer = session.createConsumer(queue);
+
+        ObjectName brokerName = assertRegisteredObjectName(domain + ":Type=Broker,BrokerName=localhost");
+        BrokerViewMBean broker = (BrokerViewMBean)MBeanServerInvocationHandler.newProxyInstance(mbeanServer, brokerName, BrokerViewMBean.class, true);
+
+        Thread.sleep(100);
+
+        assertTrue(broker.getQueueSubscribers().length == 1);
+
+        ObjectName subscriptionName = broker.getQueueSubscribers()[0];
+
+        SubscriptionViewMBean subscriberView =
+            (SubscriptionViewMBean)MBeanServerInvocationHandler.newProxyInstance(
+                    mbeanServer, subscriptionName, SubscriptionViewMBean.class, true);
+        assertNotNull(subscriberView);
+
+        ObjectName connectionName = subscriberView.getConnection();
+        assertNotNull(connectionName);
+        ConnectionViewMBean connectionView =
+            (ConnectionViewMBean)MBeanServerInvocationHandler.newProxyInstance(
+                    mbeanServer, connectionName, ConnectionViewMBean.class, true);
+        assertNotNull(connectionView);
+    }
+
     public void testUserNamePopulated() throws Exception {
         doTestUserNameInMBeans(true);
     }