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    import org.apache.activemq.command.DiscoveryEvent;
024    import org.apache.activemq.transport.CompositeTransport;
025    import org.apache.activemq.transport.TransportFilter;
026    import org.apache.activemq.util.ServiceStopper;
027    import org.apache.activemq.util.Suspendable;
028    import org.apache.activemq.util.URISupport;
029    import org.slf4j.Logger;
030    import org.slf4j.LoggerFactory;
031    
032    /**
033     * A {@link ReliableTransportChannel} which uses a {@link DiscoveryAgent} to
034     * discover remote broker instances and dynamically connect to them.
035     * 
036     * 
037     */
038    public class DiscoveryTransport extends TransportFilter implements DiscoveryListener {
039    
040        private static final Logger LOG = LoggerFactory.getLogger(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        @Override
054        public void start() throws Exception {
055            if (discoveryAgent == null) {
056                throw new IllegalStateException("discoveryAgent not configured");
057            }
058    
059            // lets pass into the agent the broker name and connection details
060            discoveryAgent.setDiscoveryListener(this);
061            discoveryAgent.start();
062            next.start();
063        }
064    
065        @Override
066        public void stop() throws Exception {
067            ServiceStopper ss = new ServiceStopper();
068            ss.stop(discoveryAgent);
069            ss.stop(next);
070            ss.throwFirstException();
071        }
072    
073        public void onServiceAdd(DiscoveryEvent event) {
074            String url = event.getServiceName();
075            if (url != null) {
076                try {
077                    URI uri = new URI(url);
078                    LOG.info("Adding new broker connection URL: " + uri);
079                    uri = URISupport.applyParameters(uri, parameters, DISCOVERED_OPTION_PREFIX);
080                    serviceURIs.put(event.getServiceName(), uri);
081                    next.add(false,new URI[] {uri});
082                } catch (URISyntaxException e) {
083                    LOG.warn("Could not connect to remote URI: " + url + " due to bad URI syntax: " + e, e);
084                }
085            }
086        }
087    
088        public void onServiceRemove(DiscoveryEvent event) {
089            URI uri = serviceURIs.get(event.getServiceName());
090            if (uri != null) {
091                next.remove(false,new URI[] {uri});
092            }
093        }
094    
095        public DiscoveryAgent getDiscoveryAgent() {
096            return discoveryAgent;
097        }
098    
099        public void setDiscoveryAgent(DiscoveryAgent discoveryAgent) {
100            this.discoveryAgent = discoveryAgent;
101        }
102    
103        public void setParameters(Map<String, String> parameters) {
104           this.parameters = parameters;      
105        }
106    
107        @Override
108        public void transportResumed() {
109            if( discoveryAgent instanceof Suspendable ) {
110                try {
111                    ((Suspendable)discoveryAgent).suspend();
112                } catch (Exception e) {
113                    e.printStackTrace();
114                }
115            }
116            super.transportResumed();
117        }
118    
119        @Override
120        public void transportInterupted() {
121            if( discoveryAgent instanceof Suspendable ) {
122                try {
123                    ((Suspendable)discoveryAgent).resume();
124                } catch (Exception e) {
125                    e.printStackTrace();
126                }
127            }
128            super.transportInterupted();
129        }
130    }