hmuxstream.java

来自「RESIN 3.2 最新源码」· Java 代码 · 共 750 行 · 第 1/2 页

      _tempStream = new MemoryStream();

    _tempStream.write(buf, offset, length, isEnd);
  }

  /**
   * The stream is readable.
   */
  public boolean canRead()
  {
    return true;
  }

  /**
   * Read data from the connection.  If the request hasn't yet been sent
   * to the server, send it.
   */
  public int read(byte []buf, int offset, int length) throws IOException
  {
    try {
      return readInt(buf, offset, length);
    } catch (IOException e) {
      _isKeepalive = false;
      throw e;
    } catch (RuntimeException e) {
      _isKeepalive = false;
      throw e;
    }
  }
  
  /**
   * Read data from the connection.  If the request hasn't yet been sent
   * to the server, send it.
   */
  public int readInt(byte []buf, int offset, int length) throws IOException
  {
    if (! _didGet)
      getConnInput();

    if (_isRequestDone)
      return -1;

    try {
      int len = length;

      if (_chunkLength == 0) {
	if (! readData())
	  _chunkLength = -1;
      }
      
      if (_chunkLength < 0)
	return -1;
      
      if (_chunkLength < len)
	len = _chunkLength;

      len = _rs.read(buf, offset, len);

      if (len < 0) {
      }
      else
	_chunkLength -= len;
    
      return len;
    } catch (IOException e) {
      _isKeepalive = false;
      throw e;
    } catch (RuntimeException e) {
      _isKeepalive = false;
      throw e;
    }
  }

  /**
   * Sends the request and initializes the response.
   */
  private void getConnInput() throws IOException
  {
    if (_didGet)
      return;

    try {
      getConnInputImpl();
    } catch (IOException e) {
      _isKeepalive = false;
      throw e;
    } catch (RuntimeException e) {
      _isKeepalive = false;
      throw e;
    }
  }

  /**
   * Send the request to the server, wait for the response and parse
   * the headers.
   */
  private void getConnInputImpl() throws IOException
  {
    if (_didGet)
      return;

    _didGet = true;

    _ws.write('C');
    _ws.write(0);
    _ws.write(0);

    if (_method != null) {
      writeString(HmuxRequest.HMUX_METHOD, _method);
    }
    else if (_isPost) {
      writeString(HmuxRequest.HMUX_METHOD, "POST");
    }
    else if (_isHead)
      writeString(HmuxRequest.HMUX_METHOD, "HEAD");
    else
      writeString(HmuxRequest.HMUX_METHOD, "GET");
    
    if (_virtualHost != null)
      writeString(HmuxRequest.HMUX_SERVER_NAME, _virtualHost);
    else {
      writeString(HmuxRequest.HMUX_SERVER_NAME, _path.getHost());
      _ws.print(_path.getHost());
      if (_path.getPort() != 80) {
	writeString(HmuxRequest.CSE_SERVER_PORT,
		    String.valueOf(_path.getPort()));
      }
    }

    // Not splitting query? Also fullpath?
      writeString(HmuxRequest.HMUX_URI, _path.getPath());

    if (_path.getQuery() != null)
      writeString(HmuxRequest.CSE_QUERY_STRING, _path.getQuery());
    
    Iterator iter = getAttributeNames();
    while (iter.hasNext()) {
      String name = (String) iter.next();
      if (_reserved.get(name.toLowerCase()) == null) {
	writeString(HmuxRequest.HMUX_HEADER, name);
	writeString(HmuxRequest.HMUX_STRING, getAttribute(name));
      }
    }

    if (_isPost) {
      MemoryStream tempStream = _tempStream;
      _tempStream = null;
      if (tempStream != null) {
	TempBuffer tb = TempBuffer.allocate();
	byte []buffer = tb.getBuffer();
	int sublen;

	ReadStream postIn = tempStream.openReadAndSaveBuffer();

	while ((sublen = postIn.read(buffer, 0, buffer.length)) > 0) {
	  _ws.write('D');
	  _ws.write(sublen >> 8);
	  _ws.write(sublen);
	  _ws.write(buffer, 0, sublen);
	}

	tempStream.destroy();

	TempBuffer.free(tb);
        tb = null;
      }
    }

    _attributes.clear();

    _ws.write('Q');

    readData();

    if (_isHead)
      _isRequestDone = true;
  }

  private void writeString(int code, String string)
    throws IOException
  {
    WriteStream ws = _ws;

    ws.write((byte) code);
    int len = string.length();
    ws.write(len >> 8);
    ws.write(len);
    ws.print(string);
  }

  private void writeString(int code, Object obj)
    throws IOException
  {
    String string = String.valueOf(obj);
    
    WriteStream ws = _ws;

    ws.write((byte) code);
    int len = string.length();
    ws.write(len >> 8);
    ws.write(len);
    ws.print(string);
  }

  /**
   * Parse the headers returned from the server.
   */
  private boolean readData()
    throws IOException
  {
    boolean isDebug = log.isLoggable(Level.FINE);
    
    int code;

    ReadStream is = _rs;

    while ((code = is.read()) > 0) {
      switch (code) {
      case HmuxRequest.HMUX_CHANNEL:
	is.read();
	is.read();
	break;
      case HmuxRequest.HMUX_QUIT:
      case HmuxRequest.HMUX_EXIT:
	is.close();

	if (isDebug)
	  log.fine("HMUX: " + (char) code);
	
	return false;
	
      case HmuxRequest.HMUX_YIELD:
	break;
	
      case HmuxRequest.HMUX_STATUS:
	String value = readString(is);
	_attributes.put("status", value.substring(0, 3));
	
	if (isDebug)
	  log.fine("HMUX: " + (char) code + " " + value);
	break;
	
      case HmuxRequest.HMUX_DATA:
	_chunkLength = 256 * (is.read() & 0xff) + (is.read() & 0xff);
	
	if (isDebug)
	  log.fine("HMUX: " + (char) code + " " + _chunkLength);
	
	return true;
	
      default:
	int len = 256 * (is.read() & 0xff) + (is.read() & 0xff);
	
	if (isDebug)
	  log.fine("HMUX: " + (char) code + " " + len);
	
	is.skip(len);
	break;
      }
    }

    return false;
  }

  private String readString(ReadStream is)
    throws IOException
  {
    int len = 256 * (is.read() & 0xff) + is.read();

    char []buf = new char[len];

    is.readAll(buf, 0, len);

    return new String(buf);
  }

  /**
   * Returns the bytes still available.
   */
  public int getAvailable() throws IOException
  {
    if (! _didGet)
      getConnInput();

    return _rs.getAvailable();
  }

  /**
   * Close the connection.
   */
  public void close() throws IOException
  {
    if (_isKeepalive) {
      // If recycling, read any unread data
      if (! _didGet)
        getConnInput();

      if (! _isRequestDone) {
        if (_tempBuffer == null)
          _tempBuffer = new byte[256];

        try {
          while (read(_tempBuffer, 0, _tempBuffer.length) > 0) {
          }
        } catch (IOException e) {
          _isKeepalive = false;
        }
      }
    }

    if (com.caucho.server.util.CauchoSystem.isTesting())
      _isKeepalive = false; // XXX:
    
    if (_isKeepalive) {
      HmuxStream oldSaved;
      long now = Alarm.getCurrentTime();
      synchronized (LOCK) {
        oldSaved = _savedStream;
        _savedStream = this;
        _saveTime = now;
      }

      if (oldSaved != null && oldSaved != this) {
        oldSaved._isKeepalive = false;
        oldSaved.close();
      }

      return;
    }

    try {
      try {
        if (_ws != null)
          _ws.close();
      } catch (Throwable e) {
      }
      _ws = null;

      try {
        if (_rs != null)
          _rs.close();
      } catch (Throwable e) {
      }
      _rs = null;

      try {
        if (_os != null)
          _os.close();
      } catch (Throwable e) {
      }
      _os = null;

      try {
        if (_is != null)
          _is.close();
      } catch (Throwable e) {
      }
      _is = null;
    } finally {
      if (_s != null)
	_s.close();
      _s = null;
    }
  }

  static {
    _reserved = new HashMap<String,String>();
    _reserved.put("user-agent", "");
    _reserved.put("content-length", "");
    _reserved.put("content-encoding", "");
    _reserved.put("connection", "");
    _reserved.put("host", "");
  }
}