defaultconnectionpool.java

OpenJMS是一个开源的Java Message Service API 1.0.2 规范的实现,它包含有以下特性： *. 它既支持点到点（point-to-point）（PTP）模型和发布/订

     *
     * @param source the managed connection that is the source of the event
     */
    public void pinged(ManagedConnection source) {
        ManagedConnectionHandle handle
                = (ManagedConnectionHandle) _handles.get(source);
        if (handle != null) {
            handle.pinged();
        }
    }

    /**
     * Closes this connection pool, cleaning up any allocated resources.
     *
     * @throws ResourceException for any error
     */
    public void close() throws ResourceException {
        ManagedConnectionAcceptor[] acceptors =
                (ManagedConnectionAcceptor[]) _acceptors.toArray(
                        new ManagedConnectionAcceptor[0]);
        _acceptors.clear();

        for (int i = 0; i < acceptors.length; ++i) {
            acceptors[i].close();
        }

        ManagedConnection[] connections =
                (ManagedConnection[]) _entries.keySet().toArray(
                        new ManagedConnection[0]);
        for (int i = 0; i < connections.length; ++i) {
            connections[i].destroy();
        }
        _entries.clear();

        _accepted.clear();
        _connections.clear();

        stopReaper();
    }

    /**
     * Invoked when an acceptor receives an error.
     *
     * @param acceptor  the acceptor which received the error
     * @param throwable the error
     */
    public void error(ManagedConnectionAcceptor acceptor,
                      Throwable throwable) {
        _acceptors.remove(acceptor);

        String uri = "<unknown>";
        try {
            uri = acceptor.getURI().toString();
        } catch (ResourceException ignore) {
            // no-op
        }
        _log.error("Failed to accept connections on URI=" + uri,
                   throwable);

        try {
            acceptor.close();
        } catch (ResourceException exception) {
            if (_log.isDebugEnabled()) {
                _log.debug("Failed to close acceptor, URI=" + uri, exception);
            }
        }
    }

    /**
     * Sets the listener for caller events.
     *
     * @param listener the listener
     */
    public void setCallerListener(CallerListener listener) {
        _listener = listener;
    }

    /**
     * Notifies when a managed connection is idle.
     *
     * @param connection the idle connection
     */
    protected synchronized void idle(ManagedConnectionHandle connection) {
        connection.clearUsed();
        if (_daemon != null) {
            _daemon.executeAfterDelay(_idlePeriod, new IdleReaper());
        }
    }

    /**
     * Adds a connection to the pool. If the connection was created, a {@link
     * ManagedConnectionHandle} will be returned, wrapping the supplied
     * connection.
     *
     * @param connection the connection to add
     * @param accepted   if <code>true</code> the connection was accepted via an
     *                   {@link ManagedConnectionAcceptor}, otherwise it was
     *                   created via
     *                   {@link ManagedConnectionFactory#createManagedConnection}
     * @return the (possibly wrapped) connection
     * @throws ResourceException if the connection cannot be added
     */
    protected ManagedConnection add(ManagedConnection connection,
                                    boolean accepted) throws ResourceException {
        ManagedConnection result;

        PoolEntry entry = new PoolEntry(connection, accepted);
        _entries.put(connection, entry);
        if (accepted) {
            _accepted.add(connection);
            result = connection;
        } else {
            _connections.add(connection);
            ManagedConnection handle = new ManagedConnectionHandle(
                    this, connection, _resolver);
            _handles.put(connection, handle);
            result = handle;
        }
        ContextInvocationHandler handler = new ContextInvocationHandler(
                _handler, _resolver, result);
        try {
            connection.setInvocationHandler(handler);
            connection.setConnectionEventListener(this);
        } catch (ResourceException exception) {
            try {
                _log.debug("Failed to initialise connection, destroying",
                           exception);
                connection.destroy();
            } catch (ResourceException nested) {
                _log.debug("Failed to destroy connection", nested);
            } finally {
                _entries.remove(connection);
                if (accepted) {
                    _accepted.remove(connection);
                } else {
                    _connections.remove(connection);
                    _handles.remove(connection);
                }
            }
            // propagate the exception
            throw exception;
        }

        // mark the connection as initialised and therefore available for
        // reaping
        entry.setInitialised();

        startReaper();

        return result;
    }

    /**
     * Remove a connection from the pool.
     *
     * @param connection the connection to remove
     */
    protected void remove(ManagedConnection connection) {
        PoolEntry entry = (PoolEntry) _entries.remove(connection);
        if (entry != null) {
            if (entry.getAccepted()) {
                _accepted.remove(connection);
            } else {
                _connections.remove(connection);
                _handles.remove(connection);
            }
            URI remoteURI = null;
            URI localURI = null;
            try {
                remoteURI = connection.getRemoteURI();
                localURI = connection.getLocalURI();
            } catch (ResourceException exception) {
                _log.debug("Failed to get connection URIs", exception);
            }

            try {
                connection.destroy();
            } catch (ResourceException exception) {
                _log.debug("Failed to destroy connection", exception);
            }
            if (remoteURI != null && localURI != null) {
                notifyDisconnection(remoteURI, localURI);
            }
        } else {
            _log.debug("ManagedConnection not found");
        }
        if (_entries.isEmpty()) {
            stopReaper();
        }
    }

    /**
     * Notify of a disconnection.
     *
     * @param remoteURI the remote address that the client is calling from
     * @param localURI  the local address that the client is calling to
     */
    private void notifyDisconnection(URI remoteURI, URI localURI) {
        CallerListener listener = _listener;
        if (listener != null) {
            listener.disconnected(new CallerImpl(remoteURI, localURI));
        }
    }

    /**
     * Starts the reaper for dead/idle connections, if needed.
     */
    private synchronized void startReaper() {
        if (_daemon == null) {
            _daemon = new ClockDaemon();
            ThreadFactory creator =
                    new ThreadFactory(null, "ManagedConnectionReaper", false);
            _daemon.setThreadFactory(creator);

            if (_reapInterval > 0) {
                _daemon.executePeriodically(_reapInterval, new DeadReaper(),
                                            false);
            }
        }
    }

    /**
     * Stops the reaper for dead/idle connections, if needed.
     */
    private synchronized void stopReaper() {
        if (_daemon != null) {
            _daemon.shutDown();
            _daemon = null;
        }
    }

    /**
     * Reap idle connections.
     */
    private void reapIdleConnections() {
        Map.Entry[] entries = (Map.Entry[]) _handles.entrySet().toArray(
                new Map.Entry[0]);
        for (int i = 0; i < entries.length && !stopReaping(); ++i) {
            Map.Entry entry = entries[i];
            ManagedConnection connection =
                    (ManagedConnection) entry.getKey();
            PoolEntry pooled = (PoolEntry) _entries.get(connection);
            if (pooled != null && pooled.isInitialised()) {
                ManagedConnectionHandle handle =
                        (ManagedConnectionHandle) entry.getValue();
                if (handle.canDestroy()) {
                    if (_log.isDebugEnabled()) {
                        try {
                            _log.debug("Reaping idle connection, URI="
                                    + connection.getRemoteURI()
                                    + ", local URI="
                                    + connection.getLocalURI());
                        } catch (ResourceException ignore) {
                            // do nothing
                        }
                    }
                    remove(connection);
                }
            }
        }
    }

    /**
     * Reap dead connections.
     */
    private void reapDeadConnections() {
        Map.Entry[] entries = (Map.Entry[]) _handles.entrySet().toArray(
                new Map.Entry[0]);
        for (int i = 0; i < entries.length && !stopReaping(); ++i) {
            Map.Entry entry = entries[i];
            ManagedConnection connection =
                    (ManagedConnection) entry.getKey();
            PoolEntry pooled = (PoolEntry) _entries.get(connection);
            if (pooled != null && pooled.isInitialised()) {
                ManagedConnectionHandle handle =
                        (ManagedConnectionHandle) entry.getValue();
                if (!handle.used()) {
                    // if the handle is unused, and is not waiting on a ping
                    // reply, ping the connection
                    if (handle.pinging()) {
                        if (handle.incPingWaits() > _reapDeadIterations) {
                            remove(connection);
                        }
                    } else {
                        try {
                            handle.ping();
                        } catch (ResourceException exception) {
                            if (_log.isDebugEnabled()) {
                                try {
                                    _log.debug(
                                            "Failed to ping connection, URI="
                                                    + connection.getRemoteURI()
                                                    + ", localURI="
                                                    + connection.getLocalURI());
                                } catch (ResourceException ignore) {
                                    // do nothing
                                }
                            }
                            remove(connection);
                        }
                    }
                } else {
                    handle.clearUsed();
                }
            }
        }
    }

    /**
     * Helper to determines if a reaper should terminate, by checking the
     * interrupt status of the current thread.
     *
     * @return <code>true</code> if the reaper should terminate
     */
    private boolean stopReaping() {
        return Thread.currentThread().isInterrupted();
    }

    /**
     * Helper class for reaping idle connections.
     */
    private class IdleReaper implements Runnable {

        /**
         * Run the reaper.
         */
        public void run() {
            synchronized (_reapLock) {
                try {
                    reapIdleConnections();
                } catch (Throwable exception) {
                    _log.error(exception, exception);
                }
            }
        }
    }

    /**
     * Helper class for reaping dead connections.
     */
    private class DeadReaper implements Runnable {

        /**
         * Run the reaper.
         */
        public void run() {
            try {
                reapDeadConnections();
            } catch (Throwable exception) {
                _log.error(exception, exception);
            }
        }
    }

}