/**
*
* 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 java.net.URI;
import java.net.URISyntaxException;
import javax.jms.JMSException;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.codehaus.activemq.broker.BrokerContainer;
import org.codehaus.activemq.broker.impl.BrokerConnectorImpl;
import org.codehaus.activemq.io.impl.DefaultWireFormat;
import org.codehaus.activemq.message.BrokerInfo;
import org.codehaus.activemq.transport.composite.CompositeTransportChannel;
import EDU.oswego.cs.dl.util.concurrent.PooledExecutor;
/**
* Represents a Boondocks 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.1 $
*/
public class RemoteNetworkChannel extends NetworkChannel implements TransportStatusEventListener {
private static final Log log = LogFactory.getLog(RemoteNetworkChannel.class);
private TransportChannel boondocksChannel;
/**
* Default Constructor
*
* @param tp
*/
public RemoteNetworkChannel(PooledExecutor tp) {
super(tp);
}
/**
* Constructor
*
* @param connector
* @param brokerContainer
* @param uri
*/
public RemoteNetworkChannel(NetworkConnector connector, BrokerContainer brokerContainer, String uri) {
super(connector,brokerContainer,uri);
}
/**
* Create a RemoteNetworkConnector from a TransportChannel
*
* @param connector
* @param brokerContainer
* @param channel
* @throws JMSException
*/
public RemoteNetworkChannel(NetworkConnector connector, BrokerContainer brokerContainer, TransportChannel channel)
throws JMSException {
super(connector,brokerContainer,channel);
}
/**
* @see org.codehaus.activemq.transport.TransportStatusEventListener#statusChanged(org.codehaus.activemq.transport.TransportStatusEvent)
*/
public void statusChanged(TransportStatusEvent event) {
if (event.getChannelStatus()== TransportStatusEvent.RECONNECTED){
try {
sendBrokerInfo();
}
catch (JMSException e) {
log.error("Failed to send Broker Info",e);
}
}
}
/**
* remote:// can only make outgoing connections - we assume we can't
* accept incomming (duck!). So we initialize the transport channel
* from this side and create the broker client as well
* @throws JMSException
*/
protected void initialize() throws JMSException {
super.initialize();
try {
boondocksChannel = TransportChannelProvider.create(new DefaultWireFormat(), new URI(uri));
boondocksChannel.addTransportStatusEventListener(this);
if (boondocksChannel instanceof CompositeTransportChannel) {
CompositeTransportChannel composite = (CompositeTransportChannel)boondocksChannel;
composite.setMaximumRetries(maximumRetries);
composite.setFailureSleepTime(reconnectSleepTime);
}
boondocksChannel.start();
//create our own broker connector ...
BrokerConnectorImpl connector = new BrokerConnectorImpl(getBrokerContainer(),"vm://uri",new DefaultWireFormat());
connector.start();
connector.addClient(boondocksChannel);
sendBrokerInfo();
}
catch (URISyntaxException e) {
log.error("Could not parse uri: " + uri + " to make remote connector",e);
}
}
private void sendBrokerInfo() throws JMSException{
//inform the other side we are a remote channel
BrokerInfo info = new BrokerInfo();
info.setBrokerName(brokerContainer.getBroker().getBrokerName());
info.setClusterName(brokerContainer.getBroker().getBrokerClusterName());
info.setRemote(true);
boondocksChannel.asyncSend(info);
}
}