scheduler.java

opennms得相关源码 请大家看看

        // Retrieve last poll time for the node from the ipInterface table.
        Connection db = null;
        try {
            db = DatabaseConnectionFactory.getInstance().getConnection();
            PreparedStatement ifStmt = db.prepareStatement(SQL_GET_LAST_POLL_TIME);
            ifStmt.setString(1, address.getHostAddress());
            ResultSet rset = ifStmt.executeQuery();
            if (rset.next()) {
                Timestamp lastPolled = rset.getTimestamp(1);
                if (lastPolled != null && rset.wasNull() == false) {
                    if (log.isDebugEnabled()) {
                        log.debug("scheduleAddress: adding node " + address + " with last poll time " + lastPolled);
		    }
	            m_knownAddresses.add(new NessusScanConfiguration(address, scanLevel, lastPolled, m_interval));
                }
            } else {
                if (log.isDebugEnabled()) {
                    log.debug("scheduleAddress: adding ipAddr " + address + " with no previous poll");
		}
                m_knownAddresses.add(new NessusScanConfiguration(address, scanLevel, new Timestamp(0), m_interval));
            }
        } finally {
            if (db != null) {
                try {
                    db.close();
                } catch (Exception e) {
                }
            }
        }
    }

    /**
     * Removes the specified node from the known node list.
     * 
     * @param address
     *            Address of interface to be removed.
     */
    void unscheduleAddress(InetAddress address) {
        synchronized (m_knownAddresses) {
            Iterator iter = m_knownAddresses.iterator();
            while (iter.hasNext()) {
                NessusScanConfiguration addressInfo = (NessusScanConfiguration) iter.next();
                if (addressInfo.getAddress() == address) {
                    ThreadCategory.getInstance(getClass()).debug("unscheduleAddress: removing node " + address + " from the scheduler.");
                    m_knownAddresses.remove(addressInfo);
                    break;
                }
            }
        }
    }

    public static InetAddress toInetAddress(long address) throws UnknownHostException {
        StringBuffer buf = new StringBuffer();
        buf.append((int) ((address >>> 24) & 0xff));
	buf.append('.');
        buf.append((int) ((address >>> 16) & 0xff));
	buf.append('.');
        buf.append((int) ((address >>> 8) & 0xff));
	buf.append('.');
        buf.append((int) (address & 0xff));
        return InetAddress.getByName(buf.toString());
    }

    /**
     * Starts the fiber.
     * 
     * @throws java.lang.IllegalStateException
     *             Thrown if the fiber is already running.
     */
    public synchronized void start() {
        if (m_worker != null)
            throw new IllegalStateException("The fiber has already run or is running");

        Category log = ThreadCategory.getInstance(getClass());

        m_worker = new Thread(this, getName());
        m_worker.start();
        m_status = STARTING;

        log.debug("Scheduler.start: scheduler started");
    }

    /**
     * Stops the fiber. If the fiber has never been run then an exception is
     * generated.
     * 
     * @throws java.lang.IllegalStateException
     *             Throws if the fiber has never been started.
     */
    public synchronized void stop() {
        if (m_worker == null)
            throw new IllegalStateException("The fiber has never been started");

        Category log = ThreadCategory.getInstance(getClass());

        m_status = STOP_PENDING;
        m_worker.interrupt();

        log.debug("Scheduler.stop: scheduler stopped");
    }

    /**
     * Pauses the scheduler if it is current running. If the fiber has not been
     * run or has already stopped then an exception is generated.
     * 
     * @throws java.lang.IllegalStateException
     *             Throws if the operation could not be completed due to the
     *             fiber's state.
     */
    public synchronized void pause() {
        if (m_worker == null)
            throw new IllegalStateException("The fiber has never been started");

        if (m_status == STOPPED || m_status == STOP_PENDING)
            throw new IllegalStateException("The fiber is not running or a stop is pending");

        if (m_status == PAUSED)
            return;

        m_status = PAUSE_PENDING;
        notifyAll();
    }

    /**
     * Resumes the scheduler if it has been paused. If the fiber has not been
     * run or has already stopped then an exception is generated.
     * 
     * @throws java.lang.IllegalStateException
     *             Throws if the operation could not be completed due to the
     *             fiber's state.
     */
    public synchronized void resume() {
        if (m_worker == null)
            throw new IllegalStateException("The fiber has never been started");

        if (m_status == STOPPED || m_status == STOP_PENDING)
            throw new IllegalStateException("The fiber is not running or a stop is pending");

        if (m_status == RUNNING)
            return;

        m_status = RESUME_PENDING;
        notifyAll();
    }

    /**
     * Returns the current of this fiber.
     * 
     * @return The current status.
     */
    public synchronized int getStatus() {
        if (m_worker != null && m_worker.isAlive() == false)
            m_status = STOPPED;
        return m_status;
    }

    /**
     * Returns the name of this fiber.
     * 
     */
    public String getName() {
        return FIBER_NAME;
    }

    /**
     * The main method of the scheduler. This method is responsible for checking
     * the runnable queues for ready objects and then enqueuing them into the
     * thread pool for execution.
     * 
     */
    public void run() {
        Category log = ThreadCategory.getInstance(getClass());

        synchronized (this) {
            m_status = RUNNING;
        }
        log.debug("Scheduler.run: scheduler running");

        /*
	 * Loop until a fatal exception occurs or until the thread
	 * is interrupted.
	 */
        boolean firstPass = true;
        while (true) {
            // Status check
            synchronized (this) {
                if (m_status != RUNNING && m_status != PAUSED && m_status != PAUSE_PENDING && m_status != RESUME_PENDING) {
                    log.debug("Scheduler.run: status = " + m_status + ", time to exit");
                    break;
                }
            }

            /*
             * If this is the first pass we want to pause momentarily
             * This allows the rest of the background processes to come
             * up and stabilize before we start generating events from rescans.
             */
            if (firstPass) {
                firstPass = false;
                synchronized (this) {
                    try {
                        log.debug("Scheduler.run: initial sleep configured for " + m_initialSleep + "ms...sleeping...");
                        wait(m_initialSleep);
                    } catch (InterruptedException ex) {
                        log.debug("Scheduler.run: interrupted exception during initial sleep...exiting.");
                        break; // exit for loop
                    }
                }
            }

            // iterate over the known node list, add any
            // nodes ready for rescan to the rescan queue
            // for processing.
            //
            int added = 0;

            synchronized (m_knownAddresses) {

                log.debug("Scheduler.run: iterating over known nodes list to schedule...");
                Iterator iter = m_knownAddresses.iterator();
                while (iter.hasNext()) {
                    NessusScanConfiguration addressInfo = (NessusScanConfiguration) iter.next();
		    log.debug("Scheduler.run: working on "  + addressInfo.getAddress().toString() );
                    // Don't schedule if already scheduled
                    if (addressInfo.isScheduled())
                        continue;

                    // Don't schedule if its not time for rescan yet
                    if (!addressInfo.isTimeForRescan())
                        continue;

                    // Must be time for a rescan!
                    //
                    try {
                        addressInfo.setScheduled(true); // Mark node as
                                                        // scheduled

                        // Create a new NessusScan object
                        // and add it to the rescan queue for execution
                        //
                        log.debug("Scheduler.run: adding node " + addressInfo.getAddress().toString() + " to the rescan queue.");

                        m_scheduledScanQ.add(new NessusScan(addressInfo));
                        added++;
                    } catch (InterruptedException ex) {
                        log.info("Scheduler.schedule: failed to add new node to rescan queue", ex);
                        throw new UndeclaredThrowableException(ex);
                    } catch (FifoQueueException ex) {
                        log.info("Scheduler.schedule: failed to add new node to rescan queue", ex);
                        throw new UndeclaredThrowableException(ex);
                    }
                }
            }

            // Wait for 60 seconds if there were no nodes
            // added to the rescan queue during this loop,
            // otherwise just start over.
            //
            synchronized (this) {
                if (added == 0) {
                    try {
                        wait(60000);
                    } catch (InterruptedException ex) {
                        break; // exit for loop
                    }
                }
            }
        } // end while(true)

        log.debug("Scheduler.run: scheduler exiting, state = STOPPED");
        synchronized (this) {
            m_status = STOPPED;
        }
    } // end run
}