Java 类org.apache.http.nio.NHttpClientConnection 实例源码

public void connected(final NHttpClientConnection conn, final Object attachment) {
    System.out.println(conn + " [proxy->origin] conn open");

    // The shared state object is expected to be passed as an attachment
    ProxyTask proxyTask = (ProxyTask) attachment;

    synchronized (proxyTask) {
        ConnState connState = proxyTask.getOriginState();
        if (connState != ConnState.IDLE) {
            throw new IllegalStateException("Illegal target connection state: " + connState);
        }

        // Set origin IO control handle
        proxyTask.setOriginIOControl(conn);
        // Store the state object in the context
        HttpContext context = conn.getContext();
        context.setAttribute(ProxyTask.ATTRIB, proxyTask);
        // Update connection state
        proxyTask.setOriginState(ConnState.CONNECTED);

        if (proxyTask.getRequest() != null) {
            conn.requestOutput();
        }
    }
}

public AsyncConnectionManager(
        HttpHost target,
        int maxConnections,
        NHttpClientHandler handler, 
        HttpParams params) throws IOReactorException {
    super();
    this.target = target;
    this.maxConnections = maxConnections;
    this.handler = handler;
    this.params = params;
    this.lock = new Object();
    this.allConns = new HashSet<NHttpClientConnection>();
    this.availableConns = new LinkedList<NHttpClientConnection>();
    this.pendingRequests = new LinkedList<AsyncConnectionRequest>();
    this.ioreactor = new DefaultConnectingIOReactor(2, params);
}

public AsyncConnectionRequest requestConnection() {
    if (this.shutdown) {
        throw new IllegalStateException("Connection manager has been shut down");
    }
    AsyncConnectionRequest request = new AsyncConnectionRequest();
    synchronized (this.lock) {
        while (!this.availableConns.isEmpty()) {
            NHttpClientConnection conn = this.availableConns.remove();
            if (conn.isOpen()) {
                System.out.println("Re-using persistent connection");
                request.setConnection(conn);
                break;
            } else {
                this.allConns.remove(conn);
            }
        }
        if (!request.isCompleted()) {
            this.pendingRequests.add(request);
            processRequests();                
        }
    }
    return request;
}

public void releaseConnection(NHttpClientConnection conn) {
    if (conn == null) {
        return;
    }
    if (this.shutdown) {
        return;
    }
    synchronized (this.lock) {
        if (this.allConns.contains(conn)) {
            if (conn.isOpen()) {
                conn.setSocketTimeout(0);
                AsyncConnectionRequest request = this.pendingRequests.poll();
                if (request != null) {
                    System.out.println("Re-using persistent connection");
                    request.setConnection(conn);
                } else {
                    this.availableConns.add(conn);
                }
            } else {
                this.allConns.remove(conn);
                processRequests();
            }
        }
    }
}

@Override
public void requestReady(NHttpClientConnection conn) throws IOException, HttpException {
    try {
        super.requestReady(conn);
    } catch (Exception ex) {
        LOGGER.error("", ex);
    }

    // 需要自动关闭连接
    if (this.liveTime > 0) {
        HttpRequest httpRequest = conn.getHttpRequest();
        if (httpRequest == null) {
            return;
        }

        HttpContext context = conn.getContext();

        long currentTimeMillis = System.currentTimeMillis();
        Object oldTimeMillisObj = context.getAttribute("t");
        if (oldTimeMillisObj == null) {
            context.setAttribute("t", currentTimeMillis);
        } else {
            long oldTimeMillis = (Long) oldTimeMillisObj;
            long dt = currentTimeMillis - oldTimeMillis;
            if (dt > 1000 * liveTime) { // 超时，重连
                tryCloseConnection(httpRequest);
                context.setAttribute("t", currentTimeMillis);
            }
        }
    }
}

public void closed(final NHttpClientConnection conn) {
    System.out.println(conn + " [proxy->origin] conn closed");
    HttpContext context = conn.getContext();
    ProxyTask proxyTask = (ProxyTask) context.getAttribute(ProxyTask.ATTRIB);

    if (proxyTask != null) {
        synchronized (proxyTask) {
            proxyTask.setOriginState(ConnState.CLOSED);
        }
    }
}

