admintopicimpl.java

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

/*
 * JORAM: Java(TM) Open Reliable Asynchronous Messaging
 * Copyright (C) 2001 - 2006 ScalAgent Distributed Technologies
 * Copyright (C) 1996 - 2000 Dyade
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or any later version.
 * 
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 * 
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307
 * USA.
 *
 * Initial developer(s): Frederic Maistre (INRIA)
 * Contributor(s): ScalAgent Distributed Technologies
 */
package org.objectweb.joram.mom.dest;

import fr.dyade.aaa.agent.AgentId;
import fr.dyade.aaa.agent.Agent;
import fr.dyade.aaa.agent.AgentServer;
import fr.dyade.aaa.agent.Channel;
import fr.dyade.aaa.agent.DeleteNot;
import fr.dyade.aaa.agent.Notification;
import fr.dyade.aaa.agent.UnknownAgent;
import fr.dyade.aaa.agent.UnknownNotificationException;
import fr.dyade.aaa.agent.UnknownServerException;
import fr.dyade.aaa.agent.ConfigController;
import fr.dyade.aaa.agent.ServerConfigHelper;
import fr.dyade.aaa.agent.conf.A3CMLConfig;
import fr.dyade.aaa.agent.conf.A3CML;
import fr.dyade.aaa.agent.conf.A3CMLNetwork;
import fr.dyade.aaa.agent.conf.A3CMLDomain;
import fr.dyade.aaa.agent.conf.A3CMLServer;
import org.objectweb.joram.mom.MomTracing;
import org.objectweb.joram.shared.admin.*;
import org.objectweb.joram.shared.admin.AdminRequest;
import org.objectweb.joram.shared.admin.AdminReply;
import org.objectweb.joram.mom.notifications.*;
import org.objectweb.joram.shared.excepts.*;
import org.objectweb.joram.shared.messages.Message;
import org.objectweb.joram.mom.proxies.AdminNotification;
import org.objectweb.joram.mom.proxies.UserAgent;
import org.objectweb.joram.mom.proxies.SendReplyNot;

import java.util.Enumeration;
import java.util.Hashtable;
import java.util.Vector;
import java.util.Properties;
import java.io.ByteArrayOutputStream;
import java.io.PrintWriter;
import java.lang.reflect.*;

import org.objectweb.util.monolog.api.BasicLevel;

/**
 * The <code>AdminTopicImpl</code> class implements the admin topic behaviour,
 * basically processing administration requests.
 */
public final class AdminTopicImpl extends TopicImpl implements AdminTopicImplMBean {

  /** Reference of the server's local AdminTopicImpl instance. */
  private static AdminTopicImpl ref;
  
  public static AdminTopicImpl getReference() {
    return ref;
  }

  /** Identifier of the server this topic is deployed on. */
  private int serverId;

  /**
   * Table holding the local server's destinations names.
   * <p>
   * <b>Key:</b> destination name<br>
   * <b>Object:</b> destination agent identifier
   */
  private Hashtable destinationsTable;

  /**
   * Table holding the TCP users identifications.
   * <p>
   * <b>Key:</b> user name<br>
   * <b>Object:</b> user password
   */
  private Hashtable usersTable;
  /**
   * Table holding the TCP users proxies identifiers.
   * <p>
   * <b>Key:</b> user name<br>
   * <b>Object:</b> proxy's identifier
   */
  private Hashtable proxiesTable;

  /**
   * Table keeping the administrator's requests.
   * <p>
   * <b>Key:</b> request's message identifier<br>
   * <b>Value:</b> request's message ReplyTo field
   */
  private Hashtable requestsTable;
  /** Counter of messages produced by this AdminTopic. */
  private long msgCounter = 0;

  /**
   * Identifier of the server's default dead message queue, kept here for
   * persisting it.
   */
  private AgentId defaultDMQId;
  /** Server's default threshold value, kept here for persisting it. */
  private Integer defaultThreshold;

