admintopicimpl.java

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

  {
    AgentId initId = AgentId.fromString(request.getInitId());
    AgentId topId = AgentId.fromString(request.getTopId());

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

  /**
   * Processes an <code>UnsetCluster<code> instance requesting a topic to
   * leave the cluster it is part of.
   */
  private void doProcess(UnsetCluster request,
                         AgentId replyTo,
                         String msgId) throws UnknownServerException
  {
    AgentId topId = AgentId.fromString(request.getTopId());

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

  /**
   * Processes a <code>SetFather<code> instance requesting to link two topics
   * in a hierarchical relationship.
   */
  private void doProcess(SetFather request,
                         AgentId replyTo,
                         String msgId) throws UnknownServerException
  {
     AgentId fatherId = AgentId.fromString(request.getFather());
     AgentId sonId = AgentId.fromString(request.getSon());

     if (checkServerId(sonId.getTo())) {
       // If the son is local, process the request.
       Channel.sendTo(sonId, new SetFatherRequest(msgId, fatherId));
       if (replyTo != null) requestsTable.put(msgId, replyTo);
    } else {
      // Forward the request to the right AdminTopic agent.
      Channel.sendTo(AdminTopic.getDefault(sonId.getTo()),
             new AdminRequestNot(replyTo, msgId, request));
    }
  }

  /**
   * Processes an <code>UnsetFather<code> instance requesting a topic to
   * unset its hierarchical father.
   */
  private void doProcess(UnsetFather request,
                         AgentId replyTo,
                         String msgId) throws UnknownServerException
  {
    AgentId topId = AgentId.fromString(request.getTopId());

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

  /**
   * Processes a <code>CreateUserRequest</code> instance requesting the
   * creation of a <code>UserAgent</code> for a given user.
   *
   * @exception UnknownServerException  If the target server does not exist.
   * @exception RequestException  If the user already exists but with a
   *              different password, or if the proxy deployment failed.
   */
  private void doProcess(CreateUserRequest request,
                         AgentId replyTo,
                         String msgId)
               throws UnknownServerException, RequestException
  {
    if (checkServerId(request.getServerId())) {
      // If this server is the target server, process the request.
      String name = request.getUserName();
      String pass = request.getUserPass();

      AgentId proxId = (AgentId) proxiesTable.get(name);
      String info;

      // The user has already been set. 
      if (proxId != null) {
        if (! pass.equals((String) usersTable.get(name))) {
          throw new RequestException("User [" + name + "] already exists"
                                     + " but with a different password.");
        }
        info = strbuf.append("Request [").append(request.getClass().getName())
          .append("], processed by AdminTopic on server [").append(serverId)
          .append("], successful [true]: proxy [").append(proxId.toString())
          .append("] of user [").append(name)
          .append("] has been retrieved").toString();
        strbuf.setLength(0);
      } else {
        UserAgent proxy = new UserAgent();
        if (name != null) {
          proxy.name = name;
        }
        proxId = proxy.getId();

        try {
          proxy.deploy();
          usersTable.put(name, request.getUserPass());
          proxiesTable.put(name, proxy.getId());
  
          info = strbuf.append("Request [").append(request.getClass().getName())
            .append("], processed by AdminTopic on server [").append(serverId)
            .append("], successful [true]: proxy [")
            .append(proxId.toString()).append("] for user [").append(name)
            .append("] has been created and deployed").toString();
          strbuf.setLength(0);
        }
        catch (Exception exc) {
          throw new RequestException("User proxy not deployed: " + exc);
        }
      }

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

      distributeReply(replyTo,
                      msgId,
                      new CreateUserReply(proxId.toString(), 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>UpdateUser</code> instance requesting to modify the
   * identification of a user.
   *
   * @exception RequestException  If the user does not exist, or if it's new
   *              name is already used.
   */
  private void doProcess(UpdateUser request,
                         AgentId replyTo,
                         String msgId)
               throws RequestException, UnknownServerException
  {
    String name = request.getUserName();
    AgentId proxId = AgentId.fromString(request.getProxId());

    if (checkServerId(proxId.getTo())) {
      // If the user belong to this server, process the request.
      String info;

      // If the user does not exist: throwing an exception:
      if (! usersTable.containsKey(name))
        throw new RequestException("User [" + name + "] does not exist");

      String newName = request.getNewName();
      // If the new name is already taken by an other user than the modified
      // one:
      if (! newName.equals(name)
          && (usersTable.containsKey(newName)))
        throw new RequestException("Name [" + newName + "] already used");

      String newPass = request.getNewPass();

      if (usersTable.containsKey(name)) {
        usersTable.remove(name);
        proxiesTable.remove(name);
        usersTable.put(newName, request.getNewPass());
        proxiesTable.put(newName, proxId);
      }

      info = strbuf.append("Request [").append(request.getClass().getName())
        .append("], processed by AdminTopic on server [").append(serverId)
        .append("], successful [true]: user [").append(name)
        .append("] has been updated to [").append(newName).
        append("]").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(proxId.getTo()),
             new AdminRequestNot(replyTo, msgId, request));
    }
  }

  /**
   * Processes a <code>DeleteUser</code> instance requesting the deletion
   * of a user.
   */
  private void doProcess(DeleteUser request,
                         AgentId replyTo,
                         String msgId) throws UnknownServerException
  {
    String name = request.getUserName();
    AgentId proxId = AgentId.fromString(request.getProxId());

    if (checkServerId(proxId.getTo())) {
      // If the user belong to this server, process the request.
      String info;

      if (usersTable.containsKey(name)) {
        Channel.sendTo(proxId, new DeleteNot());
        usersTable.remove(name);
        proxiesTable.remove(name);
    
        info = strbuf.append("Request [").append(request.getClass().getName())
          .append("], sent to AdminTopic on server [").append(serverId)
          .append("], successful [true]: proxy [").append(proxId)
          .append("], of user [").append(name)
          .append("] has been notified of deletion").toString();
        strbuf.setLength(0);
      } else {
        info = strbuf.append("Request [").append(request.getClass().getName())
          .append("], sent to AdminTopic on server [").append(serverId)
          .append("], successful [false]: user [").append(name)
          .append(" does not exist").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(proxId.getTo()),
             new AdminRequestNot(replyTo, msgId, request));
    }
  }

  /**
   * Processes a <code>SetRight</code> instance requesting to grant a user
   * a given right on a given destination.
   */
  private void doProcess(SetRight request,
                         AgentId replyTo,
                         String msgId) throws UnknownServerException
  {
    AgentId destId = AgentId.fromString(request.getDestId());

    if (checkServerId(destId.getTo())) {
      // If the destination belong to this server, process request
      
      AgentId userId = null;
      if (request.getUserProxId() != null)
        userId = AgentId.fromString(request.getUserProxId());

      int right = 0;
      if (request instanceof SetReader)
        right = READ;
      else if (request instanceof SetWriter)
        right = WRITE;
      else if (request instanceof UnsetReader)
        right = - READ;
      else if (request instanceof UnsetWriter)
        right = - WRITE;

      Channel.sendTo(destId, new SetRightRequest(msgId, userId, right));
      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>SetDefaultDMQ</code> request requesting a given
   * dead message queue to be set as the default one.
   *
   * @exception UnknownServerException  If the target server does not exist.
   */
  private void doProcess(SetDefaultDMQ request,
                         AgentId replyTo,
                         String msgId) throws UnknownServerException
  {
    if (checkServerId(request.getServerId())) {
      // If this server is the target server, process the request.
      String info;

      AgentId dmqId = null;
      if (request.getDmqId() != null)
        dmqId = AgentId.fromString(request.getDmqId());

      DeadMQueueImpl.id = dmqId;

      info = strbuf.append("Request [").append(request.getClass().getName())
        .append("], sent to AdminTopic on server [").append(serverId)
        .append("], successful [true]: dmq [").append(dmqId.toString())
        .append("], has been successfuly set as the default one").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>SetDestinationDMQ</code> request requesting a given
   * dead message queue to be set as the DMQ of a given destination.
   */
  private void doProcess(SetDestinationDMQ request,
                         AgentId replyTo,
                         String msgId) throws UnknownServerException
  {
    AgentId destId = AgentId.fromString(request.getDestId());

    if (checkServerId(destId.getTo())) {
      // The destination is local, process the request.
      AgentId dmqId = AgentId.fromString(request.getDmqId());
      Channel.sendTo(destId, new SetDMQRequest(msgId, dmqId));
      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>SetUserDMQ</code> request requesting a given
   * dead message queue to be set as the DMQ of a given user.
   */