public void setConnection(NHttpClientConnection conn) {
    if (this.completed) {
        return;
    }
    this.completed = true;
    synchronized (this) {
        this.conn = conn;
        notifyAll();
    }
}

void addConnection(NHttpClientConnection conn) {
    if (conn == null) {
        return;
    }
    if (this.shutdown) {
        return;
    }
    synchronized (this.lock) {
        this.allConns.add(conn);
    }
}

void removeConnection(NHttpClientConnection conn) {
    if (conn == null) {
        return;
    }
    if (this.shutdown) {
        return;
    }
    synchronized (this.lock) {
        if (this.allConns.remove(conn)) {
            this.availableConns.remove(conn);
        }
        processRequests();                
    }
}

public void outputReady(final NHttpClientConnection conn, final ContentEncoder encoder) {
    System.out.println(conn + " [proxy->origin] output ready");

    HttpContext context = conn.getContext();
    ProxyTask proxyTask = (ProxyTask) context.getAttribute(ProxyTask.ATTRIB);

    synchronized (proxyTask) {
        ConnState connState = proxyTask.getOriginState();
        if (connState != ConnState.REQUEST_SENT
                && connState != ConnState.REQUEST_BODY_STREAM) {
            throw new IllegalStateException("Illegal target connection state: " + connState);
        }

        try {

            ByteBuffer src = proxyTask.getInBuffer();
            src.flip();
            int bytesWritten = encoder.write(src);
            System.out.println(conn + " [proxy->origin] " + bytesWritten + " bytes written");
            System.out.println(conn + " [proxy->origin] " + encoder);
            src.compact();

            if (src.position() == 0) {
                if (proxyTask.getClientState() == ConnState.REQUEST_BODY_DONE) {
                    encoder.complete();
                } else {
                    // Input buffer is empty. Wait until the client fills up 
                    // the buffer
                    conn.suspendOutput();
                }
            }
            // Update connection state
            if (encoder.isCompleted()) {
                System.out.println(conn + " [proxy->origin] request body sent");
                proxyTask.setOriginState(ConnState.REQUEST_BODY_DONE);
            } else {
                proxyTask.setOriginState(ConnState.REQUEST_BODY_STREAM);
                // Make sure client input is active
                proxyTask.getClientIOControl().requestInput();
            }

        } catch (IOException ex) {
            shutdownConnection(conn);
        }
    }
}

public void responseReceived(final NHttpClientConnection conn) {
    System.out.println(conn + " [proxy<-origin] response received");

    HttpContext context = conn.getContext();
    ProxyTask proxyTask = (ProxyTask) context.getAttribute(ProxyTask.ATTRIB);

    synchronized (proxyTask) {
        ConnState connState = proxyTask.getOriginState();
        if (connState != ConnState.REQUEST_SENT
                && connState != ConnState.REQUEST_BODY_DONE) {
            throw new IllegalStateException("Illegal target connection state: " + connState);
        }

        HttpResponse response = conn.getHttpResponse();
        HttpRequest request = proxyTask.getRequest();

        System.out.println(conn + " [proxy<-origin] << " + response.getStatusLine());

        int statusCode = response.getStatusLine().getStatusCode();
        if (statusCode < HttpStatus.SC_OK) {
            // Ignore 1xx response
            return;
        }
        try {

            // Update connection state
            proxyTask.setResponse(response);
            proxyTask.setOriginState(ConnState.RESPONSE_RECEIVED);

            if (!canResponseHaveBody(request, response)) {
                conn.resetInput();
                if (!this.connStrategy.keepAlive(response, context)) {
                    System.out.println(conn + " [proxy<-origin] close connection");
                    proxyTask.setOriginState(ConnState.CLOSING);
                    conn.close();
                }
            }
            // Make sure client output is active
            proxyTask.getClientIOControl().requestOutput();

        } catch (IOException ex) {
            shutdownConnection(conn);
        }
    }

}