  /**
   * Constructs an <code>AdminTopicImpl</code> instance.
   *
   * @param topicId  Identifier of the agent hosting the AdminTopicImpl.
   */
  public AdminTopicImpl(AgentId topicId) {
    super(topicId, topicId, null);
    serverId = AgentServer.getServerId();
    destinationsTable = new Hashtable();
    usersTable = new Hashtable();
    proxiesTable = new Hashtable();
    requestsTable = new Hashtable();
  }

  public String toString() {
    return "AdminTopicImpl:" + destId.toString();
  }


  /**
   * Method used by <code>ConnectionManager</code> proxies to check their
   * clients identification.
   *
   * @param name
   * @param pass
   * @param inaddr
   * @exception Exception  If the user does not exist, is wrongly identified,
   *              or does not have any proxy deployed.
   * @see org.objectweb.joram.mom.proxies.ConnectionManager
   */
  public AgentId getProxyId(String name,
                            String pass,
                            String inaddr) throws Exception {
    String userPass = null;
    AgentId userProxId = null;

    userPass = (String) usersTable.get(name);
    if (userPass != null) {
      if (! userPass.equals(pass))
        throw new Exception("Invalid password for user [" + name + "]");
      userProxId = (AgentId) proxiesTable.get(name);
      if (userProxId == null)
        throw new Exception("No proxy deployed for user [" + name + "]");

      return userProxId;
    }
    else
      throw new Exception("User [" + name + "] does not exist");
  }

  /** Method used by proxies for retrieving their name. */
  public String getName(AgentId proxyId) {
    String name;
    for (Enumeration e = proxiesTable.keys(); e.hasMoreElements();) {
      name = (String) e.nextElement();
      if (proxyId.equals(proxiesTable.get(name)))
        return name;
    }
    return null;
  }

  /** Method used by proxies for retrieving their password. */
  public String getPassword(AgentId proxyId) {
    String name;
    for (Enumeration e = proxiesTable.keys(); e.hasMoreElements();) {
      name = (String) e.nextElement();
      if (proxyId.equals(proxiesTable.get(name)))
        return (String) usersTable.get(name);;
    }
    return null;
  }

  /** Method used by proxies for checking if a given name is already used. */
  public boolean isTaken(String name) {
    return usersTable.containsKey(name);
  }
  
  /** Method returning the id of the admin topic. */ 
  public AgentId getId() {
    return destId;
  }
  
  /**
   * Distributes the received notifications to the appropriate reactions.
   *
   * @exception UnknownNotificationException  If a received notification is
   *              unexpected by the AdminTopic.
   */
  public void react(AgentId from, Notification not)
              throws UnknownNotificationException {
    if (MomTracing.dbgDestination.isLoggable(BasicLevel.DEBUG))
      MomTracing.dbgDestination.log(BasicLevel.DEBUG, "--- " + this
                                    + ": got " + not
                                    + " from: " + from.toString());

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

    if (not instanceof AdminNotification)
      doReact(from, (AdminNotification) not);
    else if (not instanceof AdminRequestNot)
      doReact(from, (AdminRequestNot) not);
    else if (not instanceof org.objectweb.joram.mom.notifications.AdminReply)
      doReact(from, (org.objectweb.joram.mom.notifications.AdminReply) not);
    else if (not instanceof GetProxyIdNot)
      doReact((GetProxyIdNot)not);
    else if (not instanceof GetProxyIdListNot)
      doReact((GetProxyIdListNot)not);
    else if (not instanceof RegisterTmpDestNot)
      doReact((RegisterTmpDestNot)not);
    else if (not instanceof RegisterDestNot)
      doReact((RegisterDestNot)not);
    else if (not instanceof RegisteredDestNot)
      doReact(from, (RegisteredDestNot)not);
    else
      super.react(from, not);
  }

