hmuxstream.java

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

/*
 * 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.log.Log;
import com.caucho.util.Alarm;
import com.caucho.vfs.*;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.ConnectException;
import java.net.Socket;
import java.net.SocketException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * Underlying stream handling HTTP requests.
 */
class HmuxStream extends StreamImpl {
  private static final Logger log = Log.open(HmuxStream.class);
  // reserved headers that should not be passed to the HTTP server
  private static HashMap<String,String> _reserved;

  private static final Object LOCK = new Object();
  
  // Saved keepalive stream for a new request.
  private static HmuxStream _savedStream;
  // Time the stream was saved
  private static long _saveTime;
  
  private long _socketTimeout = 30000L;

  private boolean _isSSL;

  private Socket _s;
  private InputStream _is;
  private OutputStream _os;
  private ReadStream _rs;
  private WriteStream _ws;

  // The server's host name
  private String _host;
  // The server's port
  private int _port;

  private String _virtualHost;

  // the method
  private String _method;
  // true for a HEAD stream
  private boolean _isHead;
  // true for a POST stream
  private boolean _isPost;

  // buffer containing the POST data
  private MemoryStream _tempStream;

  // true if keepalive is allowed
  private boolean _isKeepalive = true;
  // true after the request has been sent
  private boolean _didGet;
  // length of the current chunk, -1 on eof
  private int _chunkLength;
  // the request is done
  private boolean _isRequestDone;

  private HashMap<String,Object> _attributes;

  // Used to read unread bytes on recycle.
  private byte []_tempBuffer;

  /**
   * Create a new HTTP stream.
   */
  private HmuxStream(Path path, String host, int port, Socket s)
    throws IOException
  {
    _s = s;

    _host = host;
    _port = port;
    
    _is = _s.getInputStream();
    _os = _s.getOutputStream();

    _ws = VfsStream.openWrite(_os);
    _rs = VfsStream.openRead(_is, _ws);

    _attributes = new HashMap<String,Object>();

    init(path);
  }

  /**
   * Opens a new HTTP stream for reading, i.e. a GET request.
   *
   * @param path the URL for the stream
   *
   * @return the opened stream
   */
  static HmuxStreamWrapper openRead(HmuxPath path) throws IOException
  {
    HmuxStream stream = createStream(path);
    stream._isPost = false;

    return new HmuxStreamWrapper(stream);
  }

  /**
   * Opens a new HTTP stream for reading and writing, i.e. a POST request.
   *
   * @param path the URL for the stream
   *
   * @return the opened stream
   */
  static HmuxStreamWrapper openReadWrite(HmuxPath path) throws IOException
  {
    HmuxStream stream = createStream(path);
    stream._isPost = true;

    return new HmuxStreamWrapper(stream);
  }

  /**
   * Creates a new HTTP stream.  If there is a saved connection to
   * the same host, use it.
   *
   * @param path the URL for the stream
   *
   * @return the opened stream
   */
  static private HmuxStream createStream(HmuxPath path) throws IOException
  {
    String host = path.getHost();
    int port = path.getPort();

    HmuxStream stream = null;
    long streamTime = 0;
    synchronized (LOCK) {
      if (_savedStream != null &&
          host.equals(_savedStream.getHost()) &&
          port == _savedStream.getPort()) {
        stream = _savedStream;
        streamTime = _saveTime;
        _savedStream = null;
      }
    }

    if (stream == null) {
    }
    // if the stream is still valid, use it
    else if (Alarm.getCurrentTime() < streamTime + 5000) {
      stream.init(path);
      return stream;
    }
    // if the stream has timed out, close it
    else {
      try {
        stream._isKeepalive = false;
        stream.close();
      } catch (IOException e) {
        log.log(Level.FINE, e.toString(), e);
      }
    }

    Socket s;

    try {
      s = new Socket(host, port);
    } catch (ConnectException e) {
      throw new ConnectException(path.getURL() + ": " + e.getMessage());
    } catch (Exception e) {
      throw new ConnectException(path.getURL() + ": " + e.toString());
    }

    int socketTimeout = 300 * 1000;

    try {
      s.setSoTimeout(socketTimeout);
    } catch (Exception e) {
    }
          
    return new HmuxStream(path, host, port, s);
  }

  /**
   * Initializes the stream for the next request.
   */
  private void init(Path path)
  {
    _isRequestDone = false;
    _didGet = false;
    _isPost = false;
    _isHead = false;
    _method = null;
    _attributes.clear();
    
    setPath(path);

    if (path instanceof HmuxPath)
      _virtualHost = ((HmuxPath) path).getVirtualHost();
  }

  /**
   * Set if this should be an SSL connection.
   */
  public void setSSL(boolean isSSL)
  {
    _isSSL = isSSL;
  }

  /**
   * Set if this should be an SSL connection.
   */
  public boolean isSSL()
  {
    return _isSSL;
  }

  /**
   * Sets the method
   */
  public void setMethod(String method)
  {
    _method = method;
  }

  /**
   * Sets true if we're only interested in the head.
   */
  public void setHead(boolean isHead)
  {
    _isHead = isHead;
  }

  /**
   * Returns the stream's host.
   */
  public String getHost()
  {
    return _host;
  }

  /**
   * Returns the stream's port.
   */
  public int getPort()
  {
    return _port;
  }
  
  /**
   * Returns a header from the response returned from the HTTP server.
   *
   * @param name name of the header
   * @return the header value.
   */
  public Object getAttribute(String name)
    throws IOException
  {
    if (! _didGet)
      getConnInput();

    return _attributes.get(name.toLowerCase());
  }

  /**
   * Returns an iterator of the returned header names.
   */
  public Iterator getAttributeNames()
    throws IOException
  {
    if (! _didGet)
      getConnInput();

    return _attributes.keySet().iterator();
  }

  /**
   * Sets a header for the request.
   */
  public void setAttribute(String name, Object value)
  {
    if (name.equals("method"))
      setMethod((String) value);
    else if (name.equals("socket-timeout")) {
      if (value instanceof Integer) {
        int socketTimeout = ((Integer) value).intValue();

        if (socketTimeout > 0) {
          try {
            if (_s != null)
              _s.setSoTimeout(socketTimeout);
          } catch (Exception e) {

          }
        }
      }
    }
    else
      _attributes.put(name.toLowerCase(), value);
  }

  /**
   * Remove a header for the request.
   */
  public void removeAttribute(String name)
  {
    _attributes.remove(name.toLowerCase());
  }

  /**
   * Sets the timeout.
   */
  public void setSocketTimeout(long timeout)
    throws SocketException
  {
    if (_s != null)
      _s.setSoTimeout((int) timeout);
  }

  /**
   * The stream is always writable (?)
   */
  public boolean canWrite()
  {
    return true;
  }

  /**
   * Writes a buffer to the underlying stream.
   *
   * @param buffer the byte array to write.
   * @param offset the offset into the byte array.
   * @param length the number of bytes to write.
   * @param isEnd true when the write is flushing a close.
   */
  public void write(byte []buf, int offset, int length, boolean isEnd)
    throws IOException
  {
    if (! _isPost)
      return;

    if (_tempStream == null)