hmuxrequest.java

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

  }

  public final byte []getUriBuffer()
  {
    return _uri.getBuffer();
  }

  public final int getUriLength()
  {
    return _uri.getLength();
  }

  /**
   * Returns the protocol.
   */
  public String getProtocol()
  {
    return _protocol.toString();
  }

  public CharSegment getProtocolBuffer()
  {
    return _protocol;
  }

  final int getVersion()
  {
    return _version;
  }

  /**
   * Returns true if the request is secure.
   */
  @Override
  public boolean isSecure()
  {
    return _isSecure;
  }

  /**
   * Returns the header.
   */
  public String getHeader(String key)
  {
    CharSegment buf = getHeaderBuffer(key);
    if (buf != null)
      return buf.toString();
    else
      return null;
  }

  @Override
  public CharSegment getHeaderBuffer(String key)
  {
    for (int i = 0; i < _headerSize; i++) {
      CharBuffer test = _headerKeys[i];

      if (test.equalsIgnoreCase(key))
	return _headerValues[i];
    }

    return null;
  }

  public CharSegment getHeaderBuffer(char []buf, int length)
  {
    for (int i = 0; i < _headerSize; i++) {
      CharBuffer test = _headerKeys[i];

      if (test.length() != length)
        continue;
      
      char []keyBuf = test.getBuffer();
      int j;
      for (j = 0; j < length; j++) {
        char a = buf[j];
        char b = keyBuf[j];
        if (a == b)
          continue;

        if (a >= 'A' && a <= 'Z')
          a += 'a' - 'A';
        if (b >= 'A' && b <= 'Z')
          b += 'a' - 'A';
        if (a != b)
          break;
      }

      if (j == length)
        return _headerValues[i];
    }

    return null;
  }

  @Override
  public void setHeader(String key, String value)
  {
    if (_headerKeys.length <= _headerSize)
      resizeHeaders();
    
    _headerKeys[_headerSize].clear();
    _headerKeys[_headerSize].append(key);
    _headerValues[_headerSize].clear();
    _headerValues[_headerSize].append(value);
    _headerSize++;
  }

  @Override
  public void getHeaderBuffers(String key, ArrayList<CharSegment> values)
  {
    CharBuffer cb = _cb;
    
    cb.clear();
    cb.append(key);

    int size = _headerSize;
    for (int i = 0; i < size; i++) {
      CharBuffer test = _headerKeys[i];
      if (test.equalsIgnoreCase(cb))
	values.add(_headerValues[i]);
    }
  }

  public Enumeration getHeaderNames()
  {
    HashSet<String> names = new HashSet<String>();
    for (int i = 0; i < _headerSize; i++)
      names.add(_headerKeys[i].toString());

    return Collections.enumeration(names);
  }

  /**
   * Returns the URI for the request, special casing the IIS issues.
   * Because IIS already escapes the URI before sending it, the URI
   * needs to be re-escaped.
   */
  @Override
  public String getRequestURI() 
  {
    if (_serverType == 'R')
      return super.getRequestURI();

    String _rawURI = super.getRequestURI();
    CharBuffer cb = CharBuffer.allocate();

    for (int i = 0; i < _rawURI.length(); i++) {
      char ch = _rawURI.charAt(i);

      if (ch <= ' ' || ch >= 0x80 || ch == '%') {
	addHex(cb, ch);
      }
      else
        cb.append(ch);
    }

    return cb.close();
  }

  /**
   * Adds a hex escape.
   *
   * @param cb the char buffer containing the escape.
   * @param ch the character to be escaped.
   */
  private void addHex(CharBuffer cb, int ch)
  {
    cb.append('%');

    int d = (ch >> 4) & 0xf;
    if (d < 10)
      cb.append((char) ('0' + d));
    else
      cb.append((char) ('a' + d - 10));
    
    d = ch & 0xf;
    if (d < 10)
      cb.append((char) ('0' + d));
    else
      cb.append((char) ('a' + d - 10));
  }

  /**
   * Returns the server name.
   */
  @Override
  public String getServerName()
  {
    CharBuffer host = getHost();
    if (host == null) {
      InetAddress addr = getConnection().getRemoteAddress();
      return addr.getHostName();
    }

    int p = host.indexOf(':');
    if (p >= 0)
      return host.substring(0, p);
    else
      return host.toString();
  }

  @Override
  public int getServerPort()
  {
    int len = _serverPort.length();
    int port = 0;
    for (int i = 0; i < len; i++) {
      char ch = _serverPort.charAt(i);
      port = 10 * port + ch - '0';
    }

    return port;
  }

  @Override
  public String getRemoteAddr()
  {
    return _remoteAddr.toString();
  }

  public void getRemoteAddr(CharBuffer cb)
  {
    cb.append(_remoteAddr);
  }

  @Override
  public int printRemoteAddr(byte []buffer, int offset)
    throws IOException
  {
    char []buf = _remoteAddr.getBuffer();
    int len = _remoteAddr.getLength();

    for (int i = 0; i < len; i++)
      buffer[offset + i] = (byte) buf[i];

    return offset + len;
  }

  @Override
  public String getRemoteHost()
  {
    return _remoteHost.toString();
  }

  /**
   * Called for a connection: close
   */
  @Override
  protected void connectionClose()
  {
    // ignore for hmux
  }

  // Response data
  void writeStatus(CharBuffer message)
    throws IOException
  {
    int channel = 2;

    WriteStream os = _rawWrite;
    
    os.write(HMUX_CHANNEL);
    os.write(channel >> 8);
    os.write(channel);
              
    writeString(HMUX_STATUS, message);
  }

  /**
   * Complete sending of all headers.
   */
  void sendHeader()
    throws IOException
  {
    writeString(CSE_SEND_HEADER, "");
  }

  /**
   * Writes a header to the plugin.
   *
   * @param key the header's key
   * @param value the header's value
   */
  void writeHeader(String key, String value)
    throws IOException
  {
    writeString(HMUX_HEADER, key); 
    writeString(HMUX_STRING, value); 
  }

  /**
   * Writes a header to the plugin.
   *
   * @param key the header's key
   * @param value the header's value
   */
  void writeHeader(String key, CharBuffer value)
    throws IOException
  {
    writeString(HMUX_HEADER, key); 
    writeString(HMUX_STRING, value); 
  }

  private void writeObject(WriteStream out, Serializable value)
    throws IOException
  {
    if (_out == null)
      _out = new Hessian2StreamingOutput(_rawWrite);

    _out.writeObject(value);
    _out.flush();
  }

  private void writeLong(WriteStream out, long v)
    throws IOException
  {
    out.write((int) (v << 56));
    out.write((int) (v << 48));
    out.write((int) (v << 40));
    out.write((int) (v << 32));
    out.write((int) (v << 24));
    out.write((int) (v << 16));
    out.write((int) (v << 8));
    out.write((int) (v << 0));
  }

  void writeString(int code, String value)
    throws IOException
  {
    int len = value.length();

    WriteStream os = _rawWrite;
    
    os.write(code);
    os.write(len >> 8);
    os.write(len);
    os.print(value);
    
    if (log.isLoggable(Level.FINE))
      log.fine(dbgId() + (char)code + " " + value);
  }

  void writeString(int code, CharBuffer cb)
    throws IOException
  {
    int len = cb.length();

    WriteStream os = _rawWrite;
    
    os.write(code);
    os.write(len >> 8);
    os.write(len);
    os.print(cb.getBuffer(), 0, len);

    if (log.isLoggable(Level.FINE))
      log.fine(dbgId() + (char)code + " " + cb);
  }
  
  public void protocolCloseEvent()
  {
  }

  @Override
  public final String dbgId()
  {
    String id = _server.getServerId();

    if (id.equals(""))
      return "Hmux[" + getConnection().getId() + "] ";
    else
      return "Hmux[" + id + ":" + getConnection().getId() + "] ";
  }

  @Override
  public String toString()
  {
    return "HmuxRequest" + dbgId();
  }

  /**
   * Implements the protocol for data reads and writes.  Data from the
   * web server to the JVM must be acked, except for the first data.
   * Data back to the web server needs no ack.
   */
  static class ServletFilter extends StreamImpl {
    HmuxRequest _request;
    ReadStream _is;
    WriteStream _os;
    byte []_buffer = new byte[16];
    int _pendingData;
    boolean _needsAck;
    boolean _isClosed;
    boolean _isClientClosed;

    ServletFilter()
    {
    }

    void init(HmuxRequest request,
	      ReadStream nextRead, WriteStream nextWrite)
    {
      _request = request;
      _is = nextRead;
      _os = nextWrite;
      _pendingData = 0;
      _isClosed = false;
      _isClientClosed = false;
      _needsAck = false;
    }

    void setPending(int pendingData)
    {
      _pendingData = pendingData;
    }

    void setClientClosed(boolean isClientClosed)
    {
      _isClientClosed = isClientClosed;
    }

    @Override
    public boolean canRead()
    {
      return true;
    }

    @Override
    public int getAvailable()
    {
      return _pendingData;
    }

    /**
     * Reads available data.  If the data needs an ack, then do so.
     */
    @Override
    public int read(byte []buf, int offset, int length)
      throws IOException
    {
      int sublen = _pendingData;
      ReadStream is = _is;
      
      if (sublen <= 0)
	return -1;

      if (length < sublen)
	sublen = length;

      int readLen = is.read(buf, offset, sublen);
      _pendingData -= readLen;

      if (log.isLoggable(Level.FINEST))
        log.finest(new String(buf, offset, readLen));

      while (_pendingData == 0) {
        if (_needsAck) {
          int channel = 2;
          
          _os.write(HMUX_ACK);
          _os.write(channel >> 8);
          _os.write(channel);
          
          if (log.isLoggable(Level.FINE))
            log.fine(_request.dbgId() + "A:ack channel 2");
        }
        
        _needsAck = false;

	int code = is.read();

        if (code == HMUX_DATA) {
          int len = (is.read() << 8) + is.read();
          
          if (log.isLoggable(Level.FINE))
            log.fine(_request.dbgId() + "D:post-data " + len);

          _pendingData = len;
        }
        else if (code == HMUX_QUIT) {
          if (log.isLoggable(Level.FINE))
            log.fine(_request.dbgId() + "Q:quit");
	  
          return readLen;
        }
        else if (code == HMUX_EXIT) {
          if (log.isLoggable(Level.FINE))
            log.fine(_request.dbgId() + "X:exit");
	  
	  _request.killKeepalive();
          return readLen;
        }
        else if (code == HMUX_YIELD) {
          _needsAck = true;
        }
	else if (code == HMUX_CHANNEL) {
	  int channel = (is.read() << 8) + is.read();
	  
          if (log.isLoggable(Level.FINE))
            log.fine(_request.dbgId() + "channel " + channel);
	}
        else if (code < 0) {
	  _request.killKeepalive();
	
          return readLen;
	}
        else {
	  _request.killKeepalive();
	  
          int len = (is.read() << 8) + is.read();
          
          if (log.isLoggable(Level.FINE))
            log.fine(_request.dbgId() + "unknown `" + (char) code + "' " + len);

          is.skip(len);
        }
      }

      return readLen;
    }

    @Override
    public boolean canWrite()
    {
      return true;
    }

    /**
     * Send data back to the web server
     */
    @Override
    public void write(byte []buf, int offset, int length, boolean isEnd)
      throws IOException
    {
      if (log.isLoggable(Level.FINE)) {
	log.fine(_request.dbgId() + (char) HMUX_DATA + ":data " + length);
        
        if (log.isLoggable(Level.FINEST))
          log.finest(_request.dbgId() + "data <" + new String(buf, offset, length) + ">");
      }

      byte []tempBuf = _buffer;
      
      while (length > 0) {
	int sublen = length;

	if (32 * 1024 < sublen)
	  sublen = 32 * 1024;
	
        // The 3 bytes are already allocated by setPrefixWrite
	tempBuf[0] = HMUX_DATA;
	tempBuf[1] = (byte) (sublen >> 8);
	tempBuf[2] = (byte) sublen;

	_os.write(tempBuf, 0, 3);
	_os.write(buf, offset, sublen);

	length -= sublen;
	offset += sublen;
      }
    }

    @Override
    public void flush()
      throws IOException
    {
      if (log.isLoggable(Level.FINE))
	log.fine(_request.dbgId() + (char) HMUX_FLUSH + ":flush");

      _os.write(HMUX_FLUSH);
      _os.write(0);
      _os.write(0);
      _os.flush();
    }

    @Override
    public void close()
      throws IOException
    {
      if (_isClosed)
	return;

      _isClosed = true;

      if (_pendingData > 0) {
	_is.skip(_pendingData);
	_pendingData = 0;
      }

      boolean keepalive = _request.allowKeepalive();

      if (! _isClientClosed) {
	if (log.isLoggable(Level.FINE)) {
          if (keepalive)
            log.fine(_request.dbgId() + (char) HMUX_QUIT + ": quit channel");
          else
            log.fine(_request.dbgId() + (char) HMUX_EXIT + ": exit socket");
        }

        if (keepalive)
          _os.write(HMUX_QUIT);
        else
          _os.write(HMUX_EXIT);
      }

      if (keepalive)
        _os.flush();
      else
        _os.close();
      //nextRead.close();
    }
  }
}