httpnetwork.java

一个类似于openJMS分布在ObjectWeb之下的JMS消息中间件。

/*
 * Copyright (C) 2003 - 2006 ScalAgent Distributed Technologies
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or any later version.
 * 
 * This library 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.  See the GNU
 * Lesser General Public License for more details.
 * 
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307
 * USA.
 *
 * Initial developer(s): ScalAgent Distributed Technologies
 * Contributor(s): 
 */
package fr.dyade.aaa.agent;

import java.io.*;
import java.net.*;
import java.util.Vector;

import org.objectweb.util.monolog.api.BasicLevel;
import org.objectweb.util.monolog.api.Logger;

import fr.dyade.aaa.util.*;

/**
 * <tt>HttpNetwork</tt> is a simple implementation of <tt>StreamNetwork</tt>
 * based on HTTP 1.1 protocol.
 */
public class HttpNetwork extends StreamNetwork implements HttpNetworkMBean {
  private InetAddress proxy = null;
  /**
   *  Hostname (or IP dotted address) of proxy host, if not defined there
   * is a direct connection between the client and the server.
   *  This value can be adjusted for all HttpNetwork components by setting
   * <code>proxyhost</code> global property or for a particular
   * network by setting <code>\<DomainName\>.proxyhost</code>
   * specific property.
   * <p>
   *  Theses properties can be fixed either from <code>java</code> launching
   * command, or in <code>a3servers.xml</code> configuration file.
   */
  String proxyhost = null;
  /**
   *  Port number of proxy if any.
   *  This value can be adjusted for all HttpNetwork components by setting
   * <code>proxyport</code> global property or for a particular
   * network by setting <code>\<DomainName\>.proxyport</code>
   * specific property.
   * <p>
   *  Theses properties can be fixed either from <code>java</code> launching
   * command, or in <code>a3servers.xml</code> configuration file.
   */
  int proxyport = 0;

  /**
   *  Period of time between two activation of NetServerOut, it matchs to the
   * time between two requests from the client to the server when there is no
   * message to transmit from client to server.
   *  This value can be adjusted for all HttpNetwork components by setting
   * <code>ActivationPeriod</code> global property or for a particular
   * network by setting <code>\<DomainName\>.ActivationPeriod</code>
   * specific property.
   * <p>
   *  Theses properties can be fixed either from <code>java</code> launching
   * command, or in <code>a3servers.xml</code> configuration file.
   */
  protected long activationPeriod = 10000L;

  /**
   * Gets the activationPeriod value.
   *
   * @return the activationPeriod value
   */
  public long getActivationPeriod() {
    return activationPeriod;
  }

  /**
   * Sets the activationPeriod value.
   *
   * @param activationPeriod	the activationPeriod value
   */
  public void setActivationPeriod(long activationPeriod) {
    this.activationPeriod = activationPeriod;
  }

  /**
   *  Number of listening daemon, this value is only valid for the server
   * part of the HttpNetwork.
   *  This value can be adjusted for all HttpNetwork components by setting
   * <code>NbDaemon</code> global property or for a particular network by
   * setting <code>\<DomainName\>.NbDaemon</code> specific property.
   * <p>
   *  Theses properties can be fixed either from <code>java</code> launching
   * command, or in <code>a3servers.xml</code> configuration file.
   */
  int NbDaemon = 1;

  /**
   * Gets the NbDaemon value.
   *
   * @return the NbDaemon value
   */
  public long getNbDaemon() {
    return NbDaemon;
  }

  /**
   * Creates a new network component.
   */
  public HttpNetwork() {
    super();
  }

  /**
   * Descriptor of the listen server, it is used only on the client side
   * (NetServerOut component).
   */
  ServerDesc server = null;

  MessageInputStream nis = null;
  MessageOutputStream nos = null;

  /**
   * Initializes a new network component. This method is used in order to
   * easily creates and configure a Network component from a class name.
   * So we can use the <code>Class.newInstance()</code> method for create
   * (whitout any parameter) the component, then we can initialize it with
   * this method.<br>
   * This method initializes the logical clock for the domain.
   *
   * @param name	The domain name.
   * @param port	The listen port.
   * @param servers	The list of servers directly accessible from this
   *			network interface.
   */
  public void init(String name, int port, short[] servers) throws Exception {
    super.init(name, port, servers);

    nis = new MessageInputStream();
    nos = new MessageOutputStream();

    activationPeriod = Long.getLong("ActivationPeriod",
                                    activationPeriod).longValue();
    activationPeriod = Long.getLong(name + ".ActivationPeriod",
                                    activationPeriod).longValue();
    
    NbDaemon = Integer.getInteger("NbDaemon", NbDaemon).intValue();
    NbDaemon = Integer.getInteger(name + ".NbDaemon", NbDaemon).intValue();

    proxyhost = System.getProperty("proxyhost");
    proxyhost = System.getProperty(name + ".proxyhost", proxyhost);
    if (proxyhost != null) {
      proxyport = Integer.getInteger("proxyport", 8080).intValue();
      proxyport = Integer.getInteger(name + ".proxyport", proxyport).intValue();
      proxy = InetAddress.getByName(proxyhost);
    }
  }

