replicationmanager.java

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

  }
  
  /**
   * Reacts to the reply of a JNDI server that has been called
   * as it is the owner of a naming context.
   */
  private void onReply(AgentId from, JndiReply reply) throws Exception {
    if (Trace.logger.isLoggable(BasicLevel.DEBUG))
      Trace.logger.log(BasicLevel.DEBUG, 
                       "ReplicationManager[" + 
                       getId() + "].onReply(" +
                       from + ',' + reply + ')');
    RequestContextList ctxList = 
      (RequestContextList)writeRequestContextLists.get(from);
    RequestContext ctx = ctxList.get();
    ctxList.pop();
    if (ctxList.getSize() == 0) {
      writeRequestContextLists.remove(from);
    }    
    if (ctx != null) {
      ctx.reply(reply);
      saveWriteRequestTable();
    } else {
      Trace.logger.log(BasicLevel.ERROR,
                       "Reply context not found: " + 
                       from + ", " + reply);
    }
  }

  void doReact(AgentId from, 
               InitJndiServerNot not) throws Exception {
    if (Trace.logger.isLoggable(BasicLevel.DEBUG))
      Trace.logger.log(BasicLevel.DEBUG, 
                       "ReplicationManager.doReact(" +
                       from + ',' + not + ')');
    AgentId[] jndiServerIds = not.getJndiServerIds();
    Vector initServers = new Vector();
    if (jndiServerIds != null) {
      for (int i = 0; i < jndiServerIds.length; i++) {
        if (servers.indexOf(jndiServerIds[i]) < 0) {
          initServers.addElement(jndiServerIds[i]);
        }
      }
    }
    
    // Send back an init notif if:
    // - the init notif is a request
    // - or the server 'from' is unknown
    if (not.isRequest() || servers.indexOf(from) < 0) {
      initServers.addElement(from);
    }

    if (Trace.logger.isLoggable(BasicLevel.DEBUG))
      Trace.logger.log(BasicLevel.DEBUG, 
                       " -> initServers = " + initServers);

    if (initServers.size() > 0) {
      AgentId[] localJndiServerIds = new AgentId[servers.size()];
      servers.copyInto(localJndiServerIds);
      NamingContextInfo[] localContexts = 
        getServerImpl().copyNamingContexts(getId());
      int serversInitialLength = servers.size();
      for (int i = 0; i < initServers.size(); i++) {
        AgentId newServerId = 
          (AgentId)initServers.elementAt(i);
        sendTo(newServerId, new InitJndiServerNot(
          localJndiServerIds, 
          localContexts,
          (!from.equals(newServerId))));
        if (servers.indexOf(newServerId) < 0) {
          servers.addElement(newServerId);
        }
        // else the server has already been registered.
        // (it is a recovering server)
      }
      if (servers.size() > serversInitialLength) {
        saveServers();
      }
    }

    NamingContextInfo[] contexts = not.getContexts();
    if (contexts != null) {
      Vector newNames = new Vector();
      for (int i = 0; i < contexts.length; i++) {
        NamingContext nc = getServerImpl().getNamingContext(
          contexts[i].getNamingContext().getId());
        if (nc == null) {
          getServerImpl().addNamingContext(contexts[i]);
          newNames.addElement(contexts[i].getCompositeName());
        }
        // Else the naming context has already been
        // added by an other server that is the (new)
        // owner of this context.
      }
      
      Vector retryNames = new Vector();
      Vector retryLists = new Vector();
      Enumeration names = initRequestContextLists.keys();
      Enumeration lists = initRequestContextLists.elements();
      while (lists.hasMoreElements()) {
        CompositeName name = 
          (CompositeName)names.nextElement();
        RequestContextList ctxList = 
          (RequestContextList)lists.nextElement();
        boolean retry = false;
        for (int i = 0; i < newNames.size(); i++) {
          CompositeName newName = 
            (CompositeName)newNames.elementAt(i);
          if (name.startsWith(newName)) {
            retry = true;
            break;
          }
        }
        if (retry) {
          retryNames.addElement(name);
          retryLists.addElement(ctxList);
        }
      }
      
      for (int i = 0; i < retryNames.size(); i++) {
        CompositeName name = 
          (CompositeName)retryNames.elementAt(i);
        RequestContextList ctxList = 
          (RequestContextList)retryLists.elementAt(i);
        initRequestContextLists.remove(name);
        while (ctxList.getSize() > 0) {
          RequestContext reqCtx = ctxList.get();
          JndiReply reply = invoke(reqCtx);
          if (reply != null) {
            reqCtx.reply(reply);
          }
          ctxList.pop();
        }
      }
      saveInitRequestTable();
    }
  }

  protected JndiReply onMissingContext(MissingContextException mce,
                                       RequestContext reqCtx) {
    if (Trace.logger.isLoggable(BasicLevel.DEBUG))
      Trace.logger.log(BasicLevel.DEBUG, 
                       "ReplicationManager.onMissingContext(" +
                       mce + ',' + reqCtx + ')');
    RequestContextList ctxList = 
      (RequestContextList)initRequestContextLists.get(
        mce.getName());
    if (ctxList == null) {
      ctxList = new RequestContextList();
      if (Trace.logger.isLoggable(BasicLevel.DEBUG))
        Trace.logger.log(BasicLevel.DEBUG, 
                         " -> add a waiting request context: " + 
                         mce.getName());
      initRequestContextLists.put(
        mce.getName(), ctxList);
    }
    ctxList.put(reqCtx);
    saveInitRequestTable();
    return null;
  }

  protected JndiReply onMissingRecord(MissingRecordException mre,
                                      RequestContext reqCtx) {
    if (Trace.logger.isLoggable(BasicLevel.DEBUG))
      Trace.logger.log(BasicLevel.DEBUG, 
                       "ReplicationManager.onMissingRecord(" +
                       mre + ',' + reqCtx + ')');
    CompositeName resolvedName = 
      (CompositeName)mre.getNameNotFoundException().getResolvedName();
    if (mre.getOwnerId().equals(getId()) ||
        resolvedName.equals(reqCtx.getResolvedName())) {
      // The resolved context has already been updated.
      return new JndiError(mre.getNameNotFoundException());
    } else {
      reqCtx.setResolvedName(resolvedName);
      synchronizeRequest(
        (AgentId)mre.getOwnerId(), reqCtx);
      return null;
    }
  }

  private void synchronizeRequest(AgentId owner, 
                                  RequestContext reqCtx) {
    if (Trace.logger.isLoggable(BasicLevel.DEBUG))
      Trace.logger.log(
        BasicLevel.DEBUG, 
        "ReplicationManager.synchronizeRequest(" +
        owner + ',' + reqCtx + ')');
    sendTo(owner, new SyncRequestNot());
    RequestContextList list = 
      (RequestContextList)syncRequestContextLists.get(owner);
    if (list == null) {
      list = new RequestContextList();
      syncRequestContextLists.put(owner, list);      
    }
    list.put(reqCtx);
    saveSyncRequestTable();
  }

  void doReact(AgentId from, SyncRequestNot not) {
    sendTo(from, new SyncReplyNot());
  }

  void doReact(AgentId from, SyncReplyNot not) {
    RequestContextList ctxList = 
      (RequestContextList)syncRequestContextLists.get(from);
    RequestContext ctx = ctxList.get();
    ctxList.pop();
    if (ctxList.getSize() == 0) {
      syncRequestContextLists.remove(from);
    }    
    if (ctx != null) {
      JndiReply reply = invoke(ctx);
      if (reply != null) {
        ctx.reply(reply);
        saveSyncRequestTable();
      } 
    }
  }

  public void onUpdate(UpdateEvent event) {
    for (int i = 0; i < servers.size(); i++) {
      AgentId aid = (AgentId)servers.elementAt(i);
      sendTo(aid, new JndiUpdateNot(event));
    }
  }

  protected void createSubcontext(CreateSubcontextRequest request) 
    throws NamingException {
    if (request instanceof CreateRemoteSubcontextRequest) {
      createRemoteSubcontext((CreateRemoteSubcontextRequest)request);
    } else {
      super.createSubcontext(request);
    }
  }

  private void createRemoteSubcontext(CreateRemoteSubcontextRequest request)
    throws NamingException {
    getServerImpl().createSubcontext(
      request.getName(), request.getOwnerId());
  }

  protected void changeOwner(ChangeOwnerRequest request)
    throws NamingException {
    super.changeOwner(request);

    writeRequestContextLists.remove(request.getOwnerId());
    syncRequestContextLists.remove(request.getOwnerId());
    // DF: must reply to those requests because
    // this server is the new owner.
  }

  private void saveInitRequestTable() {
    try {
      AgentServer.getTransaction().save(
        initRequestContextLists, INIT_REQUEST_TABLE);
    } catch (IOException exc) {
      throw new Error(exc.toString());
    }
  }

  private void saveWriteRequestTable() {
    try {
      AgentServer.getTransaction().save(
        writeRequestContextLists, WRITE_REQUEST_TABLE);
    } catch (IOException exc) {
      throw new Error(exc.toString());
    }
  }

  private void saveSyncRequestTable() {
    try {
      AgentServer.getTransaction().save(
        syncRequestContextLists, SYNC_REQUEST_TABLE);
    } catch (IOException exc) {
      throw new Error(exc.toString());
    }
  }

  private void saveServers() {
    try {
      AgentServer.getTransaction().save(servers, SERVER_LIST);
    } catch (IOException exc) {
      throw new Error(exc.toString());
    }
  }

  static class RequestContextList 
      implements java.io.Serializable {
    private Vector list;

    RequestContextList() {
      this.list = new Vector();
    }

    void put(RequestContext ctx) {
      list.addElement(ctx);
    }

    RequestContext get() {
      if( list.size() > 0) {
        return (RequestContext)list.elementAt(0);
      } else {
        return null;
      }
    }

    void pop() {
      if( list.size() > 0) {
        list.removeElementAt(0);
      }
    }

    int getSize() {
      return list.size();
    }

    public String toString() {
      return '(' + super.toString() + 
        ",list=" + list + ')';
    }
  }
}