activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From cshan...@apache.org
Subject activemq git commit: https://issues.apache.org/jira/browse/AMQ-6129
Date Fri, 15 Jan 2016 14:54:25 GMT
Repository: activemq
Updated Branches:
  refs/heads/master 2b84cd60b -> 10c998b0b


https://issues.apache.org/jira/browse/AMQ-6129

Adding NetworkBridgeStatistics and also a received count for bridges
when they are in duplex mode.


Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/10c998b0
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/10c998b0
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/10c998b0

Branch: refs/heads/master
Commit: 10c998b0bc9728276a738ed24d20c6fc82c6365a
Parents: 2b84cd6
Author: Christopher L. Shannon (cshannon) <christopher.l.shannon@gmail.com>
Authored: Fri Jan 15 14:39:49 2016 +0000
Committer: Christopher L. Shannon (cshannon) <christopher.l.shannon@gmail.com>
Committed: Fri Jan 15 14:54:04 2016 +0000

----------------------------------------------------------------------
 .../activemq/broker/jmx/NetworkBridgeView.java  |  16 +++
 .../broker/jmx/NetworkBridgeViewMBean.java      |   2 +
 .../network/DemandForwardingBridgeSupport.java  |  27 +++--
 .../apache/activemq/network/NetworkBridge.java  |   5 +
 .../network/NetworkBridgeStatistics.java        | 102 +++++++++++++++++++
 .../org/apache/activemq/bugs/AMQ4160Test.java   |   8 ++
 .../activemq/network/DuplexNetworkTest.java     |  15 +++
 .../activemq/network/SimpleNetworkTest.java     |  26 ++++-
 8 files changed, 190 insertions(+), 11 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/activemq/blob/10c998b0/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkBridgeView.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkBridgeView.java
b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkBridgeView.java
index c557754..756ffbe 100644
--- a/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkBridgeView.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkBridgeView.java
@@ -31,43 +31,58 @@ public class NetworkBridgeView implements NetworkBridgeViewMBean {
         this.bridge = bridge;
     }
 
+    @Override
     public void start() throws Exception {
         bridge.start();
     }
 
+    @Override
     public void stop() throws Exception {
         bridge.stop();
     }
 
+    @Override
     public String getLocalAddress() {
         return bridge.getLocalAddress();
     }
 
+    @Override
     public String getRemoteAddress() {
         return bridge.getRemoteAddress();
     }
 
+    @Override
     public String getRemoteBrokerName() {
         return bridge.getRemoteBrokerName();
     }
 
 
+    @Override
     public String getRemoteBrokerId() {
         return bridge.getRemoteBrokerId();
     }
 
+    @Override
     public String getLocalBrokerName() {
         return bridge.getLocalBrokerName();
     }
 
+    @Override
     public long getEnqueueCounter() {
         return bridge.getEnqueueCounter();
     }
 
+    @Override
     public long getDequeueCounter() {
         return bridge.getDequeueCounter();
     }
 
+    @Override
+    public long getReceivedCounter() {
+        return bridge.getNetworkBridgeStatistics().getReceivedCount().getCount();
+    }
+
+    @Override
     public boolean isCreatedByDuplex() {
         return createByDuplex;
     }
@@ -76,6 +91,7 @@ public class NetworkBridgeView implements NetworkBridgeViewMBean {
         this.createByDuplex = createByDuplex;
     }
 
