/**
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You 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.apache.activemq.network;
import java.net.URI;
import java.util.Hashtable;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;
import javax.naming.CommunicationException;
import javax.naming.Context;
import javax.naming.NamingEnumeration;
import javax.naming.directory.Attributes;
import javax.naming.directory.DirContext;
import javax.naming.directory.InitialDirContext;
import javax.naming.directory.SearchControls;
import javax.naming.directory.SearchResult;
import javax.naming.event.EventDirContext;
import javax.naming.event.NamespaceChangeListener;
import javax.naming.event.NamingEvent;
import javax.naming.event.NamingExceptionEvent;
import javax.naming.event.ObjectChangeListener;
import org.apache.activemq.util.URISupport;
import org.apache.activemq.util.URISupport.CompositeData;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
/**
* class to create dynamic network connectors listed in an directory
* server using the LDAP v3 protocol as defined in RFC 2251, the
* entries listed in the directory server must implement the ipHost
* and ipService objectClasses as defined in RFC 2307.
*
* @author Trevor Pounds
* @see <a href="http://www.faqs.org/rfcs/rfc2251.html">RFC 2251</a>
* @see <a href="http://www.faqs.org/rfcs/rfc2307.html">RFC 2307</a>
*
* @org.apache.xbean.XBean element="ldapNetworkConnector"
*/
public class LdapNetworkConnector
extends NetworkConnector
implements NamespaceChangeListener,
ObjectChangeListener
{
private static final Log LOG = LogFactory.getLog(LdapNetworkConnector.class);
// force returned entries to implement the ipHost and ipService object classes (RFC 2307)
private static final String REQUIRED_OBJECT_CLASS_FILTER = "(&(objectClass=ipHost)(objectClass=ipService))";
// connection
private URI[] availableURIs = null;
private int availableURIsIndex = 0;
private String base = null;
private boolean failover = false;
private long curReconnectDelay = 1000; /* 1 sec */
private long maxReconnectDelay = 30000; /* 30 sec */
// authentication
private String user = null;
private String password = null;
private boolean anonymousAuthentication = false;
// search
private SearchControls searchControls = new SearchControls(/* ONELEVEL_SCOPE */);
private String searchFilter = REQUIRED_OBJECT_CLASS_FILTER;
private boolean searchEventListener = false;
// connector management
private Map<URI, NetworkConnector> connectorMap = new ConcurrentHashMap();
private Map<URI, Integer> referenceMap = new ConcurrentHashMap();
private Map<String, URI> uuidMap = new ConcurrentHashMap();
// local context
private DirContext context = null;
//currently in use URI
private URI ldapURI = null;
/**
* returns the next URI from the configured list
*
* @return random URI from the configured list
*/
public URI getUri()
{ return availableURIs[++availableURIsIndex % availableURIs.length]; }
/**
* sets the LDAP server URI
*
* @param _uri LDAP server URI
*/
public void setUri(URI _uri)
throws Exception
{
CompositeData data = URISupport.parseComposite(_uri);
if(data.getScheme().equals("failover"))
{
availableURIs = data.getComponents();
failover = true;
}
else
{ availableURIs = new URI[]{ _uri }; }
}
/**
* sets the base LDAP dn used for lookup operations
*
* @param _base LDAP base dn
*/
public void setBase(String _base)
{ base = _base; }
/**
* sets the LDAP user for access credentials
*
* @param _user LDAP dn of user
*/
public void setUser(String _user)
{ user = _user; }
/**
* sets the LDAP password for access credentials
*
* @param _password user password
*/
public void setPassword(String _password)
{ password = _password; }
/**
* sets LDAP anonymous authentication access credentials
*
* @param _anonymousAuthentication set to true to use anonymous authentication
*/
public void setAnonymousAuthentication(boolean _anonymousAuthentication)
{ anonymousAuthentication = _anonymousAuthentication; }
/**
* sets the LDAP search scope
*
* @param _searchScope LDAP JNDI search scope
*/
public void setSearchScope(String _searchScope)
throws Exception
{
int scope;
if(_searchScope.equals("OBJECT_SCOPE"))
{ scope = SearchControls.OBJECT_SCOPE; }
else if(_searchScope.equals("ONELEVEL_SCOPE"))
{ scope = SearchControls.ONELEVEL_SCOPE; }
else if(_searchScope.equals("SUBTREE_SCOPE"))
{ scope = SearchControls.SUBTREE_SCOPE; }
else
{ throw new Exception("ERR: unknown LDAP search scope specified: " + _searchScope); }
searchControls.setSearchScope(scope);
}
/**
* sets the LDAP search filter as defined in RFC 2254
*
* @param _searchFilter LDAP search filter
* @see <a href="http://www.faqs.org/rfcs/rfc2254.html">RFC 2254</a>
*/
public void setSearchFilter(String _searchFilter)
{ searchFilter = "(&" + REQUIRED_OBJECT_CLASS_FILTER + "(" + _searchFilter + "))"; }
/**
* enables/disable a persistent search to the LDAP server as defined
* in draft-ietf-ldapext-psearch-03.txt (2.16.840.1.113730.3.4.3)
*
* @param _searchEventListener enable = true, disable = false (default)
* @see <a href="http://www.ietf.org/proceedings/01mar/I-D/draft-ietf-ldapext-psearch-03.txt">draft-ietf-ldapext-psearch-03.txt</a>
*/
public void setSearchEventListener(boolean _searchEventListener)
{ searchEventListener = _searchEventListener; }
/**
* start the connector
*/
public void start()
throws Exception
{
LOG.info("connecting...");
Hashtable<String, String> env = new Hashtable();
env.put(Context.INITIAL_CONTEXT_FACTORY, "com.sun.jndi.ldap.LdapCtxFactory");
this.ldapURI = getUri();
LOG.debug(" URI [" + this.ldapURI + "]");
env.put(Context.PROVIDER_URL, this.ldapURI.toString());
if(anonymousAuthentication)
{
LOG.debug(" login credentials [anonymous]");
env.put(Context.SECURITY_AUTHENTICATION, "none");
}
else
{
LOG.debug(" login credentials [" + user + ":******]");
env.put(Context.SECURITY_PRINCIPAL, user);
env.put(Context.SECURITY_CREDENTIALS, password);
}
boolean isConnected = false;
while(!isConnected)
{
try
{
context = new InitialDirContext(env);
isConnected = true;
}
catch(CommunicationException err)
{
if(failover)
{
this.ldapURI = getUri();
LOG.error("connection error [" + env.get(Context.PROVIDER_URL) + "], failover connection to [" + this.ldapURI.toString() + "]");
env.put(Context.PROVIDER_URL, this.ldapURI.toString());
Thread.sleep(curReconnectDelay);
curReconnectDelay = Math.min(curReconnectDelay * 2, maxReconnectDelay);
}
else
{ throw err; }
}
}
// add connectors from search results
LOG.info("searching for network connectors...");
LOG.debug(" base [" + base + "]");
LOG.debug(" filter [" + searchFilter + "]");
LOG.debug(" scope [" + searchControls.getSearchScope() + "]");
NamingEnumeration<SearchResult> results = context.search(base, searchFilter, searchControls);
while(results.hasMore())
{ addConnector(results.next()); }
// register persistent search event listener
if(searchEventListener)
{
LOG.info("registering persistent search listener...");
EventDirContext eventContext = (EventDirContext)context.lookup("");
eventContext.addNamingListener(base, searchFilter, searchControls, this);
}
else // otherwise close context (i.e. connection as it is no longer needed)
{ context.close(); }
}
/**
* stop the connector
*/
public void stop()
throws Exception
{
LOG.info("stopping context...");
for(NetworkConnector connector : connectorMap.values())
{ connector.stop(); }
connectorMap.clear();
referenceMap.clear();
uuidMap.clear();
context.close();
}
/**
* returns the name of the connector
*
* @return connector name
*/
public String getName() {
String name = super.getName();
if (name == null) {
name = this.getClass().getName() + " [" + ldapURI.toString() + "]";
super.setName(name);
}
return name;
}
/**
* add connector of the given URI
*
* @param result
* search result of connector to add
*/
protected synchronized void addConnector(SearchResult result)
throws Exception
{
String uuid = toUUID(result);
if(uuidMap.containsKey(uuid))
{
LOG.warn("connector already regsitered for UUID [" + uuid + "]");
return;
}
URI connectorURI = toURI(result);
if(connectorMap.containsKey(connectorURI))
{
int referenceCount = referenceMap.get(connectorURI) + 1;
LOG.warn("connector reference added for URI [" + connectorURI + "], UUID [" + uuid + "], total reference(s) [" + referenceCount + "]");
referenceMap.put(connectorURI, referenceCount);
uuidMap.put(uuid, connectorURI);
return;
}
// FIXME: disable JMX listing of LDAP managed connectors, we will
// want to map/manage these differently in the future
// boolean useJMX = getBrokerService().isUseJmx();
// getBrokerService().setUseJmx(false);
NetworkConnector connector = getBrokerService().addNetworkConnector(connectorURI);
// getBrokerService().setUseJmx(useJMX);
// propogate std connector properties that may have been set via XML
connector.setDynamicOnly(isDynamicOnly());
connector.setDecreaseNetworkConsumerPriority(isDecreaseNetworkConsumerPriority());
connector.setNetworkTTL(getNetworkTTL());
connector.setConduitSubscriptions(isConduitSubscriptions());
connector.setExcludedDestinations(getExcludedDestinations());
connector.setDynamicallyIncludedDestinations(getDynamicallyIncludedDestinations());
connector.setDuplex(isDuplex());
// XXX: set in the BrokerService.startAllConnectors method and is
// required to prevent remote broker exceptions upon connection
connector.setLocalUri(getBrokerService().getVmConnectorURI());
connector.setBrokerName(getBrokerService().getBrokerName());
connector.setDurableDestinations(getBrokerService().getBroker().getDurableDestinations());
// start network connector
connectorMap.put(connectorURI, connector);
referenceMap.put(connectorURI, 1);
uuidMap.put(uuid, connectorURI);
connector.start();
LOG.info("connector added with URI [" + connectorURI + "]");
}
/**
* remove connector of the given URI
*
* @param result search result of connector to remove
*/
protected synchronized void removeConnector(SearchResult result)
throws Exception
{
String uuid = toUUID(result);
if(!uuidMap.containsKey(uuid))
{
LOG.warn("connector not regsitered for UUID [" + uuid + "]");
return;
}
URI connectorURI = uuidMap.get(uuid);
if(!connectorMap.containsKey(connectorURI))
{
LOG.warn("connector not regisitered for URI [" + connectorURI + "]");
return;
}
int referenceCount = referenceMap.get(connectorURI) - 1;
referenceMap.put(connectorURI, referenceCount);
uuidMap.remove(uuid);
LOG.debug("connector referenced removed for URI [" + connectorURI + "], UUID [" + uuid + "], remaining reference(s) [" + referenceCount + "]");
if(referenceCount > 0)
{ return; }
NetworkConnector connector = connectorMap.remove(connectorURI);
connector.stop();
LOG.info("connector removed with URI [" + connectorURI + "]");
}
/**
* convert search result into URI
*
* @param result search result to convert to URI
*/
protected URI toURI(SearchResult result)
throws Exception
{
Attributes attributes = result.getAttributes();
String address = (String)attributes.get("iphostnumber").get();
String port = (String)attributes.get("ipserviceport").get();
String protocol = (String)attributes.get("ipserviceprotocol").get();
URI connectorURI = new URI("static:(" + protocol + "://" + address + ":" + port + ")");
LOG.debug("retrieved URI from SearchResult [" + connectorURI + "]");
return connectorURI;
}
/**
* convert search result into URI
*
* @param result search result to convert to URI
*/
protected String toUUID(SearchResult result)
{
String uuid = result.getNameInNamespace();
LOG.debug("retrieved UUID from SearchResult [" + uuid + "]");
return uuid;
}
/**
* invoked when an entry has been added during a persistent search
*/
public void objectAdded(NamingEvent event)
{
LOG.debug("entry added");
try
{ addConnector((SearchResult)event.getNewBinding()); }
catch(Exception err)
{ LOG.error("ERR: caught unexpected exception", err); }
}
/**
* invoked when an entry has been removed during a persistent search
*/
public void objectRemoved(NamingEvent event)
{
LOG.debug("entry removed");
try
{ removeConnector((SearchResult)event.getOldBinding()); }
catch(Exception err)
{ LOG.error("ERR: caught unexpected exception", err); }
}
/**
* invoked when an entry has been renamed during a persistent search
*/
public void objectRenamed(NamingEvent event)
{
LOG.debug("entry renamed");
// XXX: getNameInNamespace method does not seem to work properly,
// but getName seems to provide the result we want
String uuidOld = event.getOldBinding().getName();
String uuidNew = event.getNewBinding().getName();
URI connectorURI = uuidMap.remove(uuidOld);
uuidMap.put(uuidNew, connectorURI);
LOG.debug("connector reference renamed for URI [" + connectorURI + "], Old UUID [" + uuidOld + "], New UUID [" + uuidNew + "]");
}
/**
* invoked when an entry has been changed during a persistent search
*/
public void objectChanged(NamingEvent event)
{
LOG.debug("entry changed");
try
{
SearchResult result = (SearchResult)event.getNewBinding();
removeConnector(result);
addConnector(result);
}
catch(Exception err)
{ LOG.error("ERR: caught unexpected exception", err); }
}
/**
* invoked when an exception has occurred during a persistent search
*/
public void namingExceptionThrown(NamingExceptionEvent event)
{ LOG.error("ERR: caught unexpected exception", event.getException()); }
}