capsd.java

来自「opennms得相关源码 请大家看看」· Java 代码 · 共 487 行 · 第 1/2 页

//
// This file is part of the OpenNMS(R) Application.
//
// OpenNMS(R) is Copyright (C) 2002-2003 The OpenNMS Group, Inc.  All rights reserved.
// OpenNMS(R) is a derivative work, containing both original code, included code and modified
// code that was published under the GNU General Public License. Copyrights for modified 
// and included code are below.
//
// OpenNMS(R) is a registered trademark of The OpenNMS Group, Inc.
//
// Modifications:
//
// 2003 Oct 08: Changed thread poll water marks.
// 2003 Jan 31: Cleaned up some unused imports.
//
// Original code base Copyright (C) 1999-2001 Oculan Corp.  All rights reserved.
//
// This program is free software; you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation; either version 2 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU General Public License for more details.                                                            
//
// You should have received a copy of the GNU General Public License
// along with this program; if not, write to the Free Software
// Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
//       
// For more information contact: 
//      OpenNMS Licensing       <[e-mail]>
//      http://www.opennms.org/
//      http://www.opennms.com/
//
// Tab Stop = 8
//

package org.opennms.netmgt.capsd;

import java.io.IOException;
import java.lang.reflect.UndeclaredThrowableException;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.sql.SQLException;

import org.apache.log4j.Category;
import org.exolab.castor.xml.MarshalException;
import org.exolab.castor.xml.ValidationException;
import org.opennms.core.concurrent.RunnableConsumerThreadPool;
import org.opennms.core.fiber.PausableFiber;
import org.opennms.core.utils.ThreadCategory;
import org.opennms.netmgt.config.CapsdConfigFactory;
import org.opennms.netmgt.config.CollectdConfigFactory;
import org.opennms.netmgt.config.DatabaseConnectionFactory;
import org.opennms.netmgt.config.PollerConfigFactory;
import org.opennms.netmgt.config.SnmpPeerFactory;

/**
 * <P>
 * The Capability daemon - it is notified by the discovery process when a new
 * node is discovered - it then polls for all the capabilities for this node and
 * is responsible for loading the data collecte1d into the database.
 * </P>
 * 
 * <P>
 * Once a node is added to the database, its sends an indication back to the
 * discovery which then flags this node as 'known'.
 * </P>
 * 
 * @author <A HREF="mailto:[e-mail]">Mike Davidson </A>
 * @author <A HREF="http://www.opennms.org/">OpenNMS </A>
 * 
 */
public class Capsd implements PausableFiber {
    /**
     * The log4j category used to log messages.
     */
    private static final String LOG4J_CATEGORY = "OpenNMS.Capsd";

    /**
     * Singleton instance of the Capsd class
     */
    private static final Capsd m_singleton = new Capsd();

    /**
     * Current status of this fiber
     */
    private int m_status;

    /**
     * Database synchronization lock for synchronizing write access to the
     * database between the SuspectEventProcessor and RescanProcessor thread
     * pools
     */
    private static Object m_dbSyncLock = new Object();

    /**
     * <P>
     * Contains dotted-decimal representation of the IP address where Capsd is
     * running. Used when capsd sends events out
     * </P>
     */
    private static String m_address = null;

    /**
     * Rescan scheduler thread
     */
    private Scheduler m_scheduler;

    /**
     * Event receiver.
     */
    private BroadcastEventProcessor m_receiver;

    /**
     * The pool of threads that are used to executed the SuspectEventProcessor
     * instances queued by the event processor (BroadcastEventProcessor).
     */
    private RunnableConsumerThreadPool m_suspectRunner;

    /**
     * The pool of threads that are used to executed RescanProcessor instances
     * queued by the rescan scheduler thread.
     */
    private RunnableConsumerThreadPool m_rescanRunner;

    /**
     * <P>
     * Static initialization
     * </P>
     */

    static {
        try {
            m_address = InetAddress.getLocalHost().getHostAddress();
        } catch (UnknownHostException uhE) {
            m_address = "localhost";
            ThreadCategory.getInstance(LOG4J_CATEGORY).warn("Could not lookup the host name for the local host machine, address set to localhost", uhE);
        }
    } // end static class initialization

    /**
     * Constructs the Capsd objec
     */
    public Capsd() {
        m_scheduler = null;
        m_status = START_PENDING;
    }

    /**
     * Stop the Capsd threads.
     */
    public void stop() {
        m_status = STOP_PENDING;

        // Stop the broadcast event receiver
        //
        m_receiver.close();

        // Stop the Suspect Event Processor thread pool
        //
        m_suspectRunner.stop();

        // Stop the Rescan Processor thread pool
        //
        m_rescanRunner.stop();

        m_status = STOPPED;
    }

    /**
     * Start the Capsd threads.
     */
    public void init() {
        ThreadCategory.setPrefix(LOG4J_CATEGORY);
        Category log = ThreadCategory.getInstance();

        // Initialize the Capsd configuration factory.
        //
        try {
            CapsdConfigFactory.reload();
        } catch (MarshalException ex) {
            log.error("Failed to load Capsd configuration", ex);
            throw new UndeclaredThrowableException(ex);
        } catch (ValidationException ex) {
            log.error("Failed to load Capsd configuration", ex);
            throw new UndeclaredThrowableException(ex);
        } catch (IOException ex) {
            log.error("Failed to load Capsd configuration", ex);
            throw new UndeclaredThrowableException(ex);
        }

        // Initialize the poller configuration factory.
        //
        try {
            PollerConfigFactory.reload();
        } catch (MarshalException ex) {
            log.error("Failed to load poller configuration", ex);
            throw new UndeclaredThrowableException(ex);
        } catch (ValidationException ex) {
            log.error("Failed to load poller configuration", ex);
            throw new UndeclaredThrowableException(ex);
        } catch (IOException ex) {
            log.error("Failed to load poller configuration", ex);
            throw new UndeclaredThrowableException(ex);
        }

        // Initialize the collectd configuration factory.
        //
        try {
            CollectdConfigFactory.reload();
        } catch (MarshalException ex) {
            log.error("Failed to load collectd configuration", ex);
            throw new UndeclaredThrowableException(ex);
        } catch (ValidationException ex) {
            log.error("Failed to load collectd configuration", ex);
            throw new UndeclaredThrowableException(ex);
        } catch (IOException ex) {
            log.error("Failed to load collectd configuration", ex);
            throw new UndeclaredThrowableException(ex);
        }

        // Initialize the Database configuration factory
        //
        try {
            DatabaseConnectionFactory.init();
        } catch (IOException ie) {
            log.fatal("IOException loading database config", ie);
            throw new UndeclaredThrowableException(ie);
        } catch (MarshalException me) {
            log.fatal("Marshall Exception loading database config", me);
            throw new UndeclaredThrowableException(me);
        } catch (ValidationException ve) {
            log.fatal("Validation Exception loading database config", ve);
            throw new UndeclaredThrowableException(ve);
        } catch (ClassNotFoundException ce) {
            log.fatal("Class lookup failure loading database config", ce);
            throw new UndeclaredThrowableException(ce);
        }

        // Initialize the SNMP Peer Factory
        //