hmuxrequest.java

来自「RESIN 3.2 最新源码」· Java 代码 · 共 1,842 行 · 第 1/3 页

/*
 * Copyright (c) 1998-2008 Caucho Technology -- all rights reserved
 *
 * This file is part of Resin(R) Open Source
 *
 * Each copy or derived work must preserve the copyright notice and this
 * notice unmodified.
 *
 * Resin Open Source 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.
 *
 * Resin Open Source 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, or any warranty
 * of NON-INFRINGEMENT.  See the GNU General Public License for more
 * details.
 *
 * You should have received a copy of the GNU General Public License
 * along with Resin Open Source; if not, write to the
 *
 *   Free Software Foundation, Inc.
 *   59 Temple Place, Suite 330
 *   Boston, MA 02111-1307  USA
 *
 * @author Scott Ferguson
 */

package com.caucho.server.hmux;

import com.caucho.hessian.io.*;
import com.caucho.bam.*;
import com.caucho.server.cluster.Cluster;
import com.caucho.server.cluster.Server;
import com.caucho.server.connection.AbstractHttpRequest;
import com.caucho.server.connection.Connection;
import com.caucho.server.dispatch.DispatchServer;
import com.caucho.server.dispatch.Invocation;
import com.caucho.server.dispatch.InvocationDecoder;
import com.caucho.server.http.InvocationKey;
import com.caucho.server.port.ServerRequest;
import com.caucho.server.webapp.ErrorPageManager;
import com.caucho.util.*;
import com.caucho.vfs.ClientDisconnectException;
import com.caucho.vfs.ReadStream;
import com.caucho.vfs.StreamImpl;
import com.caucho.vfs.WriteStream;

import java.io.*;
import java.net.InetAddress;
import java.security.cert.CertificateFactory;
import java.security.cert.X509Certificate;
import java.util.*;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * Handles requests from a remote dispatcher.  For example, mod_caucho
 * and the IIS plugin use this protocol to talk to the backend.
 *
 * <p>Packets are straightforward:
 * <pre>code l2 l1 l0 contents</pre>
 * Where code is the code of the packet and l2-0 give 12 bits of length.
 *
 * <p>The protocol is designed to allow pipelining and buffering whenever
 * possible.  So most commands are not acked.  Data from the frontend (POST)
 * does need acks to prevent deadlocks at either end while waiting
 * for new data.
 *
 * <p>The overriding protocol is controlled by requests from the
 * frontend server.
 * 
 * <p>A ping request:
 * <pre>
 * Frontend       Backend
 * CSE_PING
 * CSE_END
 *                CSE_END
 * <pre>
 * 
 * <p>A GET request:
 * <pre>
 * Frontend       Backend
 * CSE_METHOD
 * ...
 * CSE_HEADER/CSE_VALUE
 * CSE_END
 *                CSE_DATA
 *                CSE_DATA
 *                CSE_END
 * <pre>
 * 
 * <p>Short POST:
 * <pre>
 * Frontend       Backend
 * CSE_METHOD
 * ...
 * CSE_HEADER/CSE_VALUE
 * CSE_DATA
 * CSE_END
 *                CSE_DATA
 *                CSE_DATA
 *                CSE_END
 * <pre>
 * 
 * <p>Long POST:
 * <pre>
 * Frontend       Backend
 * CSE_METHOD
 * ...
 * CSE_HEADER/CSE_VALUE
 * CSE_DATA
 *                CSE_DATA (optional)
 * CSE_DATA
 *                CSE_ACK
 *                CSE_DATA (optional)
 * CSE_DATA
 *                CSE_ACK
 * CSE_END
 *                CSE_DATA
 *                CSE_END
 * <pre>
 * 
 */
