/**
 * 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.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.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 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.
 *
 * @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 Logger LOG = LoggerFactory.getLogger(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<URI, NetworkConnector>();
    private Map<URI, Integer> referenceMap = new ConcurrentHashMap<URI, Integer>();
    private Map<String, URI> uuidMap = new ConcurrentHashMap<String, URI>();

    // 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) {
        this.base = base;
    }

    /**
     * sets the LDAP user for access credentials
     *
     * @param user
     *            LDAP dn of user
     */
    public void setUser(String user) {
        this.user = user;
    }

    /**
     * sets the LDAP password for access credentials
     *
     * @param password
     *            user password
     */
    @Override
    public void setPassword(String password) {
        this.password = password;
    }

    /**
     * sets LDAP anonymous authentication access credentials
     *
     * @param anonymousAuthentication
     *            set to true to use anonymous authentication
     */
    public void setAnonymousAuthentication(boolean anonymousAuthentication) {
        this.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) {
        this.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) {
        this.searchEventListener = searchEventListener;
    }

    /**
     * start the connector
     */
    @Override
    public void start() throws Exception {
        LOG.info("connecting...");
        Hashtable<String, String> env = new Hashtable<String, String>();
        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);
            if (user != null && !"".equals(user)) {
                env.put(Context.SECURITY_PRINCIPAL, user);
            } else {
                throw new Exception("Empty username is not allowed");
            }
            if (password != null && !"".equals(password)) {
                env.put(Context.SECURITY_CREDENTIALS, password);
            } else {
                throw new Exception("Empty password is not allowed");
            }
        }
        boolean isConnected = false;
        while (!isConnected) {
            try {
                context = new InitialDirContext(env);
                isConnected = true;
            } catch (CommunicationException err) {
                if (failover) {
                    this.ldapURI = getUri();
                    LOG.error("connection error [{}], failover connection to [{}]", env.get(Context.PROVIDER_URL), 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
     */
    @Override
    public void stop() throws Exception {
        LOG.info("stopping context...");
        for (NetworkConnector connector : connectorMap.values()) {
            connector.stop();
        }
        connectorMap.clear();
        referenceMap.clear();
        uuidMap.clear();
        context.close();
    }

    @Override
    public String toString() {
        return this.getClass().getName() + getName() + "[" + ldapURI.toString() + "]";
    }

    /**
     * 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 [{}], UUID [{}], total reference(s) [{}]", new Object[]{ connectorURI, uuid, 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);

        // Propagate standard connector properties that may have been set via XML
        connector.setDynamicOnly(isDynamicOnly());
        connector.setDecreaseNetworkConsumerPriority(isDecreaseNetworkConsumerPriority());
        connector.setNetworkTTL(getNetworkTTL());
        connector.setConsumerTTL(getConsumerTTL());
        connector.setMessageTTL(getMessageTTL());
        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 registered for UUID [{}]", uuid);
            return;
        }

        URI connectorURI = uuidMap.get(uuid);
        if (!connectorMap.containsKey(connectorURI)) {
            LOG.warn("connector not registered for URI [{}]", connectorURI);
            return;
        }

        int referenceCount = referenceMap.get(connectorURI) - 1;
        referenceMap.put(connectorURI, referenceCount);
        uuidMap.remove(uuid);
        LOG.debug("connector referenced removed for URI [{}], UUID[{}], remaining reference(s) [{}]", new Object[]{ connectorURI, uuid, 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
     */
    @Override
    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
     */
    @Override
    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
     */
    @Override
    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 [{}], Old UUID [{}], New UUID [{}]", new Object[]{ connectorURI, uuidOld, uuidNew });
    }

    /**
     * invoked when an entry has been changed during a persistent search
     */
    @Override
    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
     */
    @Override
    public void namingExceptionThrown(NamingExceptionEvent event) {
        LOG.error("ERR: caught unexpected exception", event.getException());
    }
}