  /**
   * Method implementing the reaction to a
   * <code>org.objectweb.joram.mom.proxies.AdminNotification</code>
   * notification notifying of the creation of an admin proxy.
   */
  protected void doReact(AgentId from, AdminNotification adminNot) {
    String name = adminNot.getName();
    String pass = adminNot.getPass();

    usersTable.put(name, pass);
    proxiesTable.put(name, adminNot.getProxyId());

    clients.put(adminNot.getProxyId(), new Integer(READWRITE));
   
    if (MomTracing.dbgDestination.isLoggable(BasicLevel.DEBUG))
      MomTracing.dbgDestination.log(BasicLevel.DEBUG, name + " successfully"
                                    + " set as admin client.");
  }

  /**
   * Method implementing the reaction to a <code>AdminRequest</code>
   * notification notifying of the creation of an admin proxy.
   */
  protected void doReact(AgentId from, AdminRequestNot adminNot) {
    // AF: verify that from is an AdminTopic
    processAdminRequests(adminNot.replyTo, adminNot.msgId, adminNot.request, from);
  }

  /**
   * Method implementing the reaction to a
   * <code>org.objectweb.joram.mom.notifications.AdminReply</code>
   * notification replying to an administration request.
   * <p>
   * A reply is sent back to the connected administrator if needed.
   */  
  protected void doReact(AgentId from,
                         org.objectweb.joram.mom.notifications.AdminReply not) {
    String requestId = not.getRequestId();
    if (requestId == null) return;

    AgentId replyTo = (AgentId) requestsTable.remove(requestId);
    if (replyTo == null) return;

    AdminReply reply;

    if (not instanceof Monit_GetUsersRep)
      reply = doProcess((Monit_GetUsersRep) not);
    else if (not instanceof Monit_FreeAccessRep)
      reply = doProcess((Monit_FreeAccessRep) not);
    else if (not instanceof Monit_GetDMQSettingsRep)
      reply = doProcess((Monit_GetDMQSettingsRep) not);
    else if (not instanceof Monit_GetFatherRep)
      reply = doProcess((Monit_GetFatherRep) not);
    else if (not instanceof Monit_GetClusterRep)
      reply = doProcess((Monit_GetClusterRep) not);
    else if (not instanceof Monit_GetNumberRep)
      reply = doProcess((Monit_GetNumberRep) not);
    else if (not instanceof Monit_GetStatRep)
      reply = doProcess((Monit_GetStatRep) not);
    else if (not instanceof Monit_GetNbMaxMsgRep)
      reply = doProcess((Monit_GetNbMaxMsgRep) not);
    else
      reply = new AdminReply(not.getSuccess(), 
                             not.getInfo(),
                             not.getReplyObject());
    
    distributeReply(replyTo, requestId, reply);
  }

  protected void doReact(GetProxyIdNot not) {
    try {
      AgentId proxyId = getProxyId(not.getUserName(), 
				   not.getPassword(),
                                   not.getInAddr());
      not.Return(proxyId);
    } catch (Exception exc) {
      not.Throw(exc);
    }
  }

  protected void doReact(GetProxyIdListNot not) {
    Vector idList = new Vector();
    Enumeration ids = proxiesTable.elements();
    while (ids.hasMoreElements()) {
      AgentId aid = (AgentId)ids.nextElement();
      idList.addElement(aid);
    }
    AgentId[] res = new AgentId[idList.size()];
    idList.copyInto(res);
    not.Return(res);
  }

  private void doReact(RegisterTmpDestNot not) {
    String destName = not.getTmpDestId().toString();
    if (not.toAdd()) {
      String type;
      String className;
      if (not.isTopic()) {
        type = "topic.tmp";
        className = Topic.class.getName();
      } else {
        type = "queue.tmp";
        className = Queue.class.getName();
      }
      DestinationDesc destDesc = 
        new DestinationDesc(
          not.getTmpDestId(),
          destName,
          className,
          type);
      destinationsTable.put(destName, destDesc);
    } else {