useragent.java

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

/*
 * JORAM: Java(TM) Open Reliable Asynchronous Messaging
 * Copyright (C) 2004 - 2006 ScalAgent Distributed Technologies
 * Copyright (C) 2004 France Telecom R&D
 *
 * 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): ScalAgent Distributed Technologies
 */
package org.objectweb.joram.mom.proxies;

import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.util.Collection;
import java.util.Enumeration;
import java.util.Hashtable;
import java.util.Iterator;

import fr.dyade.aaa.agent.Agent;
import fr.dyade.aaa.agent.AgentId;
import fr.dyade.aaa.agent.BagSerializer;
import fr.dyade.aaa.agent.Notification;
import fr.dyade.aaa.agent.UnknownNotificationException;
import fr.dyade.aaa.util.management.MXWrapper;
import fr.dyade.aaa.util.Queue;

import org.objectweb.joram.mom.dest.AdminTopicImpl;
import org.objectweb.joram.shared.client.AbstractJmsReply;
import org.objectweb.joram.shared.client.AbstractJmsRequest;
import org.objectweb.joram.shared.client.CnxCloseRequest;
import org.objectweb.joram.shared.client.JmsRequestGroup;
import org.objectweb.joram.shared.client.ProducerMessages;
import org.objectweb.joram.shared.client.ServerReply;

import org.objectweb.joram.mom.MomTracing;
import org.objectweb.util.monolog.api.BasicLevel;

/** 
 * Class of a user proxy agent.
 */
public class UserAgent extends Agent implements BagSerializer, ProxyAgentItf {
  /**
   * All the user requests are delegated
   * to the proxy
   */
  private ProxyImpl proxyImpl;

  /**
   * Table that contains the user connections
   * key = <code>Integer</code> (connection key)
   * value = <code></code>
   */
  private transient Hashtable connections;

  private transient Hashtable heartBeatTasks;

  /**
   * Counter of the connection keys
   */
  private int keyCounter;

  /**
   * Creates a new user proxy.
   *
   * @see AdminTopicImpl
   * @see ConnectionManager
   */
  public UserAgent() {
    super(true);
    init();
  }

  /**
   * Creates a new user proxy.
   *
   * @see AdminTopicImpl
   * @see ConnectionManager
   */
  public UserAgent(int stamp) {
    super("JoramAdminProxy", true, stamp);
    init();
  }

  private void init() {
    proxyImpl = new ProxyImpl(this);
    keyCounter = 0;
  }

  /** (Re)initializes the agent when (re)loading. */
  public void agentInitialize(boolean firstTime) throws Exception {
    if (MomTracing.dbgProxy.isLoggable(BasicLevel.DEBUG))
      MomTracing.dbgProxy.log(BasicLevel.DEBUG, "UserAgent.agentInitialize("
          + firstTime + ')');
    super.agentInitialize(firstTime);
    proxyImpl.initialize(firstTime);
    MXWrapper.registerMBean(proxyImpl, "Joram", getMBeanName());
  }

  /** Finalizes the agent before it is garbaged. */
  public void agentFinalize(boolean lastTime) {
    try {
      MXWrapper.unregisterMBean("Joram", getMBeanName());
    } catch (Exception exc) {
      if (MomTracing.dbgProxy.isLoggable(BasicLevel.DEBUG))
        MomTracing.dbgProxy.log(BasicLevel.DEBUG, "", exc);
    }
    super.agentFinalize(lastTime);
  }

  private String getMBeanName() {
    return new StringBuffer().append("type=User").append(",name=").append(
        (name == nullName) ? getId().toString() : name).toString();
  }

  /**
   * Overrides the <code>Agent</code> class <code>react</code> method for
   * providing the JMS client proxy with its specific behaviour.
   * <p>
   * A JMS proxy specifically reacts to the following notifications:
   * <ul>
   * <li><code>OpenConnectionNot</code></li>
   * </ul>
   */
  public void react(AgentId from, Notification not) throws Exception {
    if (MomTracing.dbgProxy.isLoggable(BasicLevel.DEBUG))
      MomTracing.dbgProxy.log(BasicLevel.DEBUG, "UserAgent.react(" + from + ','
          + not + ')');

    // set agent no save:
    // the default behavior is transient
    setNoSave();

    if (not instanceof OpenConnectionNot) {
      doReact((OpenConnectionNot) not);
    } else if (not instanceof GetConnectionNot) {
      doReact((GetConnectionNot) not);
    } else if (not instanceof CloseConnectionNot) {
      doReact((CloseConnectionNot) not);
    } else if (not instanceof ResetCollocatedConnectionsNot) {
      doReact((ResetCollocatedConnectionsNot) not);
    } else if (not instanceof SendReplyNot) {
      doReact((SendReplyNot) not);
    } else if (not instanceof RequestNot) {
      doReact((RequestNot) not);
    } else if (not instanceof ReturnConnectionNot) {
      doReact((ReturnConnectionNot) not);
    } else if (not instanceof SendRepliesNot) {
      doReact((SendRepliesNot) not);
    } else if (not instanceof ProxyRequestGroupNot) {
      doReact((ProxyRequestGroupNot) not);
    } else {
      try {
        proxyImpl.react(from, not);
      } catch (UnknownNotificationException exc) {
        super.react(from, not);
      }
    }
  }

  /**
   * Registers and starts the <code>UserConnection</code>.
   */
  private void doReact(OpenConnectionNot not) {
    // state change, so save.
    setSave();

    if (connections == null) {
      connections = new Hashtable();
      heartBeatTasks = new Hashtable();
    }

    Integer objKey = new Integer(keyCounter);
    ConnectionContext ctx;
    if (not.getReliable()) {
      ctx = new ReliableConnectionContext(
          proxyImpl, keyCounter,
          not.getHeartBeat());
      connections.put(objKey, ctx);
    } else {
      ctx = new StandardConnectionContext(
          proxyImpl, keyCounter);
      connections.put(objKey, ctx);
    }

    if (not.getHeartBeat() > 0) {
      HeartBeatTask heartBeatTask = new HeartBeatTask(2 * not.getHeartBeat(),
          objKey);
      heartBeatTasks.put(objKey, heartBeatTask);
      heartBeatTask.start();
    }

    // Differs the reply because the connection key counter
    // must be saved before the OpenConnectionNot returns.
    sendTo(getId(), new ReturnConnectionNot(not, ctx));
    keyCounter++;
  }

  /**
   * Differs the reply because the connection key counter
   * must be saved before the OpenConnectionNot returns.
   */
  private void doReact(ReturnConnectionNot not) {
    not.Return();
  }

  private void doReact(GetConnectionNot not) {
    int key = not.getKey();
    if (connections == null) {
      not.Throw(new Exception("Connection " + key + " not found"));
    } else {
      Integer objKey = new Integer(key);
      ReliableConnectionContext ctx = (ReliableConnectionContext) connections
          .get(objKey);
      if (ctx == null) {
        not.Throw(new Exception("Connection " + key + " not found"));
      } else {
        not.Return(ctx);
      }
    }
  }

  private void doReact(RequestNot not) {
    Integer key = new Integer(not.getConnectionKey());
    if (connections != null) {
      ConnectionContext ctx = (ConnectionContext) connections.get(key);
      if (ctx != null) {
        HeartBeatTask heartBeatTask = (HeartBeatTask) heartBeatTasks.get(key);
        if (heartBeatTask != null) {