replicationmanager.java

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

/*
 * JORAM: Java(TM) Open Reliable Asynchronous Messaging
 * Copyright (C) 2001 - 2003 ScalAgent Distributed Technologies
 *
 * 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): David Feliot
 */
package fr.dyade.aaa.jndi2.distributed;

import java.io.IOException;
import java.util.Enumeration;
import java.util.Hashtable;
import java.util.Vector;

import javax.naming.CompositeName;
import javax.naming.NamingException;

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

import fr.dyade.aaa.agent.AgentId;
import fr.dyade.aaa.agent.AgentServer;
import fr.dyade.aaa.jndi2.impl.BindEvent;
import fr.dyade.aaa.jndi2.impl.CreateSubcontextEvent;
import fr.dyade.aaa.jndi2.impl.DestroySubcontextEvent;
import fr.dyade.aaa.jndi2.impl.MissingContextException;
import fr.dyade.aaa.jndi2.impl.MissingRecordException;
import fr.dyade.aaa.jndi2.impl.NamingContext;
import fr.dyade.aaa.jndi2.impl.NamingContextInfo;
import fr.dyade.aaa.jndi2.impl.NotOwnerException;
import fr.dyade.aaa.jndi2.impl.RebindEvent;
import fr.dyade.aaa.jndi2.impl.UnbindEvent;
import fr.dyade.aaa.jndi2.impl.UpdateEvent;
import fr.dyade.aaa.jndi2.impl.UpdateListener;
import fr.dyade.aaa.jndi2.msg.ChangeOwnerRequest;
import fr.dyade.aaa.jndi2.msg.CreateSubcontextRequest;
import fr.dyade.aaa.jndi2.msg.JndiError;
import fr.dyade.aaa.jndi2.msg.JndiReply;
import fr.dyade.aaa.jndi2.msg.JndiRequest;
import fr.dyade.aaa.jndi2.server.JndiReplyNot;
import fr.dyade.aaa.jndi2.server.JndiScriptReplyNot;
import fr.dyade.aaa.jndi2.server.JndiScriptRequestNot;
import fr.dyade.aaa.jndi2.server.RequestContext;
import fr.dyade.aaa.jndi2.server.RequestManager;
import fr.dyade.aaa.jndi2.server.Trace;

