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 */
017package org.apache.activemq.transport.discovery;
018
019import java.net.URI;
020import java.net.URISyntaxException;
021import java.util.Map;
022import java.util.concurrent.ConcurrentHashMap;
023import org.apache.activemq.command.DiscoveryEvent;
024import org.apache.activemq.transport.CompositeTransport;
025import org.apache.activemq.transport.TransportFilter;
026import org.apache.activemq.util.ServiceStopper;
027import org.apache.activemq.util.Suspendable;
028import org.apache.activemq.util.URISupport;
029import org.slf4j.Logger;
030import 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 */
038public 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}