topicimpl.java

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

   */
  protected void doReact(AgentId from, UnclusterRequest request)
                 throws MomException
  {
    if (! isAdministrator(from))
      throw new AccessException("ADMIN right not granted");

    if (friends == null || friends.isEmpty()) {
      String info = strbuf.append("Request [")
        .append(request.getClass().getName())
        .append("], sent to Topic [").append(destId)
        .append("], successful [false]: topic not part of a cluster")
        .toString();
      strbuf.setLength(0);
      Channel.sendTo(from, new AdminReply(request, false, info));
      return;
    }

    UnclusterNot not = new UnclusterNot();
    AgentId fellowId;
    // Notifying each fellow of the leave.
    while (! friends.isEmpty()) {
      // state change, so save.
      setSave();
      fellowId = (AgentId) friends.remove(0);
      Channel.sendTo(fellowId, not);
    }
    friends = null;

    String info = strbuf.append("Request [")
      .append(request.getClass().getName())
      .append("], sent to Topic [").append(destId)
      .append("], successful [true]: topic left the cluster").toString();
    strbuf.setLength(0);
    Channel.sendTo(from, new AdminReply(request, true, info));

    if (MomTracing.dbgDestination.isLoggable(BasicLevel.DEBUG))
      MomTracing.dbgDestination.log(BasicLevel.DEBUG, info);
  }
 
  /**
   * Method implementing the reaction to an <code>UnclusterNot</code>
   * notification sent by a topic leaving the cluster.
   */
  protected void doReact(AgentId from, UnclusterNot not)
  {
    // state change, so save.
    setSave();
    friends.remove(from);

    if (friends.isEmpty()) friends = null;

    if (MomTracing.dbgDestination.isLoggable(BasicLevel.DEBUG))
      MomTracing.dbgDestination.log(BasicLevel.DEBUG, "Topic "
                                    + from.toString() + " removed from"
                                    + " cluster.");
  }

  /**
   * Method implementing the reaction to a <code>SetFatherRequest</code>
   * instance notifying this topic it is part of a hierarchy as a son.
   *
   * @exception AccessException  If the requester is not an administrator.
   */
  protected void doReact(AgentId from, SetFatherRequest request)
                 throws MomException
  {
    if (! isAdministrator(from))
      throw new AccessException("ADMIN right not granted");

    if (fatherId != null) {
      strbuf.append("Request [").append(request.getClass().getName())
        .append("], sent to Topic [").append(destId)
        .append("], successful [false]: topic already part of a hierarchy");
      Channel.sendTo(from, new AdminReply(request, false, strbuf.toString()));
      strbuf.setLength(0);
      return;
    }

    if (friends != null) {
      strbuf.append("Request [").append(request.getClass().getName())
        .append("], sent to Topic [").append(destId)
        .append("], successful [false]: topic already part of a cluster");
      Channel.sendTo(from, new AdminReply(request, false, strbuf.toString()));
      strbuf.setLength(0);
      return;
    }

    Channel.sendTo(request.getFatherId(), new FatherTest(request, from));
  }

  /**
   * Method reacting to a <code>FatherTest</code> notification checking if it
   * can be a father to a topic.
   */ 
  protected void doReact(AgentId from, FatherTest not)
  {
    if (friends != null && ! friends.isEmpty()) {
      strbuf.append("Topic [").append(destId)
        .append("] can't accept topic [").append(from)
        .append("] as a son as it is part of a cluster");
      Channel.sendTo(from, new FatherAck(not, false, strbuf.toString()));
      strbuf.setLength(0);
    } else {
      strbuf.append("Topic [").append(destId)
        .append("] accepts topic [").append(from).append("] as a son");
      Channel.sendTo(from, new FatherAck(not, true, strbuf.toString()));
      strbuf.setLength(0);
    }
  }

  /**
   * Method reacting to a <code>FatherAck</code> notification coming from
   * the topic this topic requested as a father.
   */ 
  protected void doReact(AgentId from, FatherAck not)
  {
    // The topic does not accept to join the hierarchy: doing nothing.
    if (! not.ok) {
      Channel.sendTo(not.requester,
                     new AdminReply(not.request, false, not.info));
      return;
    }
  
    // state change, so save.
    setSave();
    // The topic accepts to be a father: setting it.
    fatherId = from;
  
    String info = strbuf.append("Request [")
      .append(not.request.getClass().getName())
      .append("], sent to Topic [").append(destId)
      .append("], successful [true]: topic [")
      .append(from).append("] set as father").toString();
    strbuf.setLength(0);
    Channel.sendTo(not.requester, new AdminReply(not.request, true, info));
  
    if (MomTracing.dbgDestination.isLoggable(BasicLevel.DEBUG))
      MomTracing.dbgDestination.log(BasicLevel.DEBUG, info);
  }

  /**
   * Method implementing the reaction to an <code>UnsetFatherRequest</code>
   * instance notifying this topic to leave its father.
   *
   * @exception AccessException  If the requester is not an administrator.
   */
  protected void doReact(AgentId from, UnsetFatherRequest request)
                 throws MomException
  {
    if (! isAdministrator(from))
      throw new AccessException("ADMIN right not granted");

    String info = null;
    if (fatherId == null) {
      info = strbuf.append("Request [").append(request.getClass().getName())
        .append("], sent to Topic [").append(destId)
        .append("], successful [false]: topic is not a son").toString();
      strbuf.setLength(0);
      Channel.sendTo(from, new AdminReply(request, false, info));
      return;
    }

    // state change, so save.
    setSave();
    fatherId = null;

    info = strbuf.append("Request [").append(request.getClass().getName())
      .append("], sent to Topic [").append(destId)
      .append("], successful [true]: father unset").toString();
    strbuf.setLength(0);
    Channel.sendTo(from, new AdminReply(request, true, info));

    if (MomTracing.dbgDestination.isLoggable(BasicLevel.DEBUG))
      MomTracing.dbgDestination.log(BasicLevel.DEBUG, info);
  }

  /**
   * Method implementing the reaction to a
   * <code>Monit_GetSubscriptions</code> notification requesting the
   * number of subscriptions.
   *
   * @exception AccessException  If the requester is not the administrator.
   */
  protected void doReact(AgentId from, Monit_GetSubscriptions not)
                 throws AccessException
  {
    if (! isAdministrator(from))
      throw new AccessException("ADMIN right not granted");

    Channel.sendTo(from, new Monit_GetNumberRep(not, subscribers.size()));
  }

  /**
   * Method implementing the reaction to a <code>Monit_GetFather</code>
   * notification requesting the identifier of the hierarchical father.
   *
   * @exception AccessException  If the requester is not the administrator.
   */
  protected void doReact(AgentId from, Monit_GetFather not)
                 throws AccessException
  {
    if (! isAdministrator(from))
      throw new AccessException("ADMIN right not granted");

    String id = null;
    if (fatherId != null)
      id = fatherId.toString();

    Channel.sendTo(from, new Monit_GetFatherRep(not, id));
  }

  /**
   * Method implementing the reaction to a <code>Monit_GetCluster</code>
   * notification requesting the identifiers of the cluster's topics.
   *
   * @exception AccessException  If the requester is not the administrator.
   */
  protected void doReact(AgentId from, Monit_GetCluster not)
                 throws AccessException
  {
    if (! isAdministrator(from))
      throw new AccessException("ADMIN right not granted");

    Vector cluster = null;
    if (friends != null) {
      cluster = new Vector();
      for (int i = 0; i < friends.size(); i++)
        cluster.add(friends.get(i).toString());
      cluster.add(destId.toString());
    }

    Channel.sendTo(from, new Monit_GetClusterRep(not, cluster));
  }

  /**
   * Method implementing the reaction to a <code>SubscribeRequest</code>
   * instance. 
   *
   * @exception AccessException  If the sender is not a READER.
   */
  protected void doReact(AgentId from, SubscribeRequest not)
                 throws AccessException
  {
    if (! isReader(from))
      throw new AccessException("READ right not granted");

    // Adding new subscriber.
    if (! subscribers.contains(from)) {
      // state change, so save.
      setSave();
      subscribers.add(from);
    }

    // state change, so save.
    setSave();

    // The requester might either be a new subscriber, or an existing one;
    // setting the selector, possibly by removing or modifying an already set
    // expression.
    if (not.getSelector() != null && ! not.getSelector().equals(""))
      selectors.put(from, not.getSelector());
    else
      selectors.remove(from);

    Channel.sendTo(from, new SubscribeReply(not));

    if (MomTracing.dbgDestination.isLoggable(BasicLevel.DEBUG))
      MomTracing.dbgDestination.log(BasicLevel.DEBUG, 
                                    "Client " + from
                                    + " set as a subscriber with selector "
                                    + not.getSelector());
  }

  /**
   * Method implementing the reaction to an <code>UnsubscribeRequest</code>
   * instance, requesting to remove a subscriber.
   */
  protected void doReact(AgentId from, UnsubscribeRequest not)
  {
    // state change, so save.
    setSave();
    subscribers.remove(from);
    selectors.remove(from);

    if (MomTracing.dbgDestination.isLoggable(BasicLevel.DEBUG))
      MomTracing.dbgDestination.log(BasicLevel.DEBUG, 
                                    "Client " + from
                                    + " removed from the subscribers.");
  } 

  /**
   * Method implementing the reaction to a <code>TopicForwardNot</code>
   * instance, carrying messages forwarded by a cluster fellow or a
   * hierarchical son.
   */
  protected void doReact(AgentId from, TopicForwardNot not)
  {
    // If the forward comes from a son, forwarding it to the father, if any.
    if (not.toFather && fatherId != null) {
      Channel.sendTo(fatherId, not);
      alreadySentLocally = fatherId.getTo() == AgentServer.getServerId();
    }
    
    // Processing the received messages. 
    processMessages(not.messages);
  }

  private void doReact(AgentId from, DestinationAdminRequestNot not) {
    org.objectweb.joram.shared.admin.AdminRequest adminRequest = 
      not.getRequest();