admintopicimpl.java

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

      destinationsTable.remove(destName);
    }
  }

  private void doReact(RegisterDestNot not) {
    String name = not.getName();
    if (name == null || destinationsTable.contains(name))
      return;
    
    DestinationDesc destDesc = 
      new DestinationDesc(
        not.getId(),
        not.getName(),
        not.getClassName(),
        not.getType());
    destinationsTable.put(name, destDesc);
  }
  
  private void doReact(AgentId from, RegisteredDestNot not) {
    DestinationDesc destDesc = 
      (DestinationDesc) destinationsTable.get(not.getName());
    if (destDesc != null)
      not.setDestination(destDesc.getId());
    Channel.sendTo(not.getReply(), not);
  }
  
  /**
   * Processes a <code>Monit_GetUsersRep</code> notification holding a
   * destination's readers' or writers' identifiers.
   */
  private AdminReply doProcess(Monit_GetUsersRep not) {
    Vector users = not.getUsers();

    String name;
    AgentId proxyId;
    Monitor_GetUsersRep reply = new Monitor_GetUsersRep();

    for (Enumeration names = proxiesTable.keys(); names.hasMoreElements();) {
      name = (String) names.nextElement();
      proxyId = (AgentId) proxiesTable.get(name);

      if (users.contains(proxyId))
        reply.addUser(name, proxyId.toString());
    }
    return reply;
  }

  /**
   * Processes a <code>Monit_FreeAccesRep</code> notification holding the
   * free access status of a destination.
   */
  private AdminReply doProcess(Monit_FreeAccessRep not) {
    return new Monitor_GetFreeAccessRep(not.getFreeReading(),
                                        not.getFreeWriting());
  }

  /**
   * Processes a <code>Monit_GetDMQSettingsRep</code> notification holding the
   * DMQ settings of a destination or proxy dead message queue.
   */
  private AdminReply doProcess(Monit_GetDMQSettingsRep not) {
    return new Monitor_GetDMQSettingsRep(not.getDMQId(), not.getThreshold());
  }

  /**
   * Processes a <code>Monit_GetFatherRep</code> notification holding the
   * identifier of a topic's hierarchical father.
   */
  private AdminReply doProcess(Monit_GetFatherRep not) {
    return new Monitor_GetFatherRep(not.getFatherId());
  }

  /**
   * Processes a <code>Monit_GetClusterRep</code> notification holding the
   * identifiers of a cluster's topics.
   */
  private AdminReply doProcess(Monit_GetClusterRep not) {
    return new Monitor_GetClusterRep(not.getTopics());
  }

  /**
   * Processes a <code>Monit_GetNumberRep</code> notification holding an
   * integer value sent by a destination.
   */
  private AdminReply doProcess(Monit_GetNumberRep not) {
    return new Monitor_GetNumberRep(not.getNumber());
  }

  /**
   * Processes a <code>Monit_GetStatRep</code> notification holding a
   * statistic sent by a destination.
   */
  private AdminReply doProcess(Monit_GetStatRep not) {
    return new Monitor_GetStatRep(not.getStats());
  }

  /**
   * Processes a <code>Monit_GetNbMaxMsgRep</code> notification holding a
   * nbMaxMsg sent by a destination.
   */
  private AdminReply doProcess(Monit_GetNbMaxMsgRep not) {
    return new Monitor_GetNbMaxMsgRep(not.getNbMaxMsg());
  }

  /**
   * Overrides this <code>DestinationImpl</code> method; AdminTopics do not
   * accept <code>SetRightRequest</code> notifications.
   *
   * @exception  AccessException  Not thrown.
   */ 
  protected void doReact(AgentId from, SetRightRequest request)
                 throws AccessException {
    if (MomTracing.dbgDestination.isLoggable(BasicLevel.WARN))
      MomTracing.dbgDestination.log(BasicLevel.WARN,
                                    "Unexpected request: " + request);
  }

  /**
   * Overrides this <code>DestinationImpl</code> method; AdminTopics do not
   * accept <code>SetDMQRequest</code> notifications.
   *
   * @exception  AccessException  Not thrown.
   */ 
  protected void doReact(AgentId from, SetDMQRequest request)
                 throws AccessException {
    if (MomTracing.dbgDestination.isLoggable(BasicLevel.WARN))
      MomTracing.dbgDestination.log(BasicLevel.WARN,
                                    "Unexpected request: " + request);
  }

  protected void doReact(AgentId from, RequestGroupNot not) {
    if (MomTracing.dbgDestination.isLoggable(BasicLevel.DEBUG))
      MomTracing.dbgDestination.log(BasicLevel.DEBUG,
          "AdminTopicImpl.doReact(" + not + ')');
    Enumeration en = not.getClientMessages();
    while (en.hasMoreElements()) {
      ClientMessages cm = (ClientMessages) en.nextElement();
      try {
        doReact(from, cm);
      } catch (Exception exc) {
      }
    }
  }
  
  /**
   * Overrides this <code>DestinationImpl</code> method;
   * <code>ClientMessages</code> notifications hold requests sent by an
   * administrator.
   *
   * @exception AccessException  If the requester is not a WRITER on the
   *              AdminTopic.
   */
  protected void doReact(AgentId from, ClientMessages not)
                 throws AccessException {
    if (MomTracing.dbgDestination.isLoggable(BasicLevel.DEBUG))
      MomTracing.dbgDestination.log(BasicLevel.DEBUG,
          "AdminTopicImpl.doReact(" + not + ')');
    if (! not.getPersistent() && !not.getAsyncSend()) {
      // Means that this notification has been sent by a local
      // proxy (optimization). Must acknowledge it.
      Channel.sendTo(
        from, 
        new SendReplyNot(
          not.getClientContext(), 
          not.getRequestId()));
    }
    
    if (! isWriter(from))
      throw new AccessException("WRITE right not granted");

    // ... and processing the wrapped requests locally:
    processAdminRequests(not);
  }

  /**
   * Overrides this <code>DestinationImpl</code> method; deletion requests are
   * not accepted by AdminTopics.
   */
  protected void doReact(AgentId from, DeleteNot not) {
    if (MomTracing.dbgDestination.isLoggable(BasicLevel.WARN))
      MomTracing.dbgDestination.log(BasicLevel.WARN,
                                    "Unexpected request: " + not);
  }

  /**
   * Overrides this <code>TopicImpl</code> method; AdminTopics do not
   * accept <code>ClusterRequest</code> notifications.
   *
   * @exception  AccessException  Not thrown.
   */ 
  protected void doReact(AgentId from, ClusterRequest request)
                 throws AccessException
  {
    if (MomTracing.dbgDestination.isLoggable(BasicLevel.WARN))
      MomTracing.dbgDestination.log(BasicLevel.WARN,
                                    "Unexpected request: " + request);
  }

  /**
   * Overrides this <code>TopicImpl</code> method; AdminTopics do not
   * accept to join clusters other than their admin topics cluster.
   */ 
  protected void doReact(AgentId from, ClusterTest request) {
    Channel.sendTo(from, new ClusterAck(request, false,
                                        "Topic [" + destId
                                        + "] is an admin topic"));
  }

  /**
   * Overrides this <code>TopicImpl</code> method; a <code>ClusterAck</code>
   * is not expected by an AdminTopic.
   */ 
  protected void doReact(AgentId from, ClusterAck ack) {
    if (MomTracing.dbgDestination.isLoggable(BasicLevel.WARN))
      MomTracing.dbgDestination.log(BasicLevel.WARN,
                                    "Unexpected notification: " + ack);
  }

  /**
   * Overrides this <code>TopicImpl</code> method; if this AdminTopic is on
   * server0, new cluster fellow is notified to other fellows and other
   * fellows are notified to it.
   */
  protected void doReact(AgentId from, ClusterNot not) {
    if (MomTracing.dbgDestination.isLoggable(BasicLevel.WARN))
      MomTracing.dbgDestination.log(BasicLevel.WARN,
                                    "Unexpected notification: " + not);
  }

  /**
   * Overrides this <code>TopicImpl</code> method; AdminTopics do not
   * accept <code>UnclusterRequest</code> notifications.
   *
   * @exception AccessException  Not thrown.
   */ 
  protected void doReact(AgentId from, UnclusterRequest request)
                 throws MomException {
    if (MomTracing.dbgDestination.isLoggable(BasicLevel.WARN))
      MomTracing.dbgDestination.log(BasicLevel.WARN,
                                    "Unexpected request: " + request);
  }

  /**
   * Overrides this <code>TopicImpl</code> method; AdminTopics do not
   * accept <code>SetFatherRequest</code> notifications.
   *
   * @exception  AccessException  Not thrown.
   */ 
  protected void doReact(AgentId from, SetFatherRequest request)
                 throws MomException {
    if (MomTracing.dbgDestination.isLoggable(BasicLevel.WARN))
      MomTracing.dbgDestination.log(BasicLevel.WARN,
                                    "Unexpected request: " + request);
  }

  /**
   * Overrides this <code>TopicImpl</code> method; AdminTopics do not
   * accept to join a hierarchy.
   */ 
  protected void doReact(AgentId from, FatherTest not) {
    Channel.sendTo(from, new FatherAck(not, false,
                                       "Topic [" + destId
                                       + "] can't accept topic [" + from
                                       + "] as a son as it is an AdminTopic"));
  }

  /**
   * Overrides this <code>TopicImpl</code> method; a <code>FatherAck</code>
   * acknowledges the process of creating a hierarchy of topics.
   */ 
  protected void doReact(AgentId from, FatherAck ack) {
    if (MomTracing.dbgDestination.isLoggable(BasicLevel.WARN))
      MomTracing.dbgDestination.log(BasicLevel.WARN,
                                    "Unexpected notification: " + ack);
  }

  /**
   * Overrides this <code>TopicImpl</code> method; AdminTopics do not
   * accept <code>UnsetFatherRequest</code> notifications.
   *
   * @exception  AccessException  Not thrown.
   */ 
  protected void doReact(AgentId from, UnsetFatherRequest request)
                 throws MomException {
    if (MomTracing.dbgDestination.isLoggable(BasicLevel.WARN))
      MomTracing.dbgDestination.log(BasicLevel.WARN,
                                    "Unexpected request: " + request);
  }

  
  /**
   * Overrides this <code>TopicImpl</code> method; the forwarded messages
   * contain admin requests and will be processed.
   */
  protected void doReact(AgentId from, TopicForwardNot not) {
    processAdminRequests(not.messages);
  }

  /**
   * Specializes this <code>TopicImpl</code> reaction.
   */
  protected void doProcess(UnknownAgent uA) {
    AgentId agId = uA.agent;
    Notification not = uA.not;

    // For admin requests, notifying the administrator.
    if (not instanceof org.objectweb.joram.mom.notifications.AdminRequest) {
      String reqId =
        ((org.objectweb.joram.mom.notifications.AdminRequest) not).getId();

      if (reqId != null) {
        AgentId replyTo = (AgentId) requestsTable.remove(reqId);

        String info = strbuf.append("Request [")
          .append(not.getClass().getName())
          .append("], sent to AdminTopic on server [")
          .append(serverId)
          .append("], successful [false]: unknown agent [")
          .append(agId).append("]").toString();
        strbuf.setLength(0);

        distributeReply(replyTo, reqId, new AdminReply(false, info));
      }
    } else {
      super.doProcess(uA);
    }
  }

  /**
   * Method getting the administration requests from messages, and
   * distributing them to the appropriate reactions.
   */ 
  private void processAdminRequests(ClientMessages not) {
    Message msg;
    String msgId = null;
    AgentId replyTo = null;
    String info = null;
    AdminRequest request = null;

    Enumeration messages = not.getMessages().elements();

    while (messages.hasMoreElements()) {
      nbMsgsReceiveSinceCreation = nbMsgsReceiveSinceCreation + 1;
      msg = (Message) messages.nextElement();
      msgId = msg.getIdentifier();
      replyTo = AgentId.fromString(msg.getReplyToId());
      request = null;

      try {