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 }