adminmodule.java

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

    throws ConnectException, AdminException {

    Monitor_GetDestinations request = new Monitor_GetDestinations(serverId);
    Monitor_GetDestinationsRep reply =
      (Monitor_GetDestinationsRep) doRequest(request,delay);

    Vector list = new Vector();
    String[] ids = reply.getIds();
    String[] names = reply.getNames();
    String[] types = reply.getTypes();
    for (int i = 0; i < types.length; i++) {
      list.addElement(Destination.newInstance(
                        ids[i], names[i], types[i]));
    }
    return list;
  }

  /**
   * Returns the list of all users that exist on a given server, or an empty
   * list if none exist.
   * <p>
   * The request fails if the target server does not belong to the platform.
   *
   * @exception ConnectException  If the connection fails.
   * @exception AdminException  If the request fails.
   */
  public static List getUsers(int serverId)
    throws ConnectException, AdminException
    {
      Monitor_GetUsers request = new Monitor_GetUsers(serverId);
      Monitor_GetUsersRep reply = (Monitor_GetUsersRep) doRequest(request);

      Vector list = new Vector();
      Hashtable users = reply.getUsers();
      String name;
      for (Enumeration names = users.keys(); names.hasMoreElements();) {
        name = (String) names.nextElement();
        list.add(new User(name, (String) users.get(name)));
      }
      return list;
    }

  /**
   * Returns the list of all users that exist on a given server, or an empty
   * list if none exist.
   * The request is abort after delay.
   *
   * @exception ConnectException  If the connection fails.
   * @exception AdminException  If the request fails.
   */
  public static List getUsers(int serverId, long delay)
    throws ConnectException, AdminException {

    Monitor_GetUsers request = new Monitor_GetUsers(serverId);
    Monitor_GetUsersRep reply = (Monitor_GetUsersRep) doRequest(request,delay);

    Vector list = new Vector();
    Hashtable users = reply.getUsers();
    String name;
    for (Enumeration names = users.keys(); names.hasMoreElements();) {
      name = (String) names.nextElement();
      list.add(new User(name, (String) users.get(name)));
    }
    return list;
  }

  /**
   * Returns the list of all users that exist on the local server, or an empty
   * list if none exist.
   *
   * @exception ConnectException  If the connection fails.
   * @exception AdminException  Never thrown.
   */
  public static List getUsers() throws ConnectException, AdminException
    {
      return getUsers(localServer);
    }


  /**
   * Returns the identifier of the server the module is connected to.
   *
   * @exception ConnectException  If the admin connection is not established.
   */
  public static int getLocalServerId() throws ConnectException
    {
      if (cnx == null)
        throw new ConnectException("Administrator not connected.");

      return localServer;
    }

  /**
   * Returns the host name of the server the module is connected to.
   *
   * @exception ConnectException  If the admin connection is not established.
   */
  public static String getLocalHost() throws ConnectException
    {
      if (cnx == null)
        throw new ConnectException("Administrator not connected.");

      return localHost;
    }

  /**
   * Returns the port number of the server the module is connected to.
   *
   * @exception ConnectException  If the admin connection is not established.
   */
  public static int getLocalPort() throws ConnectException
    {
      if (cnx == null)
        throw new ConnectException("Administrator not connected.");

      return localPort;
    }

  /**
   * Method actually sending an <code>AdminRequest</code> instance to
   * the platform and getting an <code>AdminReply</code> instance.
   *
   * @exception ConnectException  If the connection to the platform fails.
   * @exception AdminException  If the platform's reply is invalid, or if
   *              the request failed.
   */
  public static AdminReply doRequest(AdminRequest request)
    throws AdminException, ConnectException {
    return doRequest(request,requestTimeout);
  }

  /**
   * Method actually sending an <code>AdminRequest</code> instance to
   * the platform and getting an <code>AdminReply</code> instance.
   *
   * @exception ConnectException  If the connection to the platform fails.
   * @exception AdminException  If the platform's reply is invalid, or if
   *              the request failed.
   */
  public static AdminReply doRequest(AdminRequest request, long timeout)
    throws AdminException, ConnectException {
    if (JoramTracing.dbgClient.isLoggable(BasicLevel.DEBUG))
      JoramTracing.dbgClient.log(
        BasicLevel.DEBUG, "AdminModule.doRequest(" + request + ')');

    if (cnx == null)
      throw new ConnectException("Admin connection not established.");

    if (timeout < 1)
      timeout = requestTimeout;

    try {
      replyMsg = (ObjectMessage) requestor.request(
        request, timeout);
      reply = (AdminReply) replyMsg.getObject();

      if (! reply.succeeded()) {
        switch (reply.getErrorCode()) {
        case AdminReply.NAME_ALREADY_USED:
          throw new NameAlreadyUsedException(reply.getInfo());
        case AdminReply.START_FAILURE:
          throw new StartFailureException(reply.getInfo());
        case AdminReply.SERVER_ID_ALREADY_USED:
          throw new ServerIdAlreadyUsedException(reply.getInfo());
        case AdminReply.UNKNOWN_SERVER:
          throw new UnknownServerException(reply.getInfo());
        default:
          throw new AdminException(reply.getInfo());
        }
      } else {
        return reply;
      }
    } catch (JMSException exc) {
      if (JoramTracing.dbgClient.isLoggable(BasicLevel.DEBUG))
        JoramTracing.dbgClient.log(
          BasicLevel.DEBUG, "", exc);
      throw new ConnectException("Connection failed: " + exc.getMessage());
    } catch (ClassCastException exc) {
      if (JoramTracing.dbgClient.isLoggable(BasicLevel.DEBUG))
        JoramTracing.dbgClient.log(
          BasicLevel.DEBUG, "", exc);
      throw new AdminException("Invalid server reply: " + exc.getMessage());
    }
  }

  public static void abortRequest() throws JMSException {
    if (requestor != null) {
      requestor.abort();
    } else throw new JMSException("Not connected");
  }

  public static class AdminRequestor {
    private javax.jms.TopicConnection cnx;
    private javax.jms.TopicSession sess;
    private javax.jms.Topic topic;
    private TemporaryTopic tmpTopic;
    private MessageProducer producer;
    private MessageConsumer consumer;

    public AdminRequestor(javax.jms.TopicConnection cnx)
      throws JMSException {
      this.cnx = cnx;
      sess = cnx.createTopicSession(
        false, Session.AUTO_ACKNOWLEDGE);
      topic = sess.createTopic("#AdminTopic");
      producer = sess.createProducer(topic);
      tmpTopic = sess.createTemporaryTopic();
      consumer = sess.createConsumer(tmpTopic);
    }

    public javax.jms.Message request(AdminRequest request,
                                     long timeout) throws JMSException {
      if (JoramTracing.dbgClient.isLoggable(BasicLevel.DEBUG))
        JoramTracing.dbgClient.log(
          BasicLevel.DEBUG,
          "AdminModule.AdminRequestor.request(" +
          request + ',' + timeout + ')');

      requestMsg = sess.createObjectMessage(request);
      requestMsg.setJMSReplyTo(tmpTopic);
      producer.send(requestMsg);
      String correlationId = requestMsg.getJMSMessageID();
      while (true) {
        javax.jms.Message reply = consumer.receive(timeout);
        if (reply == null) {
          throw new JMSException("Interrupted request");
        } else {
          if (correlationId.equals(
                reply.getJMSCorrelationID())) {
            return reply;
          } else {
            if (JoramTracing.dbgClient.isLoggable(BasicLevel.DEBUG))
              JoramTracing.dbgClient.log(
                BasicLevel.DEBUG,
                "reply id (" + reply.getJMSCorrelationID() +
                ") != request id (" + correlationId + ")");
            continue;
          }
        }
      }
    }

    public void abort() throws JMSException {
      if (JoramTracing.dbgClient.isLoggable(BasicLevel.DEBUG))
        JoramTracing.dbgClient.log(
          BasicLevel.DEBUG, "AdminModule.AdminRequestor.abort()");
      consumer.close();
      consumer = sess.createConsumer(tmpTopic);
    }

    public int getLocalServerId() {
      try {
        String topicName = topic.getTopicName();
        int ind0 = topicName.indexOf(".");
        int ind1 = topicName.indexOf(".", ind0 + 1);
        return Integer.parseInt(topicName.substring(ind0 + 1, ind1));
      } catch (JMSException exc) {
        return -1;
      }
    }

    public void close() throws JMSException {
      consumer.close();
      producer.close();
      tmpTopic.delete();
      sess.close();
    }
  }

  /**
   * This method execute the XML script file that the location is given
   * in parameter.
   *
   * @param cfgDir    The directory containing the file.
   * @param cfgFileName    The script filename.
   *
   * @since 4.3.10
   */
  public static boolean executeXMLAdmin(String cfgDir,
                                        String cfgFileName)
    throws Exception {
    return executeXMLAdmin(new File(cfgDir, cfgFileName).getPath());
  }

  /**
   * This method execute the XML script file that the pathname is given
   * in parameter.
   *
   * @param path    The script pathname.
   *
   * @since 4.3.10
   */
  public static boolean executeXMLAdmin(String path) throws Exception {
    if (JoramTracing.dbgAdmin.isLoggable(BasicLevel.DEBUG))
      JoramTracing.dbgAdmin.log(BasicLevel.DEBUG,"executeXMLAdmin(" + path + ")");

    boolean res = false;
    Reader reader = null;

    // 1st, search XML configuration file in directory.
    File cfgFile = new File(path);
    try {
      if (!cfgFile.exists() || !cfgFile.isFile() || (cfgFile.length() == 0)) {
        throw new IOException();
      }
      reader = new FileReader(cfgFile);
    } catch (IOException exc) {
      // configuration file seems not exist, search it from the
      // search path used to load classes.
      if (JoramTracing.dbgAdmin.isLoggable(BasicLevel.DEBUG))
        JoramTracing.dbgAdmin.log(BasicLevel.DEBUG,
                                  "Unable to find Joram Admin configuration file \"" +
                                  cfgFile.getPath() + "\".");
      reader = null;
    }

    // 2nd, search XML configuration file in path used to load classes.
    if (reader == null) {
      ClassLoader classLoader = null;
      InputStream is = null;
      try {
        classLoader = AdminModule.class.getClassLoader();
        if (classLoader != null) {
          if (JoramTracing.dbgAdmin.isLoggable(BasicLevel.DEBUG))
            JoramTracing.dbgAdmin.log(BasicLevel.DEBUG,
                                      "Trying to find [" + path + "] using " +
                                      classLoader + " class loader.");
          is = classLoader.getResourceAsStream(path);
        }
      } catch (Throwable t) {
        if (JoramTracing.dbgAdmin.isLoggable(BasicLevel.DEBUG))
          JoramTracing.dbgAdmin.log(BasicLevel.DEBUG,
                                    "Can't find [" + path + "] using " +
                                    classLoader + " class loader.",
                                    t);
        is = null;
      }

      if (is == null) {
        // Last ditch attempt: get the resource from the class path.
        if (JoramTracing.dbgAdmin.isLoggable(BasicLevel.DEBUG))
          JoramTracing.dbgAdmin.log(BasicLevel.DEBUG,
                                    "Trying to find [" + path +
                                    "] using ClassLoader.getSystemResource().");
        is = ClassLoader.getSystemResourceAsStream(path);
      }
      if (is != null) {
        res = executeAdmin(new InputStreamReader(is));
      }
    } else {
      res = executeAdmin(reader);
    }

    if (!res)
      throw new FileNotFoundException("xml Joram Admin configuration file not found.");

    return res;
  }

  public static boolean executeAdmin(Reader reader)
    throws Exception {
    if (JoramTracing.dbgAdmin.isLoggable(BasicLevel.DEBUG))
      JoramTracing.dbgAdmin.log(BasicLevel.DEBUG, "executeAdmin(" + reader + ")");

    String cfgName = System.getProperty(AdminModule.ADM_NAME_PROPERTY,
                                        AdminModule.DEFAULT_ADM_NAME);

    JoramSaxWrapper wrapper = new JoramSaxWrapper();
    return wrapper.parse(reader,cfgName);
  }


  public static void setHa(boolean isHa) {
    AdminModule.isHa = isHa;
  }
}