proxyagent.java

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

/*
 * Copyright (C) 2001 - 2005 ScalAgent Distributed Technologies
 * Copyright (C) 1996 - 2000 BULL
 * Copyright (C) 1996 - 2000 INRIA
 *
 * 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.
 *
 * The present code contributor is ScalAgent Distributed Technologies.
 */
package fr.dyade.aaa.agent;

import java.io.*;
import java.util.Hashtable;
import java.util.Enumeration;

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

import fr.dyade.aaa.util.*;

public abstract class ProxyAgent extends Agent {

  public static final int DRIVER_IN = 1;
  public static final int DRIVER_OUT = 2;

  /** true if connect may block */
  protected boolean blockingCnx = true;
  /** true if proxy may handle multiple connection */
  protected boolean multipleCnx = false;
  /** flow control in driver in */
  protected int inFlowControl = 20;
  /** communication with drvOut */
  protected transient Queue qout;

  /** manage input stream */
  private transient DriverIn drvIn = null;
  /** manage output stream */
  private transient DriverOut drvOut = null;
  /** manage connection step, optional */
  private transient DriverConnect drvCnx;

  /** <code>true</code> if the proxy manages multiple connections. */
  protected boolean multiConn = false;
  /** 
   * Table holding the <code>DriverMonitor</code> objects, each one holding a
   * connection set (a pair of drivers, a qout, ois, oos, ...).
   * For multi-connections management.
   *
   * @see  DriverMonitor
   */
  protected transient Hashtable driversTable;
  /**
   * Used in multi-connections context for identifying each
   * connection.
   */
  private int driversKey = 1;

  /** <code>true</code> if the proxy is being finalized. */
  boolean finalizing = false;

  /**
   * Returns default log topic for proxies. Its method  overriddes
   * the default one in Agent, the resulting logging topic is
   * <code>Debug.A3Proxy</code> dot the real classname.
   */
  protected String getLogTopic() {
    String classname = getClass().getName();
    return Debug.A3Proxy + '.' +
      classname.substring(classname.lastIndexOf('.') +1);
  }

  public ProxyAgent() {
    this(null);
  }

  public ProxyAgent(String n) {
    this(AgentServer.getServerId(), n);
  }

  public ProxyAgent(short to, String n) {
    // ProxyAgent is "pined" in memory.
    super(to, n, true);
    drvIn = null;
    drvOut = null;
    drvCnx = null;
  }

  /*
   * Constructor used to build Well Known Services agents.
   *
   * @param name	symbolic name
   * @param stamp	well known stamp
   */
  public ProxyAgent(String name, int stamp) {
    super(name, true, stamp);
    drvIn = null;
    drvOut = null;
    drvCnx = null;
  }

  /**
   * Provides a string image for this object.
   *
   * @return	printable image of this object
   */
  public String toString() {
    return "(" + super.toString() +
      ",blockingCnx=" + blockingCnx +
      ",multipleCnx=" + multipleCnx +
      ",inFlowControl=" + inFlowControl +
      ",multiConn=" + multiConn +
      ",qout=" + qout +
      ",driversKey=" + driversKey + ")";
  }
  
  /**
   * Method setting the <code>ProxyAgent</code> in multiConn mode.
   * To be called immediately after the <code>ProxyAgent</code> instanciation.
   */
  public void setMultiConn() {
    multiConn = true;
  }

  /**
   * Initializes the transient members of this agent.
   * This function is first called by the factory agent, then by the system
   * each time the agent server is restarted.
   * <p>
   * This function is not declared final so that derived classes
   * may change their reload policy.
   *
   * @param firstTime		true when first called by the factory
   *
   * @exception Exception
   *	unspecialized exception
   */
  protected void agentInitialize(boolean firstTime) throws Exception {
    super.agentInitialize(firstTime);

    // In single connection mode, creating qout once:
    if (!multiConn)
      qout = new Queue();
    // In multi connections mode, creating the driversTable:
    else
      driversTable = new Hashtable();

    reinitialize();
  }

  /**
   * Reinitializes the agent, that is reconnects its input and output.
   * This function may be called only when all drivers are null
   * if the <code>ProxyAgent</code> manages only one connection at a time.
   * Otherwise, a multiConn <code>ProxyAgent</code> will reinitialize
   * even if the current drivers are not null.
   */
  protected void reinitialize() throws IOException {
    if (drvIn != null || drvOut != null) {
      if (!multiConn) throw new IllegalStateException();
    }

    drvCnx = new DriverConnect(this, blockingCnx, multipleCnx);
    drvCnx.start();

    // If the ProxyAgent manages multi-connections, storing the connection set
    // in a DriverMonitor and putting it in the driversTable.
    if (multiConn) {
      DriverMonitor dMonitor = new DriverMonitor(drvIn, drvOut, qout, ois, oos,
                                                 drvCnx);

      driversTable.put(new Integer(driversKey), dMonitor); 
      driversKey++;
    }
  }

  /** input stream, created by subclass during connect */
  protected transient NotificationInputStream ois = null;
  /** output stream, created by subclass during connect */
  protected transient NotificationOutputStream oos = null;

  /**
   * Initializes the connection with the outside, up to creating
   * the input and output streams <code>ois</code> and <code>oos</code>.
   *
   * @exception Exception
   *	unspecialized exception
   */
  public abstract void connect() throws Exception;

  /**
   * Closes the connection with the outside.
   *
   * @exception Exception
   *	unspecialized exception
   */
  public abstract void disconnect() throws Exception;

  /**
   * Connects the streams provided by the user to this proxy agent
   * to two created drivers. The streams must be created by the
   * <code>connect</code> function defined in derived classes.
   * <p>
   * If the connection step blocks, this function is executed
   * in a separate thread controled by <code>drvCnx</code> (see
   * <code>Initialize</code>).
   * 
   * @exception java.net.SocketException  If the server socket is being closed.
   */
  void createDrivers() throws Exception {
    drvCnx.canStop = true;
    try {
      connect();
    } catch (InterruptedException exc) {
      logmon.log(BasicLevel.DEBUG, getName() + "InterruptedException");
    } finally {
      if (drvCnx != null)
        drvCnx.canStop = false;
    }

    if ((drvCnx == null) || (! drvCnx.isRunning)) return;

    if (! multiConn) {  
      if (logmon.isLoggable(BasicLevel.DEBUG))
        logmon.log(BasicLevel.DEBUG, getName() + ", connected");
      if (oos != null) {
        drvOut = new DriverOut(DRIVER_OUT, this, qout, oos);
        drvOut.start();
      }
      if (ois != null) {
        drvIn = new DriverIn(DRIVER_IN, this, ois, inFlowControl);
        drvIn.start();
      }
    }
    // If the ProxyAgent is multiConn, creating drvIn and drvOut with
    // the additionnal driversKey parameter and also creating a new qout.
    else {
      if (logmon.isLoggable(BasicLevel.DEBUG))
        logmon.log(BasicLevel.DEBUG, "connected - driversKey=" + driversKey);
      if (ois != null) {
        drvIn = new DriverIn(DRIVER_IN, this, ois, inFlowControl, driversKey);
        drvIn.start();
      }
      if (oos != null) {
        qout = new Queue(); 
        drvOut = new DriverOut(DRIVER_OUT, this, qout, oos, driversKey);
        drvOut.start();
      }
    }
  }

  /**
   * Stops all drivers (non multiConn mode).
   * May be multiply called.
   */
  protected void stop() {
    if (multiConn) return;

    if (drvCnx != null) {
      drvCnx.stop();
      drvCnx = null;
    }
    if (drvIn != null) {
      drvIn.stop();