selectchannelconnector.java

著名服务器nio的源码

    /**
     * Set the period in ms that a connection is allowed to be idle when this there are more
     * than {@link #getLowResourcesConnections()} connections.  This allows the server to rapidly close idle connections
     * in order to gracefully handle high load situations.
     * @param lowResourcesMaxIdleTime the period in ms that a connection is allowed to be idle when resources are low.
     * @see {@link #setMaxIdleTime(long)}
     */
    public void setLowResourceMaxIdleTime(int lowResourcesMaxIdleTime)
    {
        _lowResourcesMaxIdleTime=lowResourcesMaxIdleTime;
        super.setLowResourceMaxIdleTime(lowResourcesMaxIdleTime); 
    }
    
    /* ------------------------------------------------------------ */
    /*
     * @see org.mortbay.jetty.AbstractConnector#doStart()
     */
    protected void doStart() throws Exception
    {
        _manager.setSelectSets(getAcceptors());
        _manager.setMaxIdleTime(getMaxIdleTime());
        _manager.setLowResourcesConnections(getLowResourcesConnections());
        _manager.setLowResourcesMaxIdleTime(getLowResourcesMaxIdleTime());
        _manager.start();
        open();
        _manager.register(_acceptChannel);
        super.doStart();
    }

    /* ------------------------------------------------------------ */
    /*
     * @see org.mortbay.jetty.AbstractConnector#doStop()
     */
    protected void doStop() throws Exception
    {
        _manager.stop();
        super.doStop();
    }

    /* ------------------------------------------------------------ */
    protected SelectChannelEndPoint newEndPoint(SocketChannel channel, SelectSet selectSet, SelectionKey key) throws IOException
    {
        return new ConnectorEndPoint(channel,selectSet,key);
    }

    /* ------------------------------------------------------------------------------- */
    protected Connection newConnection(SocketChannel channel,SelectChannelEndPoint endpoint)
    {
        return new HttpConnection(SelectChannelConnector.this,endpoint,getServer());
    }

    /* ------------------------------------------------------------ */
    /* ------------------------------------------------------------ */
    /* ------------------------------------------------------------ */
    public static class ConnectorEndPoint extends SelectChannelEndPoint
    {
        public ConnectorEndPoint(SocketChannel channel, SelectSet selectSet, SelectionKey key)
        {
            super(channel,selectSet,key);
            scheduleIdle();
        }

        public void close() throws IOException
        {
            RetryContinuation continuation = (RetryContinuation) ((HttpConnection)getConnection()).getRequest().getContinuation();
            if (continuation != null && continuation.isPending())
                continuation.reset();

            super.close();
        }

        /* ------------------------------------------------------------ */
        public void undispatch()
        {
            RetryContinuation continuation = (RetryContinuation) ((HttpConnection)getConnection()).getRequest().getContinuation();

            if (continuation != null)
            {
                // We have a continuation
                Log.debug("continuation {}", continuation);
                if (continuation.undispatch())
                    super.undispatch();
            }
            else
            {
                super.undispatch();
            }
               
        }
    }

    /* ------------------------------------------------------------ */
    /* ------------------------------------------------------------ */
    /* ------------------------------------------------------------ */
    public static class RetryContinuation extends Timeout.Task implements Continuation, Runnable
    {
        SelectChannelEndPoint _endPoint=(SelectChannelEndPoint)HttpConnection.getCurrentConnection().getEndPoint();
        boolean _new = true;
        Object _object;
        boolean _pending = false;   // waiting for resume or timeout
        boolean _resumed = false;   // resume called.
        boolean _parked =false;     // end point dispatched, but undispatch called.
        RetryRequest _retry;
        long _timeout;

        
        public Object getObject()
        {
            return _object;
        }

        public long getTimeout()
        {
            return _timeout;
        }

        public boolean isNew()
        {
            return _new;
        }

        public boolean isPending()
        {
            return _pending;
        }

        public boolean isResumed()
        {
            return _resumed;
        }

        public void reset()
        {
            synchronized (this)
            {
                _resumed = false;
                _pending = false;
                _parked = false;
            }
            
            synchronized (_endPoint.getSelectSet())
            {
                this.cancel();   
            }
        } 

        public boolean suspend(long timeout)
        {
            boolean resumed=false;
            synchronized (this)
            {
                resumed=_resumed;
                _resumed=false;
                _new = false;
                if (!_pending && !resumed && timeout >= 0)
                {
                    _pending=true;
                    _parked = false;
                    _timeout = timeout;
                    if (_retry==null)
                     _retry = new RetryRequest();
                    throw _retry;
                }
                
                // here only if suspend called on pending continuation.
                // acts like a reset
                _resumed = false;
                _pending = false;
                _parked =false;
            }

            synchronized (_endPoint.getSelectSet())
            {
                this.cancel();   
            }

            return resumed;
        }
        
        public void resume()
        {
            boolean redispatch=false;
            synchronized (this)
            {
                if (_pending && !isExpired())
                {
                    _resumed = true;
                    redispatch=_parked;
                    _parked=false;
                }
            }

            if (redispatch)
            {
                SelectSet selectSet = _endPoint.getSelectSet();
                
                synchronized (selectSet)
                {
                    this.cancel();   
                }

                _endPoint.scheduleIdle();  // TODO maybe not needed?
                selectSet.addChange(this);
                selectSet.wakeup();
            }
        }
        
        public void expire()
        {
            boolean redispatch=false;
            synchronized (this)
            {
                redispatch=_parked && _pending && !_resumed;
                _parked=false;
            }
            if (redispatch)
            {
                _endPoint.scheduleIdle();  // TODO maybe not needed?
                _endPoint.getSelectSet().addChange(this);
                _endPoint.getSelectSet().wakeup();
            }
        }

        
        public void run()
        {
            _endPoint.run();
        }
        
        /* undispatch continuation.
         * Called when an endppoint is undispatched.  
         * Either sets timeout or dispatches if already resumed or expired */
        public boolean undispatch()
        {
            boolean redispatch=false;
        
            synchronized (this)
            {
                if (!_pending)
                    return true;
                
                redispatch=isExpired() || _resumed;
                _parked=!redispatch;
            }
            
            if (redispatch)
            {
                _endPoint.scheduleIdle();
                _endPoint.getSelectSet().addChange(this);
            }
            else if (_timeout>0)
                _endPoint.getSelectSet().scheduleTimeout(this,_timeout);
            
            _endPoint.getSelectSet().wakeup();
            return false;
        }

        public void setObject(Object object)
        {
            _object = object;
        }
        
        public String toString()
        {
            synchronized (this)
            {
                return "RetryContinuation@"+hashCode()+
                (_new?",new":"")+
                (_pending?",pending":"")+
                (_resumed?",resumed":"")+
                (isExpired()?",expired":"")+
                (_parked?",parked":"");
            }
        }

    }

}