agentserver.java

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

      }

      // initializes fixed agents
      engine.init();

      // If the server is part of an HA group starts the JGroup component
      if (jgroups != null) jgroups.init(getName());

      logmon.log(BasicLevel.WARN,
                 getName() + ", initialized at " + new Date());

      // Commit all changes.
      transaction.begin();
      transaction.commit();
      transaction.release();

      try {
        SCServerMBean bean = new SCServer();
        MXWrapper.registerMBean(bean, "AgentServer", "server=" + getName());
      } catch (Exception exc) {
        if (logmon == null)
          logmon = Debug.getLogger(Debug.A3Debug + ".AgentServer");
        logmon.log(BasicLevel.ERROR, getName() + " jmx failed", exc);
      }
    } catch (Exception exc) {
      logmon.log(BasicLevel.ERROR, getName() + "Cannot initialize", exc);
      synchronized(status) {
        // AF: Will be replaced by a BAD_INITIALIZED status allowing the
        // re-initialization..
        status.value = Status.INSTALLED;
      }
      throw exc;
    } catch (Throwable t) {
      logmon.log(BasicLevel.ERROR, getName() + "Cannot initialize", t);
      synchronized(status) {
        // AF: Will be replaced by a BAD_INITIALIZED status allowing the
        // re-initialization..
        status.value = Status.INSTALLED;
      }
      throw new Exception(t);
    }

    synchronized(status) {
      status.value = Status.INITIALIZED;
    }
  }

  static String startConsumers() throws Exception {
    StringBuffer errBuf = null;

    // Now we can start all networks.
    if (consumers != null) {
      for (Enumeration c=AgentServer.getConsumers(); c.hasMoreElements(); ) {
        MessageConsumer cons = (MessageConsumer) c.nextElement();
        if (cons != null) {
          try {
            if (! (cons instanceof Engine)) {
              cons.start();
            }
          } catch (IOException exc) {
            if (errBuf == null) errBuf = new StringBuffer();
            errBuf.append(cons.getName()).append(": ");
            errBuf.append(exc.getMessage()).append('\n');
            logmon.log(BasicLevel.FATAL,
                       getName() +
                       ", problem during " + cons.getName() + " starting", exc);
          }
        }
      }
    }

    if (errBuf == null) return null;
    return errBuf.toString();
  }

  /**
   *  Causes this AgentServer to begin its execution. This method starts all
   * <code>MessageConsumer</code> (i.e. the engine and the network components).
   */
  public static String start() throws Exception {
    if (logmon.isLoggable(BasicLevel.DEBUG))
      logmon.log(BasicLevel.DEBUG, getName() + ", start()", new Exception());
    else
      logmon.log(BasicLevel.WARN, getName() + ", start()");

    synchronized(status) {
      if ((status.value != Status.INITIALIZED) &&
          (status.value != Status.STOPPED))
        throw new Exception("cannot start, bad status: " + status.value);
      status.value = Status.STARTING;
    }

    StringBuffer errBuf = null;
    try {
      try {
        if (jgroups == null) {
          ServiceManager.start();
          // Be careful, we have to save ServiceManager after start (initialized
          // attribute).
          ServiceManager.save();
          logmon.log(BasicLevel.DEBUG,
                     getName() + ", ServiceManager started");
        }
      } catch (Exception exc) {
        logmon.log(BasicLevel.FATAL,
                   getName() + ", can't initialize services", exc);
        throw new Exception("Can't initialize services");
      }

      // Now we can start all message consumers.
      if (consumers != null) {
        for (Enumeration c=AgentServer.getConsumers(); c.hasMoreElements(); ) {
          MessageConsumer cons = (MessageConsumer) c.nextElement();
          if (cons != null) {
            try {
              if ((jgroups == null) || (cons instanceof Engine)) {
                cons.start();
              }
            } catch (IOException exc) {
              if (errBuf == null) errBuf = new StringBuffer();
              errBuf.append(cons.getName()).append(": ");
              errBuf.append(exc.getMessage()).append('\n');
              logmon.log(BasicLevel.FATAL,
                         getName() +
                         ", problem during " + cons.getName() + " starting", exc);
            }
          }
        }
      }
      // The server is running.
      logmon.log(BasicLevel.WARN, getName() + ", started at " + new Date());

      // Commit all changes.
      transaction.begin();
      transaction.commit();
      transaction.release();
    } catch (Exception exc) {
      logmon.log(BasicLevel.ERROR, getName() + "Cannot start", exc);
      synchronized(status) {
        // AF: Will be replaced by a BAD_STARTED status allowing the
        // stop and reset..
        status.value = Status.STOPPED;
      }
      throw exc;
    } catch (Throwable t) {
      logmon.log(BasicLevel.ERROR, getName() + "Cannot start", t);
      synchronized(status) {
        // AF: Will be replaced by a BAD_STARTED status allowing the
        // stop and reset..
        status.value = Status.STOPPED;
      }
      throw new Exception(t);
    }

    synchronized(status) {
      status.value = Status.STARTED;
    }

    if (errBuf == null) return null;
    return errBuf.toString();
  }

  /**
   *  Forces this AgentServer to stop executing. This method stops all
   * consumers and services. Be careful, if you specify a synchronous
   * process, this method wait for all server's thread to terminate; so
   * if this method is called from a server's thread it should result a
   * dead-lock.
   *
   * @param sync	If true the stop is precessed synchronous, otherwise
   *			a thread is created and the method returns.
   */
  public static void stop(boolean sync) {
    stop(sync, 0, false);
  }

  /**
   *  Forces this AgentServer to stop executing. This method stops all
   * consumers and services. Be careful, if you specify a synchronous
   * process, this method wait for all server's thread to terminate; so
   * if this method is called from a server's thread it should result a
   * dead-lock.
   *
   * @param sync	If true the stop is precessed synchronous, otherwise
   *			a thread is created and the method returns.
   * @param delay       if sync is false then the thread in charge of
   *                    stopping the server waits this delay before
   *                    initiating the stop.
   * @param reset	If true the server is stopped then reseted.
   */
  public static void stop(boolean sync, long delay, boolean reset) {
    ServerStopper stopper = new ServerStopper(delay, reset);
    if (sync == true) {
      stopper.run();
    } else {
      // Creates a thread to execute AgentServer.stop in order to
      // avoid deadlock.
      Thread t = new Thread(stopper);
      t.setDaemon(true);
      t.start();
    }
  }

  static class ServerStopper implements Runnable {
    private long delay;
    private boolean reset;

    ServerStopper(long delay, boolean reset) {
      this.delay = delay;
      this.reset = reset;
    }

    public void run() {
      if (delay > 0) {
        try {
          Thread.sleep(delay);
        } catch (InterruptedException exc) {}
      }
      AgentServer.stop();
      if (reset) {
        AgentServer.reset();
      }
    }
  }

  /**
   *  Forces this AgentServer to stop executing. This method stops all
   * consumers and services. Be careful, the stop process is now synchronous
   * and wait for all server's thread to terminate ; If this method is called
   * from a server's thread it should result a dead-lock.
   */
  public static void stop() {
    if (logmon.isLoggable(BasicLevel.DEBUG))
      logmon.log(BasicLevel.DEBUG, getName() + ", stop()", new Exception());
    else
      logmon.log(BasicLevel.WARN, getName() + ", stop()");

    synchronized(status) {
      if ((status.value != Status.STARTED) &&
          (status.value != Status.STOPPED)) {
        logmon.log(BasicLevel.WARN,
                   getName() + "cannot stop, bad status: " + status.value);
        return;
      }
      status.value = Status.STOPPING;
    }

    try {
      // If the server is part of an HA group stops the JGroup component
      if (jgroups != null) jgroups.disconnect();

      // Stop all message consumers.
      if (consumers != null) {
        for (Enumeration c=AgentServer.getConsumers(); c.hasMoreElements(); ) {
          MessageConsumer cons = (MessageConsumer) c.nextElement();
          if (cons != null) {
            if (logmon.isLoggable(BasicLevel.DEBUG))
              logmon.log(BasicLevel.DEBUG,
                         getName() + ", stop " + cons.getName());

            cons.stop();

            if (logmon.isLoggable(BasicLevel.DEBUG))
              logmon.log(BasicLevel.DEBUG,
                         getName() + ", " + cons.getName() + " stopped");
          }
        }
      }
      // Stop all services.
      ServiceManager.stop();
      // Stop all drivers
      Driver.stopAll();

      // Wait for all threads before stop the TM !!
      while (true) {
        int nbt = getThreadGroup().activeCount();
        if (nbt == 0) break;

        Thread[] tab = new Thread[nbt];
        getThreadGroup().enumerate(tab);
        if ((nbt == 1) && (tab[0] == Thread.currentThread())) break;

        for (int j=0; j<tab.length; j++) {
          logmon.log(BasicLevel.DEBUG,
                     "[" +  tab[j].getName() + ":" +
                     (tab[j].isAlive()?"alive":"-") + "/" +
                     (tab[j].isDaemon()?"daemon":"-") + "," +
                     tab[j]);
        }
        try {
          Thread.sleep(2500);
        } catch (InterruptedException e) {}
      }

      // Stop the transaction manager.
      if (transaction != null) transaction.stop();

      // Wait for the transaction manager stop

      Runtime.getRuntime().gc();
      System.runFinalization();

      logmon.log(BasicLevel.WARN, getName() + ", stopped at " + new Date());
    } catch (Throwable t) {
      logmon.log(BasicLevel.ERROR, getName() + "Cannot stop", t);
      synchronized(status) {
        // AF: Will be replaced by a BAD_STOPPED status allowing the
        // stop and reset..
        status.value = Status.STOPPED;
      }
    }

    synchronized(status) {
      status.value = Status.STOPPED;
    }
  }

  public static final String OKSTRING = "OK";
  public static final String ERRORSTRING = "ERROR";
  public static final String ENDSTRING = "END";
  /**
   * Main for a standard agent server.
   * The start arguments include in first position the identifier of the
   * agent server to start, and in second position the directory name where
   * the agent server stores its persistent data.
   *
   * @param args	start arguments
   *
   * @exception Exception
   *	unspecialized exception
   */
  public static void main(String args[]) throws Exception {
    try {
      init(args);
    } catch (Throwable exc) {
      System.out.println(getName() + "initialisation failed: " + ERRORSTRING);
      System.out.println(exc.toString());
      System.out.println(ENDSTRING);
      if (logmon == null)
        logmon = Debug.getLogger(Debug.A3Debug + ".AgentServer");
      logmon.log(BasicLevel.ERROR,
                 getName() + " initialisation failed", exc);
      System.exit(1);
    }

    try {
      String errStr = start();
      // Be careful, the output below is needed by some tools (AdminProxy for
      // example.
      if (errStr == null) {
        System.out.println(getName() + " started: " + OKSTRING);
      } else {
        System.out.println(getName() + " started: " + ERRORSTRING);
        System.out.print(errStr);
        System.out.println(ENDSTRING);
      }
    } catch (Throwable exc) {
      System.out.println(getName() + " start failed: " + ERRORSTRING);
      System.out.print(exc.toString());
      System.out.println(ENDSTRING);
      if (logmon == null)
        logmon = Debug.getLogger(Debug.A3Debug + ".AgentServer");
      logmon.log(BasicLevel.ERROR, getName() + " failed", exc);
      System.exit(1);
    }
  }
}