/**
*
* Copyright 2004 Protique Ltd
*
* 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.codehaus.activemq.transport;
import EDU.oswego.cs.dl.util.concurrent.ConcurrentHashMap;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.codehaus.activemq.ActiveMQConnection;
import org.codehaus.activemq.ActiveMQConnectionFactory;
import org.codehaus.activemq.broker.BrokerClient;
import org.codehaus.activemq.broker.BrokerContainer;
import org.codehaus.activemq.broker.ConsumerInfoListener;
import org.codehaus.activemq.message.ActiveMQDestination;
import org.codehaus.activemq.message.BrokerInfo;
import org.codehaus.activemq.message.ConsumerInfo;
import org.codehaus.activemq.message.Receipt;
import org.codehaus.activemq.service.MessageContainerManager;
import org.codehaus.activemq.service.Service;
import org.codehaus.activemq.transport.composite.CompositeTransportChannel;
import javax.jms.JMSException;
import javax.jms.Session;
import java.util.Iterator;
import java.util.Map;
/**
* Represents a broker's connection with a single remote broker which bridges the two brokers to form a network. <p/>
* The NetworkChannel contains a JMS connection with the remote broker. <p/>New subscriptions on the local broker are
* multiplexed into the JMS connection so that messages published on the remote broker can be replayed onto the local
* broker.
*
* @version $Revision: 1.18 $
*/
public class NetworkChannel implements Service, ConsumerInfoListener {
private static final Log log = LogFactory.getLog(NetworkChannel.class);
private String uri;
private BrokerContainer brokerContainer;
private ActiveMQConnection localConnection;
private ActiveMQConnection remoteConnection;
private ConcurrentHashMap consumerMap;
private String remoteUserName;
private String remotePassword;
private String remoteBrokerName;
private String remoteClusterName;
private int maximumRetries = 0;
private long reconnectSleepTime = 1000L;
/**
* Default Constructor
*/
public NetworkChannel() {
this.consumerMap = new ConcurrentHashMap();
}
/**
* Constructor
*
* @param brokerContainer
* @param uri
*/
public NetworkChannel(BrokerContainer brokerContainer, String uri) {
this();
this.brokerContainer = brokerContainer;
this.uri = uri;
}
/**
* @return text info on this
*/
public String toString() {
return super.toString() + "[uri=" + uri + "]";
}
/**
* Start the channel
*/
public void start() throws JMSException {
Thread runner = new Thread(new Runnable() {
public void run() {
try {
initialize();
brokerContainer.getBroker().addConsumerInfoListener(NetworkChannel.this);
startSubscriptions();
log.info("Started NetworkChannel to " + uri);
}
catch (JMSException jmsEx) {
log.error("Failed to start NetworkChannel: " + uri);
}
}
}, "NetworkChannel Starter");
runner.setDaemon(true);
runner.start();
}
/**
* stop the channel
*
* @throws JMSException on error
*/
public void stop() throws JMSException {
consumerMap.clear();
if (remoteConnection != null) {
remoteConnection.close();
remoteConnection = null;
}
if (localConnection != null) {
localConnection.close();
localConnection = null;
}
for (Iterator i = consumerMap.values().iterator();i.hasNext();) {
NetworkMessageBridge consumer = (NetworkMessageBridge) i.next();
consumer.stop();
}
}
/**
* Listen for new Consumer events at this broker
*
* @param client
* @param info
*/
public void onConsumerInfo(BrokerClient client, ConsumerInfo info) {
if (!client.isClusteredConnection()) {
if (!info.hasVisited(remoteBrokerName)) {
if (info.isStarted()) {
addConsumerInfo(info);
}
else {
removeConsumerInfo(info);
}
}
}
}
/**
* @return the uri of the broker(s) this channel is connected to
*/
public String getUri() {
return uri;
}
/**
* set the uri of the broker(s) this channel is connected to
*
* @param uri
*/
public void setUri(String uri) {
this.uri = uri;
}
/**
* @return Returns the remotePassword.
*/
public String getRemotePassword() {
return remotePassword;
}
/**
* @param remotePassword The remotePassword to set.
*/
public void setRemotePassword(String remotePassword) {
this.remotePassword = remotePassword;
}
/**
* @return Returns the remoteUserName.
*/
public String getRemoteUserName() {
return remoteUserName;
}
/**
* @param remoteUserName The remoteUserName to set.
*/
public void setRemoteUserName(String remoteUserName) {
this.remoteUserName = remoteUserName;
}
/**
* @return Returns the brokerContainer.
*/
public BrokerContainer getBrokerContainer() {
return brokerContainer;
}
/**
* @param brokerContainer The brokerContainer to set.
*/
public void setBrokerContainer(BrokerContainer brokerContainer) {
this.brokerContainer = brokerContainer;
}
public int getMaximumRetries() {
return maximumRetries;
}
public void setMaximumRetries(int maximumRetries) {
this.maximumRetries = maximumRetries;
}
public long getReconnectSleepTime() {
return reconnectSleepTime;
}
public void setReconnectSleepTime(long reconnectSleepTime) {
this.reconnectSleepTime = reconnectSleepTime;
}
public String getRemoteBrokerName() {
return remoteBrokerName;
}
public void setRemoteBrokerName(String remoteBrokerName) {
this.remoteBrokerName = remoteBrokerName;
}
// Implementation methods
//-------------------------------------------------------------------------
private void addConsumerInfo(ConsumerInfo info) {
addConsumerInfo(info.getDestination(), info.isDurableTopic());
}
private void addConsumerInfo(ActiveMQDestination destination, boolean durableTopic) {
NetworkMessageBridge key = new NetworkMessageBridge();
key.setDestination(destination);
key.setDurableTopic(durableTopic);
NetworkMessageBridge bridge = (NetworkMessageBridge) consumerMap.get(key);
if (bridge == null) {
try {
bridge = key;//might as well recycle :-)
bridge.setLocalBrokerName(brokerContainer.getBroker().getBrokerName());
bridge.setLocalSession(localConnection.createSession(false, Session.CLIENT_ACKNOWLEDGE));
bridge.setRemoteSession(remoteConnection.createSession(false, Session.CLIENT_ACKNOWLEDGE));
consumerMap.put(bridge, bridge);
bridge.start();
log.info("started NetworkMessageBridge for destination: " + destination);
}
catch (JMSException jmsEx) {
log.error("Failed to start NetworkMessageBridge for destination: " + destination);
}
}
bridge.incrementReferenceCount();
}
private void removeConsumerInfo(final ConsumerInfo info) {
NetworkMessageBridge key = new NetworkMessageBridge();
key.setDestination(info.getDestination());
key.setDurableTopic(info.isDurableTopic());
final NetworkMessageBridge bridge = (NetworkMessageBridge) consumerMap.get(key);
if (bridge != null) {
if (bridge.decrementReferenceCount() <= 0 && !bridge.isDurableTopic()
&& (bridge.getDestination().isTopic() || bridge.getDestination().isTemporary())) {
Thread runner = new Thread(new Runnable() {
public void run() {
bridge.stop();
consumerMap.remove(bridge);
log.info("stopped MetworkMessageBridge for destination: " + info.getDestination());
}
});
runner.setDaemon(true);
runner.start();
}
}
}
private void startSubscriptions() {
MessageContainerManager durableTopicMCM = brokerContainer.getBroker().getPersistentTopicContainerManager();
if (durableTopicMCM != null) {
Map map = durableTopicMCM.getDestinations();
startSubscriptions(map, true);
}
for (Iterator i = brokerContainer.getBroker().getContainerManagerMap().values().iterator();i.hasNext();) {
MessageContainerManager mcm = (MessageContainerManager) i.next();
if (mcm != durableTopicMCM) {
startSubscriptions(mcm.getDestinations(), false);
}
}
}
private void startSubscriptions(Map destinations, boolean durableTopic) {
if (destinations != null) {
for (Iterator i = destinations.values().iterator();i.hasNext();) {
ActiveMQDestination dest = (ActiveMQDestination) i.next();
addConsumerInfo(dest, durableTopic);
}
}
}
private void initialize() throws JMSException {
initializeRemote();
initializeLocal();
//start subscribing for already established subscriptions
}
private void initializeRemote() throws JMSException {
ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(remoteUserName, remotePassword, uri);
factory.setUseAsyncSend(true);
remoteConnection = (ActiveMQConnection) factory.createConnection();
remoteConnection.setClientID(brokerContainer.getBroker().getBrokerName() + "_NetworkChannel");
TransportChannel transportChannel = remoteConnection.getTransportChannel();
if (transportChannel instanceof CompositeTransportChannel) {
CompositeTransportChannel composite = (CompositeTransportChannel) transportChannel;
composite.setMaximumRetries(maximumRetries);
composite.setFailureSleepTime(reconnectSleepTime);
}
remoteConnection.start();
BrokerInfo info = new BrokerInfo();
info.setBrokerName(brokerContainer.getBroker().getBrokerName());
info.setClusterName(brokerContainer.getBroker().getBrokerClusterName());
Receipt receipt = remoteConnection.syncSendRequest(info);
remoteBrokerName = receipt.getBrokerName();
remoteClusterName = receipt.getClusterName();
}
private void initializeLocal() throws JMSException {
String brokerName = brokerContainer.getBroker().getBrokerName();
ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("vm://" + brokerName);
factory.setUseAsyncSend(true);
factory.setBrokerName(brokerName);
localConnection = (ActiveMQConnection) factory.createConnection();
localConnection.start();
BrokerInfo info = new BrokerInfo();
info.setBrokerName(remoteBrokerName);
info.setClusterName(remoteClusterName);
localConnection.asyncSendPacket(info);
}
}