public class HmuxRequest extends AbstractHttpRequest
  implements ServerRequest
{
  private static final L10N L = new L10N(HmuxRequest.class);
  private static final Logger log
    = Logger.getLogger(HmuxRequest.class.getName());

  // HMUX channel control codes
  public static final int HMUX_CHANNEL =        'C';
  public static final int HMUX_ACK =            'A';
  public static final int HMUX_ERROR =          'E';
  public static final int HMUX_YIELD =          'Y';
  public static final int HMUX_QUIT =           'Q';
  public static final int HMUX_EXIT =           'X';

  public static final int HMUX_DATA =           'D';
  public static final int HMUX_URI =            'U';
  public static final int HMUX_STRING =         'S';
  public static final int HMUX_HEADER =         'H';
  public static final int HMUX_BINARY =         'B';
  public static final int HMUX_PROTOCOL =       'P';
  public static final int HMUX_META_HEADER =    'M';
  
  // The following are HTTP codes
  public static final int CSE_NULL =            '?';
  public static final int CSE_PATH_INFO =       'b';
  public static final int CSE_PROTOCOL =        'c';
  public static final int CSE_REMOTE_USER =     'd';
  public static final int CSE_QUERY_STRING =    'e';
  public static final int HMUX_FLUSH  =         'f';
  public static final int CSE_SERVER_PORT =     'g';
  public static final int CSE_REMOTE_HOST =     'h';
  public static final int CSE_REMOTE_ADDR =     'i';
  public static final int CSE_REMOTE_PORT =     'j';
  public static final int CSE_REAL_PATH =       'k';
  public static final int CSE_SCRIPT_FILENAME = 'l';
  public static final int HMUX_METHOD =         'm';
  public static final int CSE_AUTH_TYPE =       'n';
  public static final int CSE_URI =             'o';
  public static final int CSE_CONTENT_LENGTH =  'p';
  public static final int CSE_CONTENT_TYPE =    'q';
  public static final int CSE_IS_SECURE =       'r';
  public static final int HMUX_STATUS =         's';
  public static final int CSE_CLIENT_CERT =     't';
  public static final int CSE_SERVER_TYPE =     'u';
  public static final int HMUX_SERVER_NAME =    'v';
  
  public static final int CSE_SEND_HEADER =     'G';

  public static final int CSE_DATA =            'D';
  public static final int CSE_FLUSH =           'F';
  public static final int CSE_KEEPALIVE =       'K';
  public static final int CSE_ACK =             'A';
  public static final int CSE_END =             'Z';
  public static final int CSE_CLOSE =           'X';

  // other, specialized protocols
  public static final int CSE_QUERY =           'Q';
  public static final int CSE_PING =            'P';

  public static final int HMTP_MESSAGE =        '0';
  public static final int HMTP_QUERY_GET =      '1';
  public static final int HMTP_QUERY_SET =      '2';
  public static final int HMTP_QUERY_RESULT =   '3';
  public static final int HMTP_ERROR =          '4';
  public static final int HMTP_PACKET =         '5';

  public static final int HMUX_CLUSTER_PROTOCOL = 0x101;
  public static final int HMUX_DISPATCH_PROTOCOL = 0x102;
  public static final int HMUX_JMS_PROTOCOL = 0x103;

  public enum ProtocolResult {
    QUIT,
    EXIT,
    YIELD
  };
  
  static final int HTTP_0_9 = 0x0009;
  static final int HTTP_1_0 = 0x0100;
  static final int HTTP_1_1 = 0x0101;

  private static final int HEADER_CAPACITY = 256;
  
  static final CharBuffer _getCb = new CharBuffer("GET");
  static final CharBuffer _headCb = new CharBuffer("HEAD");
  static final CharBuffer _postCb = new CharBuffer("POST");

  private final CharBuffer _method;   // "GET"
  private String _methodString;       // "GET"
  // private CharBuffer scheme;       // "http:"
  private CharBuffer _host;            // www.caucho.com
  private int _port;                   // :80

  private ByteBuffer _uri;             // "/path/test.jsp/Junk"
  private CharBuffer _protocol;        // "HTTP/1.0"
  private int _version;

  private CharBuffer _remoteAddr;
  private CharBuffer _remoteHost;
  private CharBuffer _serverName;
  private CharBuffer _serverPort;
  private CharBuffer _remotePort;
  
  private boolean _isSecure;
  private ByteBuffer _clientCert;

  private CharBuffer []_headerKeys;
  private CharBuffer []_headerValues;
  private int _headerSize;

  private byte []_lengthBuf;

  private int _serverType;

  // write stream from the connection
  private WriteStream _rawWrite;
  // servlet write stream
  private WriteStream _writeStream;
  
  // StreamImpl to break reads and writes to the underlying protocol
  private ServletFilter _filter;
  private int _pendingData;

  private InvocationKey _invocationKey = new InvocationKey();

  private CharBuffer _cb1;
  private CharBuffer _cb2;
  private boolean _hasRequest;

  private Hessian2StreamingInput _in;
  private Hessian2StreamingOutput _out;

  private Server _server;
  private AbstractClusterRequest _clusterRequest;
  private HmuxDispatchRequest _dispatchRequest;
  private Cluster _cluster;

  private HmuxProtocol _hmuxProtocol;
  private ErrorPageManager _errorManager = new ErrorPageManager(null);

  private int _srunIndex;

  public HmuxRequest(DispatchServer server,
		     Connection conn,
		     HmuxProtocol protocol)
  {
    super(server, conn);

    _server = (Server) server;
    
    _hmuxProtocol = protocol;

    _response = new HmuxResponse(this);

    _rawWrite = conn.getWriteStream();
    _writeStream = new WriteStream();
    _writeStream.setReuseBuffer(true);

    // XXX: response.setIgnoreClientDisconnect(server.getIgnoreClientDisconnect());

    _cluster = Cluster.getLocal();
    if (_cluster != null) {
      try {
	Class cl = Class.forName("com.caucho.server.hmux.HmuxClusterRequest");

	_clusterRequest = (AbstractClusterRequest) cl.newInstance();
	_clusterRequest.setRequest(this);
	_clusterRequest.setCluster(_cluster);
      } catch (ClassNotFoundException e) {
	log.finer(e.toString());
      } catch (Throwable e) {
	log.log(Level.FINER, e.toString(), e);
      }
    }
    
    _dispatchRequest = new HmuxDispatchRequest(this);
    
    _uri = new ByteBuffer();

    _method = new CharBuffer();
    _host = new CharBuffer();
    _protocol = new CharBuffer();

    _headerKeys = new CharBuffer[HEADER_CAPACITY];
    _headerValues = new CharBuffer[_headerKeys.length];
    for (int i = 0; i < _headerKeys.length; i++) {
      _headerKeys[i] = new CharBuffer();
      _headerValues[i] = new CharBuffer();
    }

    _remoteHost = new CharBuffer();
    _remoteAddr = new CharBuffer();
    _serverName = new CharBuffer();
    _serverPort = new CharBuffer();
    _remotePort = new CharBuffer();

    _clientCert = new ByteBuffer();

    _cb1 = new CharBuffer();
    _cb2 = new CharBuffer();

    _lengthBuf = new byte[16];

    _filter = new ServletFilter();
  }

  public boolean isWaitForRead()
  {
    return true;
  }

  /**
   * Called when the connection starts.
   */
  public void startConnection()
  {
    _in = null;
    _out = null;
  }

  /**
   * Handles a new request.  Initializes the protocol handler and
   * the request streams.
   *
   * <p>Note: ClientDisconnectException must be rethrown to
   * the caller.
   */
  public boolean handleRequest()
    throws IOException
  {
    // XXX: should be moved to TcpConnection
    Thread thread = Thread.currentThread();
    thread.setContextClassLoader(_server.getClassLoader());
    
    if (log.isLoggable(Level.FINE))
      log.fine(dbgId() + "start request");

    _filter.init(this, _rawRead, _rawWrite);
    _writeStream.init(_filter);
    // _writeStream.setWritePrefix(3);

    _response.init(_writeStream);

    _serverType = 0;
    _uri.setLength(0);

    boolean hasRequest = false;
    
    try {
      start();
      _response.start();

      try {
        if (! scanHeaders()) {
          killKeepalive();
          return false;
        }
        else if (_uri.size() == 0) {
          return true;
	}
      } catch (InterruptedIOException e) {
        killKeepalive();
        log.fine(dbgId() + "interrupted keepalive");
        return false;
      }

      if (_isSecure)
        getClientCertificate();

      hasRequest = true;
      // setStartDate();

      if (_server == null || _server.isDestroyed()) {
	log.fine(dbgId() + "server is closed");
	
	try {
	  _writeStream.setDisableClose(false);
	  _writeStream.close();
	} catch (Throwable e) {
	}
	
	try {
	  _readStream.setDisableClose(false);
	  _readStream.close();
	} catch (Throwable e) {
	}

	return false;
      }

      _filter.setPending(_pendingData);

      try {
        if (_method.getLength() == 0)
          throw new RuntimeException("HTTP protocol exception");
          
        _invocationKey.init(_isSecure,
			    getHost(), getServerPort(),
                            _uri.getBuffer(), _uri.getLength());

        Invocation invocation;

        invocation = _server.getInvocation(_invocationKey);

        if (invocation == null) {
          invocation = _server.createInvocation();

          if (_host != null)
            invocation.setHost(_host.toString());
          
          invocation.setPort(getServerPort());

          InvocationDecoder decoder = _server.getInvocationDecoder();

	  decoder.splitQueryAndUnescape(invocation,
					_uri.getBuffer(),
					_uri.getLength());

          invocation = _server.buildInvocation(_invocationKey.clone(),
					       invocation);
        }

	invocation = invocation.getRequestInvocation(this);

	setInvocation(invocation);
      
        invocation.service(this, _response);
      } catch (ClientDisconnectException e) {
        throw e;
      } catch (Throwable e) {
	log.log(Level.FINER, e.toString(), e);
	
        try {
          _errorManager.sendServletError(e, this, _response);
        } catch (ClientDisconnectException e1) {
          throw e1;
        } catch (Exception e1) {
          log.log(Level.FINE, e1.toString(), e1);
        }

	return false;
      }
    } finally {
      if (! hasRequest)
	_response.setHeaderWritten(true);
      
      try {
	finish();
	_response.finish();
      } catch (ClientDisconnectException e) {
        throw e;
      } catch (Exception e) {
	killKeepalive();
        log.log(Level.FINE, dbgId() + e, e);
      }

      try {
	_writeStream.setDisableClose(false);
	_writeStream.close();
      } catch (ClientDisconnectException e) {
	killKeepalive();
        log.log(Level.FINE, dbgId() + e, e);

        throw e;
      } catch (Exception e) {
	killKeepalive();
        log.log(Level.FINE, dbgId() + e, e);
      }

      try {
	_readStream.setDisableClose(false);
	_readStream.close();
      } catch (Exception e) {
	killKeepalive();
        log.log(Level.FINE, dbgId() + e, e);
      }
    }

    boolean allowKeepalive = isKeepalive();
    
    if (log.isLoggable(Level.FINE)) {
      if (allowKeepalive)
	log.fine(dbgId() + "complete request - keepalive");
      else
	log.fine(dbgId() + "complete request");
    }

    return allowKeepalive;
  }

  /**
   * Initialize the read stream from the raw stream.
   */
  protected boolean initStream(ReadStream readStream,
                               ReadStream rawStream)
    throws IOException
  {
    readStream.init(_filter, null);

    return true;
  }

  private void getClientCertificate()
  {
    String cipher = getHeader("SSL_CIPHER");
    if (cipher == null)
      cipher = getHeader("HTTPS_CIPHER");
    if (cipher != null)
      setAttribute("javax.servlet.request.cipher_suite", cipher);
    
    String keySize = getHeader("SSL_CIPHER_USEKEYSIZE");
    if (keySize == null)
      keySize = getHeader("SSL_SECRETKEYSIZE");
    if (keySize != null)
      setAttribute("javax.servlet.request.key_size", keySize);
    
    if (_clientCert.size() == 0)
      return;
    
    try {
      CertificateFactory cf = CertificateFactory.getInstance("X.509");
      InputStream is = _clientCert.createInputStream();
      Object cert = cf.generateCertificate(is);
      is.close();
      setAttribute("javax.servlet.request.X509Certificate", cert);
      setAttribute(com.caucho.server.security.AbstractAuthenticator.LOGIN_NAME,
                   ((X509Certificate) cert).getSubjectDN());
    } catch (Throwable e) {
      log.log(Level.FINE, e.toString(), e);
    }
  }

  /**
   * Returns true for the top-level request, but false for any include()
   * or forward()
   */
  @Override
  public boolean isTop()
  {
    return true;
  }
  
  protected boolean checkLogin()
  {
    return true;
  }

  /**
   * Clears variables at the start of a new request.
   */
  @Override
  protected void start()
    throws IOException
  {
    super.start();

    _method.clear();
    _methodString = null;
    _protocol.clear();
    _version = 0;
    _uri.clear();
    _host.clear();
    _port = 0;

    _headerSize = 0;

    _remoteHost.clear();
    _remoteAddr.clear();
    _serverName.clear();
    _serverPort.clear();
    _remotePort.clear();

    _clientCert.clear();

    _pendingData = 0;

    _isSecure = _conn.isSecure();
  }
  
  /**
   * Fills request parameters from the stream.
   */
  private boolean scanHeaders()
    throws IOException
  {
    boolean hasURI = false;
    CharBuffer cb = _cb;
    boolean isLoggable = log.isLoggable(Level.FINE);
    ReadStream is = _rawRead;