admintopicimpl.java

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

  private void doProcess(SetUserDMQ request,
                         AgentId replyTo,
                         String msgId) throws UnknownServerException
  {
    AgentId userId = AgentId.fromString(request.getUserProxId());

    if (checkServerId(userId.getTo())) {
      // The user is local, process the request.
      AgentId dmqId = AgentId.fromString(request.getDmqId());
      Channel.sendTo(userId, new SetDMQRequest(msgId, dmqId));
      if (replyTo != null) requestsTable.put(msgId, replyTo);
    } else {
      // Forward the request to the right AdminTopic agent.
      Channel.sendTo(AdminTopic.getDefault(userId.getTo()),
             new AdminRequestNot(replyTo, msgId, request));
    }
  } 

  /**
   * Processes a <code>SetDefaultThreshold</code> request requesting a given
   * threshold value to be set as the default one.
   *
   * @exception UnknownServerException  If the target server does not exist.
   */
  private void doProcess(SetDefaultThreshold request,
                         AgentId replyTo,
                         String msgId) throws UnknownServerException
  {
    if (checkServerId(request.getServerId())) {
      // If this server is the target server, process the request.
      String info;

      DeadMQueueImpl.threshold = new Integer(request.getThreshold());

      info = strbuf.append("Request [").append(request.getClass().getName())
        .append("], sent to AdminTopic on server [").append(serverId)
        .append("], successful [true]: default threshold [")
        .append(request.getThreshold()).append("] has been set").toString();
      strbuf.setLength(0);

      distributeReply(replyTo, msgId, new AdminReply(true, info)); 

      if (MomTracing.dbgDestination.isLoggable(BasicLevel.DEBUG))
        MomTracing.dbgDestination.log(BasicLevel.DEBUG, info);
    } else {
      // Forward the request to the right AdminTopic agent.
      Channel.sendTo(AdminTopic.getDefault((short) request.getServerId()),
             new AdminRequestNot(replyTo, msgId, request));
    }
  }

  /**
   * Processes a <code>SetNbMaxMsg</code> request requesting
   * a given nbMaxMsg value to be set in queue or subscription.
   */
  private void doProcess(SetNbMaxMsg request,
                         AgentId replyTo,
                         String msgId) 
    throws UnknownServerException {
    AgentId destId = AgentId.fromString(request.getId());

    if (checkServerId(destId.getTo())) {
      // The destination is not local, doing nothing.
      int nbMaxMsg = request.getNbMaxMsg();
      String subName = request.getSubName();
      Channel.sendTo(destId, 
                     new SetNbMaxMsgRequest(msgId, nbMaxMsg, subName));
      if (replyTo != null) requestsTable.put(msgId, replyTo);
    } else {
      // Forward the request to the right AdminTopic agent.
      Channel.sendTo(AdminTopic.getDefault(destId.getTo()),
             new AdminRequestNot(replyTo, msgId, request));
    }
  }

  /**
   * Processes a <code>SetQueueThreshold</code> request requesting
   * a given threshold value to be set as the threshold of a given
   * queue.
   */
  private void doProcess(SetQueueThreshold request,
                         AgentId replyTo,
                         String msgId) throws UnknownServerException
  {
    AgentId destId = AgentId.fromString(request.getQueueId());

    if (checkServerId(destId.getTo())) {
      // The destination is not local, doing nothing.
      int thresh = request.getThreshold();
      Channel.sendTo(destId, new SetThreshRequest(msgId, new Integer(thresh)));
      if (replyTo != null) requestsTable.put(msgId, replyTo);
    } else {
      // Forward the request to the right AdminTopic agent.
      Channel.sendTo(AdminTopic.getDefault(destId.getTo()),
             new AdminRequestNot(replyTo, msgId, request));
    }
  }

  /**
   * Processes a <code>SetUserThreshold</code> request requesting
   * a given threshold value to be set as the threshold of a given
   * user.
   */
  private void doProcess(SetUserThreshold request,
                         AgentId replyTo,
                         String msgId) throws UnknownServerException
  {
    AgentId userId = AgentId.fromString(request.getUserProxId());

    if (checkServerId(userId.getTo())) {
      // The user is local, process the request.
      int thresh = request.getThreshold();
      Channel.sendTo(userId, new SetThreshRequest(msgId, new Integer(thresh)));
      if (replyTo != null) requestsTable.put(msgId, replyTo);
    } else {
      // Forward the request to the right AdminTopic agent.
      Channel.sendTo(AdminTopic.getDefault(userId.getTo()),
             new AdminRequestNot(replyTo, msgId, request));
    }
  }

  /**
   * Processes an <code>UnsetDefaultDMQ</code> request requesting to unset
   * the default DMQ of a given server.
   *
   * @exception UnknownServerException  If the target server does not exist.
   */
  private void doProcess(UnsetDefaultDMQ request,
                         AgentId replyTo,
                         String msgId) throws UnknownServerException
  {
    if (checkServerId(request.getServerId())) {
      // If this server is the target server, process the request.
      String info;

      DeadMQueueImpl.id = null;

      info = strbuf.append("Request [").append(request.getClass().getName())
        .append("], sent to AdminTopic on server [").append(serverId)
        .append("], successful [true]: default dmq has been unset").toString();
      strbuf.setLength(0);
      
      distributeReply(replyTo, msgId, new AdminReply(true, info)); 
      
      if (MomTracing.dbgDestination.isLoggable(BasicLevel.DEBUG))
        MomTracing.dbgDestination.log(BasicLevel.DEBUG, info);
    } else {
      // Forward the request to the right AdminTopic agent.
      Channel.sendTo(AdminTopic.getDefault((short) request.getServerId()),
             new AdminRequestNot(replyTo, msgId, request));
    }
  }

  /**
   * Processes an <code>UnsetDestinationDMQ</code> request requesting to unset
   * the DMQ of a given destination.
   */
  private void doProcess(UnsetDestinationDMQ request,
                         AgentId replyTo,
                         String msgId) throws UnknownServerException
  {
    AgentId destId = AgentId.fromString(request.getDestId());

    if (checkServerId(destId.getTo())) {
      // The destination is local, process the request.
      Channel.sendTo(destId, new SetDMQRequest(msgId, null));
      if (replyTo != null) requestsTable.put(msgId, replyTo);
    } else {
      // Forward the request to the right AdminTopic agent.
      Channel.sendTo(AdminTopic.getDefault(destId.getTo()),
             new AdminRequestNot(replyTo, msgId, request));
    }
  }

  /**
   * Processes an <code>UnsetUserDMQ</code> request requesting to unset
   * the DMQ of a given user.
   */
  private void doProcess(UnsetUserDMQ request,
                         AgentId replyTo,
                         String msgId) throws UnknownServerException
  {
    AgentId userId = AgentId.fromString(request.getUserProxId());

    if (checkServerId(userId.getTo())) {
      // The user is local, process the request.
      Channel.sendTo(userId, new SetDMQRequest(msgId, null));
      if (replyTo != null)requestsTable.put(msgId, replyTo);
    } else {
      // Forward the request to the right AdminTopic agent.
      Channel.sendTo(AdminTopic.getDefault(userId.getTo()),
             new AdminRequestNot(replyTo, msgId, request));
    }
  }

  /**
   * Processes an <code>UnsetDefaultThreshold</code> request requesting
   * to unset the default threshold value.
   *
   * @exception UnknownServerException  If the target server does not exist.
   */
  private void doProcess(UnsetDefaultThreshold request,
                         AgentId replyTo,
                         String msgId) throws UnknownServerException
  {
    if (checkServerId(request.getServerId())) {
      // If this server is the target server, process the request.
      String info;

      DeadMQueueImpl.threshold = null;

      info = strbuf.append("Request [").append(request.getClass().getName())
        .append("], sent to AdminTopic on server [").append(serverId)
        .append("], successful [true]: default threshold has been unset")
        .toString();
      strbuf.setLength(0);

      distributeReply(replyTo, msgId, new AdminReply(true, info)); 

      if (MomTracing.dbgDestination.isLoggable(BasicLevel.DEBUG))
        MomTracing.dbgDestination.log(BasicLevel.DEBUG,
                                      "Default threshold unset.");
    } else {
      // Forward the request to the right AdminTopic agent.
      Channel.sendTo(AdminTopic.getDefault((short) request.getServerId()),
             new AdminRequestNot(replyTo, msgId, request));
    }
  }

  /**
   * Processes an <code>UnsetQueueThreshold</code> request requesting
   * to unset the threshold of a given queue.
   */
  private void doProcess(UnsetQueueThreshold request,
                         AgentId replyTo,
                         String msgId) throws UnknownServerException
  {
     AgentId destId = AgentId.fromString(request.getQueueId());

     if (checkServerId(destId.getTo())) {
       // The destination is local, process the request.
       Channel.sendTo(destId, new SetThreshRequest(msgId, null));
       if (replyTo != null) requestsTable.put(msgId, replyTo);
     } else {
       // Forward the request to the right AdminTopic agent.
       Channel.sendTo(AdminTopic.getDefault(destId.getTo()),
             new AdminRequestNot(replyTo, msgId, request));
    }
  }

  /**
   * Processes an <code>UnsetUserThreshold</code> request requesting to unset
   * the threshold of a given user.
   */
  private void doProcess(UnsetUserThreshold request,
                         AgentId replyTo,
                         String msgId) throws UnknownServerException
  {
    AgentId userId = AgentId.fromString(request.getUserProxId());

    if (checkServerId(userId.getTo())) {
      // The user is local, process the request.
      Channel.sendTo(userId, new SetThreshRequest(msgId, null));
      if (replyTo != null) requestsTable.put(msgId, replyTo);
    } else {
      // Forward the request to the right AdminTopic agent.
      Channel.sendTo(AdminTopic.getDefault(userId.getTo()),
             new AdminRequestNot(replyTo, msgId, request));
    }
  }

  /**
   * Processes a <code>Monitor_GetServersIds</code> request by sending 
   * the list of the platform servers' ids.
   *
   * @exception UnknownServerException  If the target server does not exist.
   */
  private void doProcess(Monitor_GetServersIds request,
                         AgentId replyTo,
                         String msgId) throws UnknownServerException {
    if (checkServerId(request.getServerId())) {
      try {
        int[] ids;
        String[] names;
        String[] hostNames;
        String domainName = request.getDomainName();
        Enumeration servers;
        A3CMLConfig config = AgentServer.getConfig();
        int serversCount;
        if (domainName != null) {
          A3CMLDomain domain = config.getDomain(domainName);
          servers = domain.servers.elements();
          serversCount = domain.servers.size();
        } else {
          servers = config.servers.elements();
          serversCount = config.servers.size();
        }
        ids = new int[serversCount];
        names = new String[serversCount];
        hostNames = new String[serversCount];
        int i = 0;
        while (servers.hasMoreElements()) {
          A3CMLServer server = (A3CMLServer)servers.nextElement();
          ids[i] = (int)server.sid;
          names[i] = server.name;
          hostNames[i] = server.hostname;
          i++;
        }
        Monitor_GetServersIdsRep reply = new Monitor_GetServersIdsRep(
          ids, names, hostNames);
        distributeReply(replyTo, msgId, reply);
      } catch (Exception exc) {
        if (MomTracing.dbgDestination.isLoggable(BasicLevel.DEBUG))
          MomTracing.dbgDestination.log(BasicLevel.DEBUG, "", exc);
        distributeReply(replyTo, msgId,
                        new AdminReply(false, exc.toString()));
      }
    } else {
      // Forward the request to the right AdminTopic agent.
      Channel.sendTo(AdminTopic.getDefault((short) request.getServerId()),
             new AdminRequestNot(replyTo, msgId, request));
    }
  }

  private void doProcess(GetLocalServer request,
                         AgentId replyTo,
                         String msgId) throws UnknownServerException {
    try {
      A3CMLConfig config = AgentServer.getConfig();
      A3CMLServer a3cmlServer = config.getServer(AgentServer.getServerId());
      distributeReply(replyTo, msgId,
                      new GetLocalServerRep(a3cmlServer.sid,
                                            a3cmlServer.name,
                                            a3cmlServer.hostname));
    } catch (Exception exc) {
      if (MomTracing.dbgDestination.isLoggable(BasicLevel.DEBUG))
        MomTracing.dbgDestination.log(BasicLevel.DEBUG, "", exc);
      distributeReply(replyTo, msgId,
                      new AdminReply(false, exc.toString()));
    }
  }

  private void doProcess(GetDomainNames request,
                         AgentId replyTo,
                         String msgId) {
    try