proxyagent.java

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

      drvIn = null;
    }
    if (drvOut != null) {
      drvOut.stop();
      drvOut = null;
    }
    if (logmon.isLoggable(BasicLevel.DEBUG))
      logmon.log(BasicLevel.DEBUG, "stopped");
  }

  /**
   * Method stopping the specified connection set (multi-connections mode). 
   *
   * @param drvKey  key identifying the connection set to stop.
   */
  protected void stop(int drvKey) {
    DriverMonitor dMonitor =
      (DriverMonitor) driversTable.get(new Integer(drvKey));

    if (dMonitor != null) {
      if (dMonitor.drvCnx != null) {
        (dMonitor.drvCnx).stop();
        dMonitor.drvCnx = null;
      }
      if (dMonitor.drvIn != null) {
        (dMonitor.drvIn).stop();
        dMonitor.drvIn = null;
      }
      if (dMonitor.drvOut != null) {
        (dMonitor.drvOut).stop();
        dMonitor.drvOut = null;
      }
    if (logmon.isLoggable(BasicLevel.DEBUG))
      logmon.log(BasicLevel.DEBUG, "stopped - driversKey=" + driversKey);
    }
  }

  /** Method cleaning DriverOut. Single connection mode only. */
  public void cleanDriverOut() {
    if (! multiConn) {
      if (drvOut != null)
        drvOut.clean();
    }
  }

  /**
   * Method cleaning the <code>DriverOut</code> specified
   * by the key parameter (multi-connections mode).
   *
   * @param  drvKey key identifying the connection set.
   */ 
  public void cleanDriverOut(int drvKey) {
    DriverMonitor dMonitor =
      (DriverMonitor) driversTable.get(new Integer(drvKey));
    if (dMonitor != null) {
      if (dMonitor.drvOut != null)
        (dMonitor.drvOut).clean();
    }
  }

  /** Closes all the connections. */
  protected void closeAllConnections() {
    Enumeration keys = driversTable.keys();
    while (keys.hasMoreElements()) {
      Integer key = (Integer) keys.nextElement();

      DriverMonitor dMonitor = (DriverMonitor) driversTable.get(key);
      if (dMonitor != null) {
        if (dMonitor.ois != null) {
          try {
           (dMonitor.ois).close();
          } catch (IOException exc) {}
          dMonitor.ois = null;
        }
        if (dMonitor.oos != null) {
          try {
           (dMonitor.oos).close();
          } catch (IOException exc) {}
          dMonitor.oos = null;
          dMonitor.qout = null;
        }
        stop(key.intValue());
      }
    }
    driversTable.clear();
  }

  /** 
   * Method called by the ProxyAgent <code>DriverIn</code> instances to
   * forward the notifications they got from their input streams. 
   * <p>
   * May be overridden for specific behaviour as long as the proxy state
   * is not modified by the method, because it does not occur within a
   * transaction.
   *
   * @param key  Driver identifier.
   * @param not  Notification to forward.
   */ 
  protected void driverReact(int key, Notification not) {
    if (logmon.isLoggable(BasicLevel.DEBUG))
      logmon.log(BasicLevel.DEBUG, "Proxy " + this + " gets not " + not 
                 + " from driver " + key);

    sendTo(this.getId(), not);
  }

  /**
   * Method called by subclasses to directly send their notifications
   * to the right <code>DriverOut</code>.
   *
   * @param key  Driver identifier.
   * @param not  Notification to send out.
   * @exception Exception  If the driver to pass the notification to can't
   *              be retrieved from the key parameter.
   */
  protected void sendOut(int key, Notification not) throws Exception {
    if (logmon.isLoggable(BasicLevel.DEBUG))
      logmon.log(BasicLevel.DEBUG, "Proxy " + this + " gets not " + not 
                 + " to pass to driver " + key);
    try {
      DriverMonitor dMon = (DriverMonitor) driversTable.get(new Integer(key));
      (dMon.getQout()).push(not);
    }
    catch (Exception e ) {
      throw new Exception("Can't forward notification " + not
                           + " to driver out " + key + ": " + e);
    }
  }
    
  /**
   * Method implementing the <code>ProxyAgent</code> reactions to
   * notifications.
   * Forwards notifications coming from an identified agent onto the outgoing
   * connection.
   *
   * @param from	agent sending notification
   * @param not		notification to react to
   *
   * @exception Exception
   *	unspecialized exception
   */
  public void react(AgentId from, Notification not) throws Exception {
    int drvKey;
    DriverMonitor dMon;
    DriverIn dIn;
    Queue qo;
    try {
      if (not instanceof DriverDone)
        driverDone((DriverDone) not);
      else if (not instanceof FlowControlNot) {
        // Allowing drvIn to read more data
        drvKey = ((FlowControlNot) not).driverKey;
        // MultiConn proxy: getting the right driverIn to control:
        if (drvKey != 0) {
          dMon = (DriverMonitor) driversTable.get(new Integer(drvKey));
          dIn = dMon.drvIn;
          dIn.recvFlowControl((FlowControlNot) not);
        }
        else
          drvIn.recvFlowControl((FlowControlNot) not);
      }
      else if (not instanceof DeleteNot) {
        closeAllConnections();
        super.react(from, not);
      } 
      // If notification comes from an identified agent:
      else if (! from.equals(this.getId()))
        qout.push(not);
      else
        super.react(from, not);
    } catch (Exception exc) {
      if (logmon.isLoggable(BasicLevel.ERROR))
        logmon.log(BasicLevel.ERROR,
                   "error in " + this + ".react(" + from + ", " + not + ")",
                   exc);
      stop();
      // the proxy agent may restart
    }
  }

  /**
   * Reacts to end of driver execution.
   * <p>
   * This is the end of the driver thread, however the thread resources
   * may not have been released. This is why <code>close</code> is called
   * on the notification streams, which requires from the stream classes
   * to cope with a call to <code>close</code> when some resources may have
   * been released.
   */
  protected void driverDone(DriverDone not) throws IOException { 
    if (!multiConn) {
      switch (not.getDriver()) {
        case DRIVER_IN:
          try {
            ois.close();
          } catch (Exception e) {}
          ois = null;
          drvIn = null;
          break;
        case DRIVER_OUT:
          try {
            oos.close();
          } catch (Exception e) {}
          oos = null;
          drvOut = null;
          break;
      }
    }
    // In case of multiConn, the driver to close is identified in the
    // DriverDone notification.
    else {
      int drvKey = not.getDriverKey();
      DriverMonitor dMonitor = (DriverMonitor)
        driversTable.get(new Integer(drvKey));

      if (dMonitor != null) {
        switch (not.getDriver()) {
          case DRIVER_IN:
	    if (dMonitor.drvIn != null)
	      (dMonitor.drvIn).close();
	    dMonitor.ois = null;
	    dMonitor.drvIn = null;
            break;
          case DRIVER_OUT:
	    if (dMonitor.drvOut != null)
	      (dMonitor.drvOut).close();
	    dMonitor.oos = null;
	    dMonitor.drvOut = null;
            break;
        }
        // When both drivers have been closed, removing the entry
        // corresponding to their pair from the driversTable.
        if (dMonitor.drvIn == null && dMonitor.drvOut == null)
          driversTable.remove(new Integer(drvKey));
      }
    } 
  }

  /**
   * Finalizes this proxy agent execution. Calls <code>disconnect</code> to
   * close the open streams, and <code>stop</code> to stop the drivers.
   *
   * @param lastTime	true when last called by the factory on agent deletion.
   */
  public void agentFinalize(boolean lastime) {
    if (logmon.isLoggable(BasicLevel.DEBUG))
      logmon.log(BasicLevel.DEBUG,
                 toString() + " agentFinalize -> " + drvCnx);

    finalizing = true;

    if (multiConn)
      closeAllConnections();
    else {
      try {
        ois.close();
      } catch (Exception exc) {}
      try {
        oos.close();
      } catch (Exception exc) {}
    }

    try {
      disconnect();
    } catch (Exception exc) {}

    stop();

    qout = null;
    ois = null;
    oos = null;
  }
}