  /** Daemon component */
  Daemon dmon[] = null;

  /**
   * Causes this network component to begin execution.
   */
  public void start() throws Exception {
    logmon.log(BasicLevel.DEBUG, getName() + ", starting");
    try {
      if (isRunning()) return;

      // AF: May be, we have to verify that there is only one 'listen' network.
      for (int i=0; i<servers.length; i++) {
        server = AgentServer.getServerDesc(servers[i]);
        if ((server.getServerId() != AgentServer.getServerId()) &&
            (server.getPort() > 0)) {
          logmon.log(BasicLevel.DEBUG, getName() + ", server=" + server);
          break;
        }
        server = null;
      }

      if (port != 0) {
        dmon = new Daemon[NbDaemon];
        ServerSocket listen = createServerSocket();

        for (int i=0; i<NbDaemon; i++) {
          dmon[i] = new NetServerIn(getName(), listen, logmon);
        }
      } else {
        dmon = new Daemon[1];
        dmon[0] = new NetServerOut(getName(), logmon);
      }

      for (int i=0; i<dmon.length; i++) {
        dmon[i].start();
      }
    } catch (IOException exc) {
      logmon.log(BasicLevel.ERROR, getName() + ", can't start", exc);
      throw exc;
    }
    logmon.log(BasicLevel.DEBUG, getName() + ", started");
  }

  /**
   * Wakes up the watch-dog thread.
   */
  public void wakeup() {
//     if (dmon != null) dmon.wakeup();
  }

  /**
   * Forces the network component to stop executing.
   */
  public void stop() {
    if (dmon != null) {
      for (int i=0; i<dmon.length; i++) {
        dmon[i].stop();
      }
    }
    logmon.log(BasicLevel.DEBUG, getName() + ", stopped");
  }

  /**
   * Tests if the network component is alive.
   *
   * @return	true if this <code>MessageConsumer</code> is alive; false
   * 		otherwise.
   */
  public boolean isRunning() {
    if (dmon != null) {
      for (int i=0; i<dmon.length; i++) {
        if (dmon[i].isRunning()) return true;
      }
    }
    return false;
  }

  /**
   * Returns a string representation of this consumer, including the
   * daemon's name and status.
   *
   * @return	A string representation of this consumer. 
   */
  public String toString() {
    StringBuffer strbuf = new StringBuffer();

    strbuf.append(super.toString()).append("\n\t");
    if (dmon != null) {
      for (int i=0; i<dmon.length; i++) {
        strbuf.append(dmon[i].toString()).append("\n\t");
      }
    }
    return strbuf.toString();
  }

  protected String readLine(InputStream is, byte[] buf) throws IOException {
    int i = 0;
    while ((buf[i++] = (byte) is.read()) != -1) {
      if ((buf[i-1] == '\n') && (buf[i-2] == '\r')) {
	i -= 2;
	break;
      }
    }

    if (i > 0) return new String(buf, 0, i);

    return null;
  }
  
  protected void sendRequest(Message msg,
                             OutputStream os, int ack,
                             long currentTimeMillis) throws Exception {
    StringBuffer strbuf = new StringBuffer();

    strbuf.append("PUT ");
    if (proxy != null) {
      strbuf.append("http://").append(server.getHostname()).append(':').append(server.getPort());
    }
    strbuf.append("/msg?from=").append(AgentServer.getServerId());
    strbuf.append("&stamp=");
    if (msg != null) {
      strbuf.append(msg.getStamp());
    } else {
      strbuf.append("-1");
    }
    strbuf.append(" HTTP/1.1");
    nos.writeMessage(msg, ack, currentTimeMillis);

    if (proxy != null)
      strbuf.append("\r\nHost: ").append(server.getHostname());
    strbuf.append("\r\n" +
                  "User-Agent: ScalAgent 1.0\r\n" +
                  "Accept: image/jpeg;q=0.2\r\n" +
                  "Accept-Language: fr, en-us;q=0.50\r\n" +
                  "Accept-Encoding: gzip;q=0.9\r\n" +
                  "Accept-Charset: ISO-8859-1, utf-8;q=0.66\r\n" +
                  "Cache-Control: no-cache\r\n" +
                  "Cache-Control: no-store\r\n" +
                  "Keep-Alive: 300\r\n" +
                  "Connection: keep-alive\r\n" +
                  "Proxy-Connection: keep-alive\r\n" +
                  "Pragma: no-cache\r\n");
    strbuf.append("Content-Length: ").append(nos.size());
    strbuf.append("\r\n" +
                  "Content-Type: image/jpeg\r\n");
    strbuf.append("\r\n");

    os.write(strbuf.toString().getBytes());
    
    if (logmon.isLoggable(BasicLevel.DEBUG))
      logmon.log(BasicLevel.DEBUG, name + ", writes:" + nos.size());
    nos.writeTo(os);
    nos.reset();

    os.flush();
  }

