001    /**
002     * Licensed to the Apache Software Foundation (ASF) under one or more
003     * contributor license agreements.  See the NOTICE file distributed with
004     * this work for additional information regarding copyright ownership.
005     * The ASF licenses this file to You under the Apache License, Version 2.0
006     * (the "License"); you may not use this file except in compliance with
007     * the License.  You may obtain a copy of the License at
008     *
009     *      http://www.apache.org/licenses/LICENSE-2.0
010     *
011     * Unless required by applicable law or agreed to in writing, software
012     * distributed under the License is distributed on an "AS IS" BASIS,
013     * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
014     * See the License for the specific language governing permissions and
015     * limitations under the License.
016     */
017    package org.apache.activemq.transport.discovery;
018    
019    import java.net.URI;
020    import java.net.URISyntaxException;
021    import java.util.Map;
022    import java.util.concurrent.ConcurrentHashMap;
023    
024    import org.apache.activemq.command.DiscoveryEvent;
025    import org.apache.activemq.transport.CompositeTransport;
026    import org.apache.activemq.transport.TransportFilter;
027    import org.apache.activemq.util.ServiceStopper;
028    import org.apache.activemq.util.URISupport;
029    import org.apache.commons.logging.Log;
030    import org.apache.commons.logging.LogFactory;
031    
032    /**
033     * A {@link ReliableTransportChannel} which uses a {@link DiscoveryAgent} to
034     * discover remote broker instances and dynamically connect to them.
035     * 
036     * @version $Revision$
037     */
038    public class DiscoveryTransport extends TransportFilter implements DiscoveryListener {
039    
040        private static final Log LOG = LogFactory.getLog(DiscoveryTransport.class);
041    
042        private final CompositeTransport next;
043        private DiscoveryAgent discoveryAgent;
044        private final ConcurrentHashMap<String, URI> serviceURIs = new ConcurrentHashMap<String, URI>();
045    
046        private Map<String, String> parameters;
047    
048        public DiscoveryTransport(CompositeTransport next) {
049            super(next);
050            this.next = next;
051        }
052    
053        public void start() throws Exception {
054            if (discoveryAgent == null) {
055                throw new IllegalStateException("discoveryAgent not configured");
056            }
057    
058            // lets pass into the agent the broker name and connection details
059            discoveryAgent.setDiscoveryListener(this);
060            discoveryAgent.start();
061            next.start();
062        }
063    
064        public void stop() throws Exception {
065            ServiceStopper ss = new ServiceStopper();
066            ss.stop(discoveryAgent);
067            ss.stop(next);
068            ss.throwFirstException();
069        }
070    
071        public void onServiceAdd(DiscoveryEvent event) {
072            String url = event.getServiceName();
073            if (url != null) {
074                try {
075                    URI uri = new URI(url);
076                    serviceURIs.put(event.getServiceName(), uri);
077                    LOG.info("Adding new broker connection URL: " + uri);
078                    next.add(new URI[] {URISupport.applyParameters(uri, parameters)});
079                } catch (URISyntaxException e) {
080                    LOG.warn("Could not connect to remote URI: " + url + " due to bad URI syntax: " + e, e);
081                }
082            }
083        }
084    
085        public void onServiceRemove(DiscoveryEvent event) {
086            URI uri = serviceURIs.get(event.getServiceName());
087            if (uri != null) {
088                next.remove(new URI[] {uri});
089            }
090        }
091    
092        public DiscoveryAgent getDiscoveryAgent() {
093            return discoveryAgent;
094        }
095    
096        public void setDiscoveryAgent(DiscoveryAgent discoveryAgent) {
097            this.discoveryAgent = discoveryAgent;
098        }
099    
100        public void setParameters(Map<String, String> parameters) {
101           this.parameters = parameters;      
102        }
103    
104    }