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

Display the establised Neteowork Connector Bridges via JMX - https://issues.apache.org/activemq/browse/AMQ-1299 

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@551443 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Hiram R. Chirino 2007-06-28 04:34:57 +00:00
parent b56197408a
commit ec6e6a55a7
12 changed files with 497 additions and 93 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -278,6 +278,7 @@ public class BrokerService implements Service {
* network
*/
public NetworkConnector addNetworkConnector(NetworkConnector connector) throws Exception {
connector.setBrokerService(this);
URI uri = getVmConnectorURI();
HashMap map = new HashMap(URISupport.parseParamters(uri));
map.put("network", "true");
@ -1217,6 +1218,7 @@ public class BrokerService implements Service {
NetworkConnectorViewMBean view = new NetworkConnectorView(connector);
try {
ObjectName objectName = createNetworkConnectorObjectName(connector);
connector.setObjectName(objectName);
mbeanServer.registerMBean(view, objectName);
registeredMBeanNames.add(objectName);
}

61
activemq-core/src/main/java/org/apache/activemq/broker/jmx/NetworkBridgeView.java Normal file
View File

@ -0,0 +1,61 @@
/**
*
* Copyright 2005-2006 The Apache Software Foundation
*
* Licensed 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.broker.jmx;
import org.apache.activemq.network.NetworkBridge;
public class NetworkBridgeView implements NetworkBridgeViewMBean {
private final NetworkBridge bridge;
public NetworkBridgeView(NetworkBridge bridge) {
this.bridge = bridge;
}
public void start() throws Exception {
bridge.start();
}
public void stop() throws Exception {
bridge.stop();
}
public String getLocalAddress() {
return bridge.getLocalAddress();
}
public String getRemoteAddress() {
return bridge.getRemoteAddress();
}
public String getRemoteBrokerName() {
return bridge.getRemoteBrokerName();
}
public String getLocalBrokerName() {
return bridge.getLocalBrokerName();
}
public long getEnqueueCounter() {
return bridge.getEnqueueCounter();
}
public long getDequeueCounter() {
return bridge.getDequeueCounter();
}
}

30
activemq-core/src/main/java/org/apache/activemq/broker/jmx/NetworkBridgeViewMBean.java Normal file
View File

@ -0,0 +1,30 @@
/**
*
* Copyright 2005-2006 The Apache Software Foundation
*
* Licensed 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.broker.jmx;
import org.apache.activemq.Service;
public interface NetworkBridgeViewMBean extends Service {
public String getLocalAddress();
public String getRemoteAddress();
public String getRemoteBrokerName();
public String getLocalBrokerName();
public long getEnqueueCounter();
public long getDequeueCounter();
}

82
activemq-core/src/main/java/org/apache/activemq/broker/jmx/NetworkConnectorView.java
View File

@ -32,6 +32,86 @@ public class NetworkConnectorView implements NetworkConnectorViewMBean {
}
public void stop() throws Exception {
connector.stop();
connector.stop();
}
public String getName() {
return connector.getName();
}
public int getNetworkTTL() {
return connector.getNetworkTTL();
}
public int getPrefetchSize() {
return connector.getPrefetchSize();
}
public String getUserName() {
return connector.getUserName();
}
public boolean isBridgeTempDestinations() {
return connector.isBridgeTempDestinations();
}
public boolean isConduitSubscriptions() {
return connector.isConduitSubscriptions();
}
public boolean isDecreaseNetworkConsumerPriority() {
return connector.isDecreaseNetworkConsumerPriority();
}
public boolean isDispatchAsync() {
return connector.isDispatchAsync();
}
public boolean isDynamicOnly() {
return connector.isDynamicOnly();
}
public void setBridgeTempDestinations(boolean bridgeTempDestinations) {
connector.setBridgeTempDestinations(bridgeTempDestinations);
}
public void setConduitSubscriptions(boolean conduitSubscriptions) {
connector.setConduitSubscriptions(conduitSubscriptions);
}
public void setDispatchAsync(boolean dispatchAsync) {
connector.setDispatchAsync(dispatchAsync);
}
public void setDynamicOnly(boolean dynamicOnly) {
connector.setDynamicOnly(dynamicOnly);
}
public void setNetworkTTL(int networkTTL) {
connector.setNetworkTTL(networkTTL);
}
public void setPassword(String password) {
connector.setPassword(password);
}
public void setPrefetchSize(int prefetchSize) {
connector.setPrefetchSize(prefetchSize);
}
public void setUserName(String userName) {
connector.setUserName(userName);
}
public String getPassword() {
String pw = connector.getPassword();
// Hide the password for security reasons.
if( pw!= null )
pw = pw.replaceAll(".", "*");
return pw;
}
public void setDecreaseNetworkConsumerPriority(boolean decreaseNetworkConsumerPriority) {
connector.setDecreaseNetworkConsumerPriority(decreaseNetworkConsumerPriority);
}
}

20
activemq-core/src/main/java/org/apache/activemq/broker/jmx/NetworkConnectorViewMBean.java
View File

@ -21,4 +21,24 @@ import org.apache.activemq.Service;
public interface NetworkConnectorViewMBean extends Service {
public String getName();
public int getNetworkTTL();
public int getPrefetchSize();
public String getUserName();
public boolean isBridgeTempDestinations();
public boolean isConduitSubscriptions();
public boolean isDecreaseNetworkConsumerPriority();
public boolean isDispatchAsync();
public boolean isDynamicOnly();
public void setBridgeTempDestinations(boolean bridgeTempDestinations);
public void setConduitSubscriptions(boolean conduitSubscriptions);
public void setDispatchAsync(boolean dispatchAsync);
public void setDynamicOnly(boolean dynamicOnly);
public void setNetworkTTL(int networkTTL);
public void setPassword(String password);
public void setPrefetchSize(int prefetchSize);
public void setUserName(String userName);
public String getPassword();
public void setDecreaseNetworkConsumerPriority(boolean decreaseNetworkConsumerPriority);
}

67
activemq-core/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java
View File

@ -18,6 +18,12 @@
package org.apache.activemq.network;
import java.io.IOException;
import java.security.GeneralSecurityException;
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.activemq.advisory.AdvisorySupport;
import org.apache.activemq.command.ActiveMQDestination;
@ -61,12 +67,6 @@ import org.apache.activemq.util.ServiceSupport;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import java.security.GeneralSecurityException;
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicBoolean;
/**
* A useful base class for implementing demand forwarding bridges.
*
@ -102,9 +102,14 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge {
protected final AtomicBoolean remoteInterupted = new AtomicBoolean(false);
protected final AtomicBoolean lastConnectSucceeded = new AtomicBoolean(false);
protected NetworkBridgeConfiguration configuration;
private NetworkBridgeFailedListener bridgeFailedListener;
private NetworkBridgeListener networkBridgeListener;
private boolean createdByDuplex;
private BrokerInfo localBrokerInfo;
private BrokerInfo remoteBrokerInfo;
final AtomicLong enqueueCounter = new AtomicLong();
final AtomicLong dequeueCounter = new AtomicLong();
public DemandForwardingBridgeSupport(NetworkBridgeConfiguration configuration, Transport localBroker, Transport remoteBroker) {
this.configuration=configuration;
@ -194,6 +199,12 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge {
}catch(IOException e){
log.warn("Caught exception from remote start",e);
}
NetworkBridgeListener l = this.networkBridgeListener;
if (l!=null) {
l.onStart(this);
}
}
protected void triggerLocalStartBridge() throws IOException {
@ -308,6 +319,11 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge {
log.debug(" stopping "+configuration.getBrokerName()+" bridge to "+remoteBrokerName+" is disposed already ? "+disposed);
boolean wasDisposedAlready=disposed;
if(!disposed){
NetworkBridgeListener l = this.networkBridgeListener;
if (l!=null) {
l.onStop(this);
}
try{
disposed=true;
remoteBridgeStarted.set(false);
@ -364,6 +380,8 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge {
}else if(command.isBrokerInfo()){
lastConnectSucceeded.set(true);
remoteBrokerInfo = ((BrokerInfo)command);
serviceRemoteBrokerInfo(command);
// Let the local broker know the remote broker's ID.
localBroker.oneway(command);
@ -507,6 +525,7 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge {
final boolean trace=log.isTraceEnabled();
try{
if(command.isMessageDispatch()){
enqueueCounter.incrementAndGet();
waitStarted();
final MessageDispatch md=(MessageDispatch) command;
DemandSubscription sub=(DemandSubscription) subscriptionMapByLocalId.get(md.getConsumerId());
@ -523,6 +542,7 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge {
// by bridging it using an async send (small chance of message loss).
remoteBroker.oneway(message);
localBroker.oneway(new MessageAck(md,MessageAck.STANDARD_ACK_TYPE,1));
dequeueCounter.incrementAndGet();
} else {
@ -537,6 +557,7 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge {
serviceLocalException(er.getException());
} else {
localBroker.oneway(new MessageAck(md,MessageAck.STANDARD_ACK_TYPE,1));
dequeueCounter.incrementAndGet();
}
} catch (IOException e) {
serviceLocalException(e);
@ -551,6 +572,7 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge {
if (trace)log.trace("No subscription registered with this network bridge for consumerId " + md.getConsumerId() + " for message: " + md.getMessage());
}
}else if(command.isBrokerInfo()){
localBrokerInfo = ((BrokerInfo)command);
serviceLocalBrokerInfo(command);
}else if(command.isShutdownInfo()){
log.info(configuration.getBrokerName()+" Shutting down");
@ -812,14 +834,39 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge {
protected abstract BrokerId[] getRemoteBrokerPath();
public void setNetworkBridgeFailedListener(NetworkBridgeFailedListener listener){
this.bridgeFailedListener=listener;
public void setNetworkBridgeListener(NetworkBridgeListener listener){
this.networkBridgeListener=listener;
}
private void fireBridgeFailed() {
NetworkBridgeFailedListener l = this.bridgeFailedListener;
NetworkBridgeListener l = this.networkBridgeListener;
if (l!=null) {
l.bridgeFailed();
}
}
public String getRemoteAddress() {
return remoteBroker.getRemoteAddress();
}
public String getLocalAddress() {
return localBroker.getRemoteAddress();
}
public String getRemoteBrokerName() {
return remoteBrokerInfo == null ? null : remoteBrokerInfo.getBrokerName();
}
public String getLocalBrokerName() {
return localBrokerInfo == null ? null : localBrokerInfo.getBrokerName();
}
public long getDequeueCounter() {
return dequeueCounter.get();
}
public long getEnqueueCounter() {
return enqueueCounter.get();
}
}

13
activemq-core/src/main/java/org/apache/activemq/network/DiscoveryNetworkConnector.java
View File

@ -124,7 +124,7 @@ public class DiscoveryNetworkConnector extends NetworkConnector implements Disco
}
}
}
public void onServiceRemove(DiscoveryEvent event) {
String url = event.getServiceName();
if (url != null) {
@ -186,7 +186,7 @@ public class DiscoveryNetworkConnector extends NetworkConnector implements Disco
}
protected NetworkBridge createBridge(Transport localTransport, Transport remoteTransport, final DiscoveryEvent event) {
NetworkBridgeFailedListener listener = new NetworkBridgeFailedListener() {
NetworkBridgeListener listener = new NetworkBridgeListener() {
public void bridgeFailed(){
if( !serviceSupport.isStopped() ) {
@ -197,6 +197,15 @@ public class DiscoveryNetworkConnector extends NetworkConnector implements Disco
}
}
public void onStart(NetworkBridge bridge) {
registerNetworkBridgeMBean(bridge);
}
public void onStop(NetworkBridge bridge) {
unregisterNetworkBridgeMBean(bridge);
}
};
DemandForwardingBridge result = NetworkBridgeFactory.createBridge(this,localTransport,remoteTransport,listener);

51
activemq-core/src/main/java/org/apache/activemq/network/ForwardingBridge.java
View File

@ -18,6 +18,7 @@
package org.apache.activemq.network;
import java.io.IOException;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.activemq.Service;
import org.apache.activemq.command.ActiveMQQueue;
@ -46,6 +47,7 @@ import org.apache.activemq.util.ServiceSupport;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
/**
* Forwards all messages from the local broker to the remote broker.
*
@ -74,7 +76,13 @@ public class ForwardingBridge implements Service{
BrokerId localBrokerId;
BrokerId remoteBrokerId;
private NetworkBridgeFailedListener bridgeFailedListener;
private NetworkBridgeListener bridgeFailedListener;
BrokerInfo localBrokerInfo;
BrokerInfo remoteBrokerInfo;
final AtomicLong enqueueCounter = new AtomicLong();
final AtomicLong dequeueCounter = new AtomicLong();
public ForwardingBridge(Transport localBroker, Transport remoteBroker) {
this.localBroker = localBroker;
@ -187,7 +195,8 @@ public class ForwardingBridge implements Service{
try {
if(command.isBrokerInfo() ) {
synchronized( this ) {
remoteBrokerId = ((BrokerInfo)command).getBrokerId();
remoteBrokerInfo = ((BrokerInfo)command);
remoteBrokerId = remoteBrokerInfo.getBrokerId();
if( localBrokerId !=null) {
if( localBrokerId.equals(remoteBrokerId) ) {
log.info("Disconnecting loop back connection.");
@ -213,6 +222,9 @@ public class ForwardingBridge implements Service{
protected void serviceLocalCommand(Command command) {
try {
if( command.isMessageDispatch() ) {
enqueueCounter.incrementAndGet();
final MessageDispatch md = (MessageDispatch) command;
Message message = md.getMessage();
message.setProducerId(producerInfo.getProducerId());
@ -223,10 +235,10 @@ public class ForwardingBridge implements Service{
if( !message.isResponseRequired() ) {
// If the message was originally sent using async send, we will preserve that QOS
// by bridging it using an async send (small chance of message loss).
remoteBroker.oneway(message);
dequeueCounter.incrementAndGet();
localBroker.oneway(new MessageAck(md,MessageAck.STANDARD_ACK_TYPE,1));
} else {
@ -241,6 +253,7 @@ public class ForwardingBridge implements Service{
ExceptionResponse er=(ExceptionResponse) response;
serviceLocalException(er.getException());
} else {
dequeueCounter.incrementAndGet();
localBroker.oneway(new MessageAck(md,MessageAck.STANDARD_ACK_TYPE,1));
}
} catch (IOException e) {
@ -273,7 +286,8 @@ public class ForwardingBridge implements Service{
// }
} else if(command.isBrokerInfo() ) {
synchronized( this ) {
localBrokerId = ((BrokerInfo)command).getBrokerId();
localBrokerInfo = ((BrokerInfo)command);
localBrokerId = localBrokerInfo.getBrokerId();
if( remoteBrokerId !=null) {
if( remoteBrokerId.equals(localBrokerId) ) {
log.info("Disconnecting loop back connection.");
@ -320,14 +334,39 @@ public class ForwardingBridge implements Service{
}
public void setNetworkBridgeFailedListener(NetworkBridgeFailedListener listener){
public void setNetworkBridgeFailedListener(NetworkBridgeListener listener){
this.bridgeFailedListener=listener;
}
private void fireBridgeFailed() {
NetworkBridgeFailedListener l = this.bridgeFailedListener;
NetworkBridgeListener l = this.bridgeFailedListener;
if (l!=null) {
l.bridgeFailed();
}
}
public String getRemoteAddress() {
return remoteBroker.getRemoteAddress();
}
public String getLocalAddress() {
return localBroker.getRemoteAddress();
}
public String getLocalBrokerName() {
return localBrokerInfo == null ? null : localBrokerInfo.getBrokerName();
}
public String getRemoteBrokerName() {
return remoteBrokerInfo == null ? null : remoteBrokerInfo.getBrokerName();
}
public long getDequeueCounter() {
return dequeueCounter.get();
}
public long getEnqueueCounter() {
return enqueueCounter.get();
}
}

15
activemq-core/src/main/java/org/apache/activemq/network/NetworkBridge.java
View File

@ -43,5 +43,18 @@ public interface NetworkBridge extends Service {
* Set the NetworkBridgeFailedListener
* @param listener
*/
public void setNetworkBridgeFailedListener(NetworkBridgeFailedListener listener);
public void setNetworkBridgeListener(NetworkBridgeListener listener);
public String getRemoteAddress();
public String getRemoteBrokerName();
public String getLocalAddress();
public String getLocalBrokerName();
public long getEnqueueCounter();
public long getDequeueCounter();
}

83
activemq-core/src/main/java/org/apache/activemq/network/NetworkBridgeFactory.java
View File

@ -21,45 +21,50 @@ import org.apache.activemq.transport.Transport;
*
* @version $Revision: 1.1 $
*/
public class NetworkBridgeFactory{
public class NetworkBridgeFactory {
/**
* Create a network bridge
*
* @param config
* @param localTransport
* @param remoteTransport
* @return the NetworkBridge
*/
public static DemandForwardingBridge createBridge(NetworkBridgeConfiguration config,Transport localTransport,
Transport remoteTransport){
return createBridge(config,localTransport,remoteTransport,null);
}
/**
* Create a network bridge
*
* @param config
* @param localTransport
* @param remoteTransport
* @return the NetworkBridge
*/
public static DemandForwardingBridge createBridge(
NetworkBridgeConfiguration config, Transport localTransport,
Transport remoteTransport) {
return createBridge(config, localTransport, remoteTransport, null);
}
/**
* create a network bridge
*
* @param configuration
* @param localTransport
* @param remoteTransport
* @param listener
* @return the NetworkBridge
*/
public static DemandForwardingBridge createBridge(NetworkBridgeConfiguration configuration,Transport localTransport,
Transport remoteTransport,NetworkBridgeFailedListener listener){
DemandForwardingBridge result=null;
if(configuration.isConduitSubscriptions()){
if(configuration.isDynamicOnly()){
result=new ConduitBridge(configuration,localTransport,remoteTransport);
}else{
result=new DurableConduitBridge(configuration,localTransport,remoteTransport);
}
}else{
result=new DemandForwardingBridge(configuration,localTransport,remoteTransport);
}
if(listener!=null){
result.setNetworkBridgeFailedListener(listener);
}
return result;
}
/**
* create a network bridge
*
* @param configuration
* @param localTransport
* @param remoteTransport
* @param listener
* @return the NetworkBridge
*/
public static DemandForwardingBridge createBridge(
NetworkBridgeConfiguration configuration, Transport localTransport,
Transport remoteTransport, final NetworkBridgeListener listener) {
DemandForwardingBridge result = null;
if (configuration.isConduitSubscriptions()) {
if (configuration.isDynamicOnly()) {
result = new ConduitBridge(configuration, localTransport,
remoteTransport);
} else {
result = new DurableConduitBridge(configuration,
localTransport, remoteTransport);
}
} else {
result = new DemandForwardingBridge(configuration, localTransport,
remoteTransport);
}
if (listener != null) {
result.setNetworkBridgeListener(listener);
}
return result;
}
}

81
activemq-core/src/main/java/org/apache/activemq/network/NetworkBridgeFailedListener.java → activemq-core/src/main/java/org/apache/activemq/network/NetworkBridgeListener.java
View File

@ -1,34 +1,47 @@
/**
*
* 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;
/**
*called when a bridge fails
*
* @version $Revision: 1.1 $
*/
public interface NetworkBridgeFailedListener{
/**
* called when the transport fails
*
*/
public void bridgeFailed();
}
/**
*
* 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;
/**
*called when a bridge fails
*
* @version $Revision: 1.1 $
*/
public interface NetworkBridgeListener{
/**
* called when the transport fails
*
*/
public void bridgeFailed();
/**
* called after the bridge is started.
*
*/
public void onStart(NetworkBridge bridge);
/**
* called before the bridge is stopped.
*
*/
public void onStop(NetworkBridge bridge);
}

85
activemq-core/src/main/java/org/apache/activemq/network/NetworkConnector.java
View File

@ -16,13 +16,23 @@ package org.apache.activemq.network;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.Hashtable;
import java.util.List;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArrayList;
import javax.management.MBeanServer;
import javax.management.MalformedObjectNameException;
import javax.management.ObjectName;
import org.apache.activemq.Service;
import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.broker.jmx.NetworkBridgeView;
import org.apache.activemq.broker.jmx.NetworkBridgeViewMBean;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.transport.Transport;
import org.apache.activemq.transport.TransportFactory;
import org.apache.activemq.util.JMXSupport;
import org.apache.activemq.util.ServiceStopper;
import org.apache.activemq.util.ServiceSupport;
import org.apache.commons.logging.Log;
@ -40,6 +50,9 @@ public abstract class NetworkConnector extends NetworkBridgeConfiguration implem
private List dynamicallyIncludedDestinations=new CopyOnWriteArrayList();
private List staticallyIncludedDestinations=new CopyOnWriteArrayList();
protected ConnectionFilter connectionFilter;
private BrokerService brokerService;
private ObjectName objectName;
protected ServiceSupport serviceSupport=new ServiceSupport(){
protected void doStart() throws Exception{
@ -188,4 +201,76 @@ public abstract class NetworkConnector extends NetworkBridgeConfiguration implem
protected void handleStop(ServiceStopper stopper) throws Exception{
log.info("Network Connector "+getName()+" Stopped");
}
public ObjectName getObjectName() {
return objectName;
}
public void setObjectName(ObjectName objectName) {
this.objectName = objectName;
}
public BrokerService getBrokerService() {
return brokerService;
}
public void setBrokerService(BrokerService brokerService) {
this.brokerService = brokerService;
}
protected void registerNetworkBridgeMBean(NetworkBridge bridge) {
if (!getBrokerService().isUseJmx())
return;
MBeanServer mbeanServer = getBrokerService().getManagementContext()
.getMBeanServer();
if (mbeanServer != null) {
NetworkBridgeViewMBean view = new NetworkBridgeView(bridge);
try {
ObjectName objectName = createNetworkBridgeObjectName(bridge);
mbeanServer.registerMBean(view, objectName);
} catch (Throwable e) {
log.debug("Network bridge could not be registered in JMX: "
+ e.getMessage(), e);
}
}
}
protected void unregisterNetworkBridgeMBean(NetworkBridge bridge) {
if (!getBrokerService().isUseJmx())
return;
MBeanServer mbeanServer = getBrokerService().getManagementContext()
.getMBeanServer();
if (mbeanServer != null) {
try {
ObjectName objectName = createNetworkBridgeObjectName(bridge);
mbeanServer.unregisterMBean(objectName);
} catch (Throwable e) {
log.debug("Network bridge could not be unregistered in JMX: "
+ e.getMessage(), e);
}
}
}
protected ObjectName createNetworkBridgeObjectName(NetworkBridge bridge)
throws MalformedObjectNameException {
ObjectName connectorName = getObjectName();
Hashtable map = connectorName.getKeyPropertyList();
return new ObjectName(connectorName.getDomain()
+ ":"
+ "BrokerName="
+ JMXSupport.encodeObjectNamePart((String) map
.get("BrokerName"))
+ ","
+ "Type=NetworkBridge,"
+ "NetworkConnectorName="
+ JMXSupport.encodeObjectNamePart((String) map
.get("NetworkConnectorName"))
+ ","
+ "Name="
+ JMXSupport.encodeObjectNamePart(JMXSupport
.encodeObjectNamePart(bridge.getRemoteAddress())));
}
}