  protected short getRequest(InputStream is, byte[] buf) throws Exception {
    String line = null;

    line = readLine(is, buf);
    if ((line == null) ||
        (! (line.startsWith("GET ") || line.startsWith("PUT ")))) {
      throw new Exception("Bad request: " + line);
    }

    int idx1 = line.indexOf("?from=");
    if (idx1 == -1) throw new Exception("Bad request: " + line);
    int idx2 = line.indexOf("&", idx1);
    if (idx2 == -1) throw new Exception("Bad request: " + line);
    short from = Short.parseShort(line.substring(idx1+6, idx2));

    // Skip all header lines, get length
    int length = 0;
    while (line != null) {
      line = readLine(is, buf);
      if ((line != null) && line.startsWith("Content-Length: ")) {
        // get content length
	length = Integer.parseInt(line.substring(16));
        if (logmon.isLoggable(BasicLevel.DEBUG))
          logmon.log(BasicLevel.DEBUG, name + ", length:" + length);
      }
    }

    if (nis.readFrom(is) != length)
      logmon.log(BasicLevel.WARN, name + "Bad request length: " + length);

    return from;
  }

  protected void sendReply(Message msg,
                           OutputStream os, int ack,
                           long currentTimeMillis) throws Exception {
    StringBuffer strbuf = new StringBuffer();

    strbuf.append("HTTP/1.1 200 OK\r\n");

    nos.writeMessage(msg, ack, currentTimeMillis);

    strbuf.append("Date: ").append("Fri, 21 Feb 2003 14:30:51 GMT");
    strbuf.append("\r\n" +
                  "Server: ScalAgent 1.0\r\n" +
                  "Last-Modified: ").append("Wed, 19 Apr 2000 08:16:28 GMT");
    strbuf.append("\r\n" +
                  "Cache-Control: no-cache\r\n" +
                  "Cache-Control: no-store\r\n" +
                  "Accept-Ranges: bytes\r\n" +
                  "Keep-Alive: timeout=15, max=100\r\n" +
                  "Connection: Keep-Alive\r\n" +
                  "Proxy-Connection: Keep-Alive\r\n" +
                  "Pragma: no-cache\r\n");
    strbuf.append("Content-Length: ").append(nos.size());
    strbuf.append("\r\n" +
                  "Content-Type: image/gif\r\n");
    strbuf.append("\r\n");

    os.write(strbuf.toString().getBytes());
    
    if (logmon.isLoggable(BasicLevel.DEBUG))
      logmon.log(BasicLevel.DEBUG, name + ", writes:" + nos.size());
    nos.writeTo(os);
    nos.reset();

    os.flush();
  }

  protected void getReply(InputStream is, byte[] buf) throws Exception {
    String line = null;

    line = readLine(is, buf);
    if ((line == null) ||
        ((! line.equals("HTTP/1.1 200 OK")) &&
         (! line.equals("HTTP/1.1 204 No Content")))) {
      throw new Exception("Bad reply: " + line);
    }

    // Skip all header lines, get length
    int length = 0;
    while (line != null) {
      line = readLine(is, buf);
      if ((line != null) && line.startsWith("Content-Length: ")) {
        // get content length
	length = Integer.parseInt(line.substring(16));
        if (logmon.isLoggable(BasicLevel.DEBUG))
          logmon.log(BasicLevel.DEBUG, name + ", length:" + length);
      }
    }

    if (nis.readFrom(is) != length)
      logmon.log(BasicLevel.WARN, name + "Bad reply length: " + length);
  }

  protected int handle(Message msgout) throws Exception {
    int ack = nis.getAckStamp();
    if ((msgout != null) && (msgout.stamp == ack)) {
      AgentServer.getTransaction().begin();
      //  Suppress the processed notification from message queue,
      // and deletes it.
      qout.removeMessage(msgout);
      msgout.delete();
      msgout.free();
      AgentServer.getTransaction().commit();
      AgentServer.getTransaction().release();
    }

    Message msg = nis.getMessage();
    if (logmon.isLoggable(BasicLevel.DEBUG))
      logmon.log(BasicLevel.DEBUG,
                 this.getName() + ", get: " + msg);

    if (msg != null) {
      testBootTS(msg.getSource(), nis.getBootTS());
      deliver(msg);
      return msg.stamp;
    }

    return -1;
  }

  /**
   * This method creates a tunnelling socket if a proxy is used.
   *