Apache
/
activemq
mirror of https://github.com/apache/activemq.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix for https://issues.apache.org/jira/browse/AMQ-4918

This commit is contained in:
rajdavies 2013-12-05 08:55:56 +00:00
parent 9b88902459
commit 374cab9cd5
10 changed files with 116 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
activemq-broker/src/main/java/org/apache/activemq/broker/TransportConnection.java
View File

@ -1300,7 +1300,7 @@ public class TransportConnection implements Connection, Task, CommandVisitor {
if (duplexName.contains("#")) {
duplexName = duplexName.substring(duplexName.lastIndexOf("#"));
}
MBeanNetworkListener listener = new MBeanNetworkListener(broker.getBrokerService(), broker.getBrokerService().createDuplexNetworkConnectorObjectName(duplexName));
MBeanNetworkListener listener = new MBeanNetworkListener(broker.getBrokerService(), config, broker.getBrokerService().createDuplexNetworkConnectorObjectName(duplexName));
listener.setCreatedByDuplex(true);
duplexBridge = NetworkBridgeFactory.createBridge(config, localTransport, remoteBridgeTransport, listener);
duplexBridge.setBrokerService(broker.getBrokerService());

1
activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkBridgeView.java
View File

@ -72,6 +72,7 @@ public class NetworkBridgeView implements NetworkBridgeViewMBean {
}
public void resetStats(){
bridge.resetStats();
for (NetworkDestinationView networkDestinationView:networkDestinationViewList){
networkDestinationView.resetStats();
}

6
activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkDestinationView.java
View File

@ -73,7 +73,13 @@ public class NetworkDestinationView implements NetworkDestinationViewMBean {
lastTime=currentTime;
}
public long getLastAccessTime(){
return timeStatistic.getLastSampleTime();
}
public void close(){
networkBridgeView.removeNetworkDestinationView(this);
}
}

5
activemq-broker/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java
View File

@ -1538,6 +1538,11 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge, Br
return mbeanObjectName;
}
public void resetStats(){
enqueueCounter.set(0);
dequeueCounter.set(0);
}
/*
* Used to allow for async tasks to await receipt of the BrokerInfo from the local and
* remote sides of the network bridge.

2
activemq-broker/src/main/java/org/apache/activemq/network/DiscoveryNetworkConnector.java
View File

@ -231,7 +231,7 @@ public class DiscoveryNetworkConnector extends NetworkConnector implements Disco
class DiscoverNetworkBridgeListener extends MBeanNetworkListener {
public DiscoverNetworkBridgeListener(BrokerService brokerService, ObjectName connectorName) {
super(brokerService, connectorName);
super(brokerService, DiscoveryNetworkConnector.this, connectorName);
}
@Override

68
activemq-broker/src/main/java/org/apache/activemq/network/MBeanBridgeDestination.java
View File

@ -16,6 +16,7 @@
*/
package org.apache.activemq.network;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
@ -27,6 +28,7 @@ import org.apache.activemq.broker.jmx.NetworkBridgeView;
import org.apache.activemq.broker.jmx.NetworkDestinationView;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.command.Message;
import org.apache.activemq.thread.Scheduler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -35,14 +37,24 @@ public class MBeanBridgeDestination {
private final BrokerService brokerService;
private final NetworkBridge bridge;
private final NetworkBridgeView networkBridgeView;
private final NetworkBridgeConfiguration networkBridgeConfiguration;
private final Scheduler scheduler;
private final Runnable purgeInactiveDestinationViewTask;
private Map<ActiveMQDestination, ObjectName> destinationObjectNameMap = new ConcurrentHashMap<ActiveMQDestination, ObjectName>();
private Map<ActiveMQDestination, NetworkDestinationView> outboundDestinationViewMap = new ConcurrentHashMap<ActiveMQDestination, NetworkDestinationView>();
private Map<ActiveMQDestination, NetworkDestinationView> inboundDestinationViewMap = new ConcurrentHashMap<ActiveMQDestination, NetworkDestinationView>();
public MBeanBridgeDestination(BrokerService brokerService, NetworkBridge bridge, NetworkBridgeView networkBridgeView) {
public MBeanBridgeDestination(BrokerService brokerService, NetworkBridgeConfiguration networkBridgeConfiguration, NetworkBridge bridge, NetworkBridgeView networkBridgeView) {
this.brokerService = brokerService;
this.networkBridgeConfiguration = networkBridgeConfiguration;
this.bridge = bridge;
this.networkBridgeView = networkBridgeView;
this.scheduler = brokerService.getScheduler();
purgeInactiveDestinationViewTask = new Runnable() {
public void run() {
purgeInactiveDestinationViews();
}
};
}
@ -55,7 +67,7 @@ public class MBeanBridgeDestination {
ObjectName bridgeObjectName = bridge.getMbeanObjectName();
try {
ObjectName objectName = BrokerMBeanSupport.createNetworkOutBoundDestinationObjectName(bridgeObjectName, destination);
networkDestinationView = new NetworkDestinationView(networkBridgeView,destination.getPhysicalName());
networkDestinationView = new NetworkDestinationView(networkBridgeView, destination.getPhysicalName());
AnnotatedMBean.registerMBean(brokerService.getManagementContext(), networkDestinationView, objectName);
destinationObjectNameMap.put(destination, objectName);
outboundDestinationViewMap.put(destination, networkDestinationView);
@ -79,7 +91,7 @@ public class MBeanBridgeDestination {
ObjectName bridgeObjectName = bridge.getMbeanObjectName();
try {
ObjectName objectName = BrokerMBeanSupport.createNetworkInBoundDestinationObjectName(bridgeObjectName, destination);
networkDestinationView= new NetworkDestinationView(networkBridgeView,destination.getPhysicalName());
networkDestinationView = new NetworkDestinationView(networkBridgeView, destination.getPhysicalName());
networkBridgeView.addNetworkDestinationView(networkDestinationView);
AnnotatedMBean.registerMBean(brokerService.getManagementContext(), networkDestinationView, objectName);
destinationObjectNameMap.put(destination, objectName);
@ -93,11 +105,21 @@ public class MBeanBridgeDestination {
networkDestinationView.messageSent();
}
public void close() {
public void start() {
if (networkBridgeConfiguration.isGcDestinationViews()) {
long period = networkBridgeConfiguration.getGcSweepTime();
if (period > 0) {
scheduler.executePeriodically(purgeInactiveDestinationViewTask, period);
}
}
}
public void stop() {
if (!brokerService.isUseJmx()) {
return;
}
scheduler.cancel(purgeInactiveDestinationViewTask);
for (ObjectName objectName : destinationObjectNameMap.values()) {
try {
if (objectName != null) {
@ -112,4 +134,42 @@ public class MBeanBridgeDestination {
inboundDestinationViewMap.clear();
}
private void purgeInactiveDestinationViews() {
if (!brokerService.isUseJmx()) {
return;
}
purgeInactiveDestinationView(inboundDestinationViewMap);
purgeInactiveDestinationView(outboundDestinationViewMap);
}
private void purgeInactiveDestinationView(Map<ActiveMQDestination, NetworkDestinationView> map) {
long time = System.currentTimeMillis() - networkBridgeConfiguration.getGcSweepTime();
Map<ActiveMQDestination, NetworkDestinationView> gc = null;
for (Map.Entry<ActiveMQDestination, NetworkDestinationView> entry : map.entrySet()) {
if (entry.getValue().getLastAccessTime() <= time) {
if (gc == null) {
gc = new HashMap<ActiveMQDestination, NetworkDestinationView>();
}
gc.put(entry.getKey(), entry.getValue());
}
}
if (gc != null) {
for (Map.Entry<ActiveMQDestination, NetworkDestinationView> entry : gc.entrySet()) {
map.remove(entry.getKey());
ObjectName objectName = destinationObjectNameMap.get(entry.getKey());
if (objectName != null) {
try {
if (objectName != null) {
brokerService.getManagementContext().unregisterMBean(objectName);
}
} catch (Throwable e) {
LOG.debug("Network bridge could not be unregistered in JMX: {}", e.getMessage(), e);
}
}
entry.getValue().close();
}
}
}
}

16
activemq-broker/src/main/java/org/apache/activemq/network/MBeanNetworkListener.java
View File

@ -33,12 +33,15 @@ public class MBeanNetworkListener implements NetworkBridgeListener {
private static final Logger LOG = LoggerFactory.getLogger(MBeanNetworkListener.class);
BrokerService brokerService;
ObjectName connectorName;
boolean createdByDuplex = false;
private final BrokerService brokerService;
private final ObjectName connectorName;
private final NetworkBridgeConfiguration networkBridgeConfiguration;
private boolean createdByDuplex = false;
private Map<NetworkBridge,MBeanBridgeDestination> destinationObjectNameMap = new ConcurrentHashMap<NetworkBridge,MBeanBridgeDestination>();
public MBeanNetworkListener(BrokerService brokerService, ObjectName connectorName) {
public MBeanNetworkListener(BrokerService brokerService, NetworkBridgeConfiguration networkBridgeConfiguration, ObjectName connectorName) {
this.brokerService = brokerService;
this.networkBridgeConfiguration = networkBridgeConfiguration;
this.connectorName = connectorName;
}
@ -57,8 +60,9 @@ public class MBeanNetworkListener implements NetworkBridgeListener {
ObjectName objectName = createNetworkBridgeObjectName(bridge);
AnnotatedMBean.registerMBean(brokerService.getManagementContext(), view, objectName);
bridge.setMbeanObjectName(objectName);
MBeanBridgeDestination mBeanBridgeDestination = new MBeanBridgeDestination(brokerService,bridge,view);
MBeanBridgeDestination mBeanBridgeDestination = new MBeanBridgeDestination(brokerService,networkBridgeConfiguration,bridge,view);
destinationObjectNameMap.put(bridge,mBeanBridgeDestination);
mBeanBridgeDestination.start();
LOG.debug("registered: {} as: {}", bridge, objectName);
} catch (Throwable e) {
LOG.debug("Network bridge could not be registered in JMX: {}", e.getMessage(), e);
@ -77,7 +81,7 @@ public class MBeanNetworkListener implements NetworkBridgeListener {
}
MBeanBridgeDestination mBeanBridgeDestination = destinationObjectNameMap.remove(bridge);
if (mBeanBridgeDestination != null){
mBeanBridgeDestination.close();
mBeanBridgeDestination.stop();
}
} catch (Throwable e) {
LOG.debug("Network bridge could not be unregistered in JMX: {}", e.getMessage(), e);

2
activemq-broker/src/main/java/org/apache/activemq/network/NetworkBridge.java
View File

@ -83,4 +83,6 @@ public interface NetworkBridge extends Service {
* @return the MBean name used to identify this bridge in the MBean server.
*/
ObjectName getMbeanObjectName();
void resetStats();
}

25
activemq-broker/src/main/java/org/apache/activemq/network/NetworkBridgeConfiguration.java
View File

@ -16,13 +16,13 @@
*/
package org.apache.activemq.network;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import org.apache.activemq.advisory.AdvisorySupport;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.command.ConsumerInfo;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
/**
* Configuration for a NetworkBridge
*/
@ -59,6 +59,8 @@ public class NetworkBridgeConfiguration {
private boolean useCompression = false;
private boolean advisoryForFailedForward = false;
private boolean useBrokerNamesAsIdSeed = true;
private boolean gcDestinationViews = true;
private long gcSweepTime = 60 * 1000;
/**
* @return the conduitSubscriptions
@ -421,4 +423,21 @@ public class NetworkBridgeConfiguration {
public void setUseBrokerNameAsIdSees(boolean val) {
useBrokerNamesAsIdSeed = val;
}
public boolean isGcDestinationViews() {
return gcDestinationViews;
}
public void setGcDestinationViews(boolean gcDestinationViews) {
this.gcDestinationViews = gcDestinationViews;
}
public long getGcSweepTime() {
return gcSweepTime;
}
public void setGcSweepTime(long gcSweepTime) {
this.gcSweepTime = gcSweepTime;
}
}

4
activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4160Test.java
View File

@ -331,6 +331,10 @@ public class AMQ4160Test extends JmsMultipleBrokersTestSupport {
public ObjectName getMbeanObjectName() {
return next.getMbeanObjectName();
}
public void resetStats(){
next.resetStats();
}
};
}
};