+    @Override
     public void resetStats(){
         bridge.resetStats();
         for (NetworkDestinationView networkDestinationView:networkDestinationViewList){

http://git-wip-us.apache.org/repos/asf/activemq/blob/10c998b0/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkBridgeViewMBean.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkBridgeViewMBean.java
b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkBridgeViewMBean.java
index 639a81a..82fc9ca 100644
--- a/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkBridgeViewMBean.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkBridgeViewMBean.java
@@ -34,6 +34,8 @@ public interface NetworkBridgeViewMBean extends Service {
 
     long getDequeueCounter();
 
+    long getReceivedCounter();
+
     boolean isCreatedByDuplex();
 
     void resetStats();

http://git-wip-us.apache.org/repos/asf/activemq/blob/10c998b0/activemq-broker/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java
b/activemq-broker/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java
index 112f5ff..7f3eeb4 100644
--- a/activemq-broker/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java
@@ -34,7 +34,6 @@ import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicLong;
 
 import javax.management.ObjectName;
 
@@ -139,8 +138,7 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge,
Br
     protected final BrokerId remoteBrokerPath[] = new BrokerId[]{null};
     protected BrokerId remoteBrokerId;
 
-    final AtomicLong enqueueCounter = new AtomicLong();
-    final AtomicLong dequeueCounter = new AtomicLong();
+    protected final NetworkBridgeStatistics networkBridgeStatistics = new NetworkBridgeStatistics();
 
     private NetworkBridgeListener networkBridgeListener;
     private boolean createdByDuplex;
@@ -181,6 +179,8 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge,
Br
                 throw new IllegalArgumentException("BrokerService is null on " + this);
             }
 
+            networkBridgeStatistics.setEnabled(brokerService.isEnableStatistics());
+
             if (isDuplex()) {
                 duplexInboundLocalBroker = NetworkBridgeFactory.createLocalTransport(brokerService.getBroker());
                 duplexInboundLocalBroker.setTransportListener(new DefaultTransportListener()
{
@@ -640,6 +640,8 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge,
Br
                                                     Response reply = resp.getResult();
                                                     reply.setCorrelationId(correlationId);
                                                     remoteBroker.oneway(reply);
+                                                    //increment counter when messages are
received in duplex mode
+                                                    networkBridgeStatistics.getReceivedCount().increment();
                                                 } catch (IOException error) {
                                                     LOG.error("Exception: {} on duplex forward
of: {}", error, message);
                                                     serviceRemoteException(error);
@@ -648,6 +650,7 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge,
Br
                                         });
                                     } else {
                                         duplexInboundLocalBroker.oneway(message);
+                                        networkBridgeStatistics.getReceivedCount().increment();
                                     }
                                     serviceInboundMessage(message);
                                 } else {
@@ -968,7 +971,7 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge,
Br
             try {
                 if (command.isMessageDispatch()) {
                     safeWaitUntilStarted();
-                    enqueueCounter.incrementAndGet();
+                    networkBridgeStatistics.getEnqueues().increment();
                     final MessageDispatch md = (MessageDispatch) command;
                     final DemandSubscription sub = subscriptionMapByLocalId.get(md.getConsumerId());
                     if (sub != null && md.getMessage() != null && sub.incrementOutstandingResponses())
{
@@ -1016,7 +1019,7 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge,
Br
                                             serviceLocalException(md, er.getException());
                                         } else {
                                             localBroker.oneway(new MessageAck(md, MessageAck.INDIVIDUAL_ACK_TYPE,
1));
-                                            dequeueCounter.incrementAndGet();
+                                            networkBridgeStatistics.getDequeues().increment();
                                         }
                                     } catch (IOException e) {
                                         serviceLocalException(md, e);
@@ -1033,7 +1036,7 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge,
Br
                             try {
                                 remoteBroker.oneway(message);
                                 localBroker.oneway(new MessageAck(md, MessageAck.INDIVIDUAL_ACK_TYPE,
1));
-                                dequeueCounter.incrementAndGet();
+                                networkBridgeStatistics.getDequeues().increment();
                             } finally {
                                 sub.decrementOutstandingResponses();
                             }
@@ -1559,12 +1562,17 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge,
Br
 
     @Override
     public long getDequeueCounter() {
-        return dequeueCounter.get();
+        return networkBridgeStatistics.getDequeues().getCount();
     }
 
     @Override
     public long getEnqueueCounter() {
-        return enqueueCounter.get();
+        return networkBridgeStatistics.getEnqueues().getCount();
+    }
+
+    @Override
+    public NetworkBridgeStatistics getNetworkBridgeStatistics() {
+        return networkBridgeStatistics;
     }
 
     protected boolean isDuplex() {
@@ -1594,8 +1602,7 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge,
Br
 
     @Override
     public void resetStats() {
-        enqueueCounter.set(0);
-        dequeueCounter.set(0);
+        networkBridgeStatistics.reset();
     }
 
     /*

http://git-wip-us.apache.org/repos/asf/activemq/blob/10c998b0/activemq-broker/src/main/java/org/apache/activemq/network/NetworkBridge.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/network/NetworkBridge.java
b/activemq-broker/src/main/java/org/apache/activemq/network/NetworkBridge.java
index 4ed03dc..fb9e3d9 100644
--- a/activemq-broker/src/main/java/org/apache/activemq/network/NetworkBridge.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/network/NetworkBridge.java
@@ -79,6 +79,11 @@ public interface NetworkBridge extends Service {
     long getDequeueCounter();
 
     /**
+     * @return the statistics for this NetworkBridge
+     */
+    NetworkBridgeStatistics getNetworkBridgeStatistics();
+
+    /**
      * @param objectName
      *      The ObjectName assigned to this bridge in the MBean server.
      */

http://git-wip-us.apache.org/repos/asf/activemq/blob/10c998b0/activemq-broker/src/main/java/org/apache/activemq/network/NetworkBridgeStatistics.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/network/NetworkBridgeStatistics.java
b/activemq-broker/src/main/java/org/apache/activemq/network/NetworkBridgeStatistics.java
new file mode 100644
index 0000000..6b39d0a
--- /dev/null
+++ b/activemq-broker/src/main/java/org/apache/activemq/network/NetworkBridgeStatistics.java
@@ -0,0 +1,102 @@
+/**
+ * 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.activemq.network;
+
+import org.apache.activemq.management.CountStatisticImpl;
+import org.apache.activemq.management.StatsImpl;
+
+/**
+ * The Statistics for a NewtorkBridge.
+ */
+public class NetworkBridgeStatistics extends StatsImpl {
+
+    protected CountStatisticImpl enqueues;
+    protected CountStatisticImpl dequeues;
+    protected CountStatisticImpl receivedCount;
+
+    public NetworkBridgeStatistics() {
+        enqueues = new CountStatisticImpl("enqueues", "The current number of enqueues this
bridge has, which is the number of potential messages to be forwarded.");
+        dequeues = new CountStatisticImpl("dequeues", "The current number of dequeues this
bridge has, which is the number of messages received by the remote broker.");
+        receivedCount = new CountStatisticImpl("receivedCount", "The number of messages that
have been received by the NetworkBridge from the remote broker.  Only applies for Duplex bridges.");
+
+        addStatistic("enqueues", enqueues);
+        addStatistic("dequeues", dequeues);
+        addStatistic("receivedCount", receivedCount);
+    }
+
+    /**
+     * The current number of enqueues this bridge has, which is the number of potential messages
to be forwarded
+     * Messages may not be forwarded if there is no subscription
+     *
+     * @return
+     */
+    public CountStatisticImpl getEnqueues() {
+        return enqueues;
+    }
+
+    /**
+     * The current number of dequeues this bridge has, which is the number of
+     * messages actually sent to and received by the remote broker.
+     *
+     * @return
+     */
+    public CountStatisticImpl getDequeues() {
+        return dequeues;
+    }
+
+    /**
+     * The number of messages that have been received by the NetworkBridge from the remote
broker.
+     * Only applies for Duplex bridges.
+     *
+     * @return
+     */
+    public CountStatisticImpl getReceivedCount() {
+        return receivedCount;
+    }
+
+    @Override
+    public void reset() {
+        if (this.isDoReset()) {
+            super.reset();
+            enqueues.reset();
+            dequeues.reset();
+            receivedCount.reset();
+        }
+    }
+
+    @Override
+    public void setEnabled(boolean enabled) {
+        super.setEnabled(enabled);
+        enqueues.setEnabled(enabled);
+        dequeues.setEnabled(enabled);
+        receivedCount.setEnabled(enabled);
+    }
+
+    public void setParent(NetworkBridgeStatistics parent) {
+        if (parent != null) {
+            enqueues.setParent(parent.enqueues);
+            dequeues.setParent(parent.dequeues);
+            receivedCount.setParent(parent.receivedCount);
+        } else {
+            enqueues.setParent(null);
+            dequeues.setParent(null);
+            receivedCount.setParent(null);
+        }
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/activemq/blob/10c998b0/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4160Test.java
----------------------------------------------------------------------
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4160Test.java b/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4160Test.java
index 4867f28..5563ded 100644
--- a/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4160Test.java
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4160Test.java
@@ -35,6 +35,7 @@ import org.apache.activemq.command.DiscoveryEvent;
 import org.apache.activemq.network.DiscoveryNetworkConnector;
 import org.apache.activemq.network.NetworkBridge;
 import org.apache.activemq.network.NetworkBridgeListener;
+import org.apache.activemq.network.NetworkBridgeStatistics;
 import org.apache.activemq.network.NetworkConnector;
 import org.apache.activemq.thread.TaskRunnerFactory;
 import org.apache.activemq.transport.Transport;
@@ -56,6 +57,7 @@ public class AMQ4160Test extends JmsMultipleBrokersTestSupport {
      * Since these tests involve wait conditions, protect against indefinite
      * waits (due to unanticipated issues).
      */
+    @Override
     public void setUp() throws Exception {
         setAutoFail(true);
         setMaxTestTime(MAX_TEST_TIME);
@@ -328,6 +330,11 @@ public class AMQ4160Test extends JmsMultipleBrokersTestSupport {
                     }
 
                     @Override
+                    public NetworkBridgeStatistics getNetworkBridgeStatistics() {
+                        return next.getNetworkBridgeStatistics();
+                    }
+
+                    @Override
                     public void setMbeanObjectName(ObjectName objectName) {
                         next.setMbeanObjectName(objectName);
                     }
@@ -337,6 +344,7 @@ public class AMQ4160Test extends JmsMultipleBrokersTestSupport {
                         return next.getMbeanObjectName();
                     }
 
+                    @Override
                     public void resetStats(){
                         next.resetStats();
                     }

http://git-wip-us.apache.org/repos/asf/activemq/blob/10c998b0/activemq-unit-tests/src/test/java/org/apache/activemq/network/DuplexNetworkTest.java
----------------------------------------------------------------------
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/network/DuplexNetworkTest.java
b/activemq-unit-tests/src/test/java/org/apache/activemq/network/DuplexNetworkTest.java
index 14b2e0f..da29dd6 100755
--- a/activemq-unit-tests/src/test/java/org/apache/activemq/network/DuplexNetworkTest.java
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/network/DuplexNetworkTest.java
@@ -86,4 +86,19 @@ public class DuplexNetworkTest extends SimpleNetworkTest {
         return id;
     }
 
+    @Override
+    protected void assertNetworkBridgeStatistics(final long expectedLocalSent, final long
expectedRemoteSent) throws Exception {
+
+        final NetworkBridge localBridge = localBroker.getNetworkConnectors().get(0).activeBridges().iterator().next();
+
+        assertTrue(Wait.waitFor(new Wait.Condition() {
+            @Override
+            public boolean isSatisified() throws Exception {
+                return expectedLocalSent == localBridge.getNetworkBridgeStatistics().getDequeues().getCount()
&&
+                        expectedRemoteSent == localBridge.getNetworkBridgeStatistics().getReceivedCount().getCount();
+            }
+        }));
+
+    }
+
 }

http://git-wip-us.apache.org/repos/asf/activemq/blob/10c998b0/activemq-unit-tests/src/test/java/org/apache/activemq/network/SimpleNetworkTest.java
----------------------------------------------------------------------
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/network/SimpleNetworkTest.java
b/activemq-unit-tests/src/test/java/org/apache/activemq/network/SimpleNetworkTest.java
index d7e987c..a49301f 100755
--- a/activemq-unit-tests/src/test/java/org/apache/activemq/network/SimpleNetworkTest.java
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/network/SimpleNetworkTest.java
@@ -96,6 +96,8 @@ public class SimpleNetworkTest {
         }
         // ensure no more messages received
         assertNull(consumer1.receive(1000));
+
+        assertNetworkBridgeStatistics(MESSAGE_COUNT, 0);
     }
 
     @Test(timeout = 60 * 1000)
@@ -128,6 +130,8 @@ public class SimpleNetworkTest {
             assertNotNull(result);
             LOG.info(result.getText());
         }
+
+        assertNetworkBridgeStatistics(MESSAGE_COUNT, MESSAGE_COUNT);
     }
 
     @Test(timeout = 60 * 1000)
@@ -136,13 +140,15 @@ public class SimpleNetworkTest {
         MessageConsumer excludedConsumer = remoteSession.createConsumer(excluded);
         MessageProducer includedProducer = localSession.createProducer(included);
         MessageProducer excludedProducer = localSession.createProducer(excluded);
-        // allow for consumer infos to perculate arround
+        // allow for consumer infos to perculate around
         Thread.sleep(2000);
         Message test = localSession.createTextMessage("test");
         includedProducer.send(test);
         excludedProducer.send(test);
         assertNull(excludedConsumer.receive(1000));
         assertNotNull(includedConsumer.receive(1000));
+
+        assertNetworkBridgeStatistics(1, 0);
     }
 
     @Test(timeout = 60 * 1000)
@@ -163,6 +169,8 @@ public class SimpleNetworkTest {
         // ensure no more messages received
         assertNull(consumer1.receive(1000));
         assertNull(consumer2.receive(1000));
+
+        assertNetworkBridgeStatistics(MESSAGE_COUNT, 0);
     }
 
     private void waitForConsumerRegistration(final BrokerService brokerService, final int
min, final ActiveMQDestination destination) throws Exception {
@@ -319,4 +327,20 @@ public class SimpleNetworkTest {
     protected BrokerService createRemoteBroker() throws Exception {
         return createBroker(getRemoteBrokerURI());
     }
+
+    protected void assertNetworkBridgeStatistics(final long expectedLocalSent, final long
expectedRemoteSent) throws Exception {
+
+        final NetworkBridge localBridge = localBroker.getNetworkConnectors().get(0).activeBridges().iterator().next();
+        final NetworkBridge remoteBridge = remoteBroker.getNetworkConnectors().get(0).activeBridges().iterator().next();
+
+        assertTrue(Wait.waitFor(new Wait.Condition() {
+            @Override
+            public boolean isSatisified() throws Exception {
+                return expectedLocalSent == localBridge.getNetworkBridgeStatistics().getDequeues().getCount()
&&
+                       0 == localBridge.getNetworkBridgeStatistics().getReceivedCount().getCount()
&&
+                       expectedRemoteSent == remoteBridge.getNetworkBridgeStatistics().getDequeues().getCount()
&&
+                       0 == remoteBridge.getNetworkBridgeStatistics().getReceivedCount().getCount();
+            }
+        }));
+    }
 }


Mime
View raw message