public void inputReady(final NHttpClientConnection conn, final ContentDecoder decoder) {
    System.out.println(conn + " [proxy<-origin] input ready");

    HttpContext context = conn.getContext();
    ProxyTask proxyTask = (ProxyTask) context.getAttribute(ProxyTask.ATTRIB);

    synchronized (proxyTask) {
        ConnState connState = proxyTask.getOriginState();
        if (connState != ConnState.RESPONSE_RECEIVED
                && connState != ConnState.RESPONSE_BODY_STREAM) {
            throw new IllegalStateException("Illegal target connection state: " + connState);
        }

        HttpResponse response = proxyTask.getResponse();
        try {

            ByteBuffer dst = proxyTask.getOutBuffer();
            int bytesRead = decoder.read(dst);
            System.out.println(conn + " [proxy<-origin] " + bytesRead + " bytes read");
            System.out.println(conn + " [proxy<-origin] " + decoder);
            if (!dst.hasRemaining()) {
                // Output buffer is full. Suspend origin input until 
                // the client handler frees up some space in the buffer
                conn.suspendInput();
            }
            // If there is some content in the buffer make sure client output 
            // is active
            if (dst.position() > 0) {
                proxyTask.getClientIOControl().requestOutput();
            }

            if (decoder.isCompleted()) {
                System.out.println(conn + " [proxy<-origin] response body received");
                proxyTask.setOriginState(ConnState.RESPONSE_BODY_DONE);

                if (!this.connStrategy.keepAlive(response, context)) {
                    System.out.println(conn + " [proxy<-origin] close connection");
                    proxyTask.setOriginState(ConnState.CLOSING);
                    conn.close();
                }
            } else {
                proxyTask.setOriginState(ConnState.RESPONSE_BODY_STREAM);
            }

        } catch (IOException ex) {
            shutdownConnection(conn);
        }
    }
}

public void exception(final NHttpClientConnection conn, final HttpException ex) {
    shutdownConnection(conn);
    System.out.println(conn + " [proxy->origin] HTTP error: " + ex.getMessage());
}

public void exception(final NHttpClientConnection conn, final IOException ex) {
    shutdownConnection(conn);
    System.out.println(conn + " [proxy->origin] I/O error: " + ex.getMessage());
}

public void timeout(final NHttpClientConnection conn) {
    System.out.println(conn + " [proxy->origin] timeout");
    closeConnection(conn);
}

public NHttpClientConnection getConnection() {
    return this.conn;
}

public void connected(NHttpClientConnection conn, Object attachment) {
    AsyncConnectionRequest request = (AsyncConnectionRequest) attachment;
    this.handler.connected(conn, attachment);
    this.connMgr.addConnection(conn);
    request.setConnection(conn);
}

public void closed(NHttpClientConnection conn) {
    this.connMgr.removeConnection(conn);
    this.handler.closed(conn);
}

public void requestReady(NHttpClientConnection conn) {
    this.handler.requestReady(conn);
}

public void outputReady(NHttpClientConnection conn, ContentEncoder encoder) {
    this.handler.outputReady(conn, encoder);
}

public void responseReceived(NHttpClientConnection conn) {
    this.handler.responseReceived(conn);
}

public void inputReady(NHttpClientConnection conn, ContentDecoder decoder) {
    this.handler.inputReady(conn, decoder);
}

public void exception(NHttpClientConnection conn, HttpException ex) {
    this.handler.exception(conn, ex);
}

public void exception(NHttpClientConnection conn, IOException ex) {
    this.handler.exception(conn, ex);
}

public void timeout(NHttpClientConnection conn) {
    this.handler.timeout(conn);
}

public RequestHandle(AsyncConnectionManager connMgr, NHttpClientConnection conn) {
    super();
    this.connMgr = connMgr;
    this.conn = conn;
}

public void connected(final NHttpClientConnection conn,
        final Object attachment) throws IOException, HttpException {
    System.out.println("[proxy->origin] connection open " + conn);
    ConsoleFactory.printToConsole("[proxy->origin] connection open " + conn,true);
    super.connected(conn, attachment);
}

public void closed(final NHttpClientConnection conn) {
    System.out.println("[proxy->origin] connection closed " + conn);
    ConsoleFactory.printToConsole("[proxy->origin] connection closed " + conn,true);
    super.closed(conn);
}

public ProxyConnPool(
        final ConnectingIOReactor ioreactor,
        final NIOConnFactory<HttpHost, NHttpClientConnection> connFactory,
        final int connectTimeout) {
    super(ioreactor, connFactory, connectTimeout);
}