public class ReplicationManager 
    extends RequestManager implements UpdateListener {

  public final static String INIT_REQUEST_TABLE = "initRequestTable";
  public final static String SYNC_REQUEST_TABLE = "syncRequestTable";
  public final static String WRITE_REQUEST_TABLE = "writeRequestTable";
  public final static String SERVER_LIST = "serverList";

  /**
   * Identifier of the server that owns
   * the root naming context.
   */
  private AgentId rootOwnerId;

  /**
   * List of the initially known
   * servers.
   */
  private short[] serverIds;

  /**
   * List of the JNDI servers discovered by
   * this server. These servers first receive
   * an initialization notification that contain
   * the naming data owned by this server. Then they
   * receive the update notifications about
   * the contexts owned by this server.
   */
  private transient Vector servers;

  /**
   * Table that contains the write requests
   * forwarded to the owner.
   * key = owner identifier (AgentId)
   * value = requests list (RequestContextList)
   */
  private transient Hashtable writeRequestContextLists;

  /**
   * Table that contains the requests (read or write)
   * waiting for the initialization of the context.
   * key = id of the missing context
   * value = requests list (RequestContextList)
   */
  private transient Hashtable initRequestContextLists;

  /**
   * Table that contains the requests (read or write)
   * waiting for the synchronization of the context.
   * key = owner identifier (AgentId)
   * value = requests list (RequestContextList)
   */
  private transient Hashtable syncRequestContextLists;

  public ReplicationManager(short[] serverIds) {
    this.serverIds = serverIds;
  }

  /**
   * Overrides the <code>JndiServer</code> behavior.
   */
  protected AgentId getRootOwnerId() {
    return rootOwnerId;
  }

  public void agentInitialize(boolean firstTime) throws Exception {
    if (firstTime) {
      if (serverIds.length > 0) {
        rootOwnerId = 
          DistributedJndiServer.getDefault(serverIds[0]);
      } else {
        rootOwnerId = getId();
      }
    }

    super.agentInitialize(firstTime);

    writeRequestContextLists = 
      (Hashtable)AgentServer.getTransaction().load(
        WRITE_REQUEST_TABLE);
    if (writeRequestContextLists == null) {
      writeRequestContextLists = new Hashtable();
    }

    initRequestContextLists = 
      (Hashtable)AgentServer.getTransaction().load(
        INIT_REQUEST_TABLE);
    if (initRequestContextLists == null) {
      initRequestContextLists = new Hashtable();
    }

    syncRequestContextLists =
      (Hashtable)AgentServer.getTransaction().load(
        SYNC_REQUEST_TABLE);
    if (syncRequestContextLists == null) {
      syncRequestContextLists = new Hashtable();
    }

    servers = 
      (Vector)AgentServer.getTransaction().load(
        SERVER_LIST);
    if (servers == null) {
      servers = new Vector();      
      for (int i = 0; i < serverIds.length; i++) {        
        AgentId aid = DistributedJndiServer.getDefault(serverIds[i]);
        servers.addElement(aid);
        sendTo(aid, new InitJndiServerNot(
          null, null, true));
      }
      saveServers();
    }

    getServerImpl().setUpdateListener(this);
  }

  /**
   * Reacts to an update notification from an other JNDI server.
   */
  void doReact(AgentId from, JndiUpdateNot not) throws Exception {
    if (Trace.logger.isLoggable(BasicLevel.DEBUG))
      Trace.logger.log(BasicLevel.DEBUG, 
                       "ReplicationManager[" + getId() + "].doReact(" +
                       from + ',' + not + ')');
    UpdateEvent updateEvent = not.getUpdateEvent();
    try {
      if (updateEvent instanceof BindEvent) {
        onUpdateEvent(from, (BindEvent)updateEvent);
      } else if (updateEvent instanceof RebindEvent) {
        onUpdateEvent(from, (RebindEvent)updateEvent);
      } else if (updateEvent instanceof UnbindEvent) {
        onUpdateEvent(from, (UnbindEvent)updateEvent);
      } else if (updateEvent instanceof CreateSubcontextEvent) {
        onUpdateEvent(from, (CreateSubcontextEvent)updateEvent);
      } else if (updateEvent instanceof DestroySubcontextEvent) {
        onUpdateEvent(from, (DestroySubcontextEvent)updateEvent);
//       } else if (updateEvent instanceof ChangeOwnerEvent) {
//         onUpdateEvent(from, (ChangeOwnerEvent)updateEvent);
      }
    } catch (NotOwnerException exc) {       
      // This may happen after a change owner event
      Trace.logger.log(BasicLevel.WARN,
                       "Distributed jndi update warn:",
                       exc);
    } catch (NamingException exc) {       
      Trace.logger.log(BasicLevel.ERROR, 
                       "Distributed jndi update error:",
                       exc);
      throw new Error(exc.toString());
    }
  }  

  private void onUpdateEvent(AgentId from, BindEvent evt) 
    throws NamingException {
    getServerImpl().bind(
      getServerImpl().getNamingContext(
        evt.getUpdatedContextId()),
      evt.getName(), 
      evt.getObject(),
      from);
  }

  private void onUpdateEvent(AgentId from, RebindEvent evt) 
    throws NamingException {
    getServerImpl().rebind(
      getServerImpl().getNamingContext(
        evt.getUpdatedContextId()),
      evt.getName(), 
      evt.getObject(),
      from);
  }

  private void onUpdateEvent(AgentId from, UnbindEvent evt) 
    throws NamingException {
    getServerImpl().unbind(
      getServerImpl().getNamingContext(
        evt.getUpdatedContextId()),
      evt.getName(),
      from);
  }

  private void onUpdateEvent(AgentId from, CreateSubcontextEvent evt) 
    throws NamingException {
    getServerImpl().createSubcontext(
      getServerImpl().getNamingContext(
        evt.getUpdatedContextId()),
      evt.getName(),
      evt.getPath(),
      evt.getContextId(),
      evt.getOwnerId(),
      from);
  }

  private void onUpdateEvent(AgentId from, DestroySubcontextEvent evt) 
    throws NamingException {
    getServerImpl().destroySubcontext(
      getServerImpl().getNamingContext(
        evt.getUpdatedContextId()),
      evt.getName(), 
      evt.getPath(),
      from);
  }

  // private void onUpdateEvent(AgentId from, ChangeOwnerEvent evt) 
//     throws NamingException {
//     NamingContextInfo[] contexts = evt.getNamingContexts();
//     for (int i = 0; i < contexts.length; i++) {
//       NamingContext nc = getServerImpl().getNamingContext(
//         contexts[i].getNamingContext().getId());
//       if (nc == null) {
//         // The InitJndiServerNot sent by 
//         // the server that created this context may not
//         // have been received.
//         getServerImpl().addNamingContext(contexts[i]);
//         retryRequestsWaitingForMissingContext(
//           contexts[i].getNamingContext().getId());
//       } else {
//         getServerImpl().resetNamingContext(
//           contexts[i].getNamingContext());
//         // DF: must retry the sync and write
//         // requests to the new owner.
//       }
//     }
//   }  

  /**
   * Overrides the <code>JndiServer</code> behavior.
   * Send a JNDI request to the owner (JNDI server).
   * Waits for the asynchronous reply.
   */
  protected JndiReply invokeOwner(AgentId owner,
                                  RequestContext reqCtx) {
    if (Trace.logger.isLoggable(BasicLevel.DEBUG))
      Trace.logger.log(BasicLevel.DEBUG, 
                       "ReplicationManager.invokeOwner(" +
                       owner + ',' + reqCtx + ')');
    JndiRequest request = reqCtx.getRequest();    
    if (request instanceof CreateSubcontextRequest) {
      CreateSubcontextRequest csr = 
        (CreateSubcontextRequest)request;
      request = new CreateRemoteSubcontextRequest(
        csr.getName(), 
        getId());
    }

    sendTo(owner,
           new JndiScriptRequestNot(
             new JndiRequest[]{request}, true));
    RequestContextList list = 
      (RequestContextList)writeRequestContextLists.get(owner);
    if (list == null) {
      list = new RequestContextList();
      writeRequestContextLists.put(owner, list);
    }
    list.put(reqCtx);
    saveWriteRequestTable();
    return null;
  }

  void doReact(AgentId from, JndiScriptReplyNot not) 
    throws Exception {
    onReply(from, not.getReplies()[0]);
  }

  void doReact(AgentId from, JndiReplyNot not) 
    throws Exception {
    onReply(from, not.getReply());