serverscope.java

java开源的企业总线.xmlBlaster

/*------------------------------------------------------------------------------
Name:      Global.java
Project:   xmlBlaster.org
Copyright: xmlBlaster.org, see xmlBlaster-LICENSE file
Comment:   Handling global data on server side
------------------------------------------------------------------------------*/
package org.xmlBlaster.engine;

import org.xmlBlaster.util.IsoDateParser;
import org.xmlBlaster.util.Timeout;
import org.xmlBlaster.util.ThreadLister;
import org.xmlBlaster.util.XmlBlasterException;
import org.xmlBlaster.util.context.ContextNode;
import org.xmlBlaster.util.cluster.NodeId;
import org.xmlBlaster.util.def.Constants;
import org.xmlBlaster.util.def.ErrorCode;
import org.xmlBlaster.authentication.SessionInfo;
import org.xmlBlaster.engine.xml2java.XmlKey;
import org.xmlBlaster.engine.cluster.ClusterManager;
import org.xmlBlaster.engine.admin.CommandManager;
import org.xmlBlaster.engine.admin.extern.MomClientGateway;
import org.xmlBlaster.protocol.CbProtocolManager;
import org.xmlBlaster.protocol.I_Authenticate;
import org.xmlBlaster.util.dispatch.DispatchManager;
import org.xmlBlaster.util.dispatch.DispatchConnectionsHandler;
import org.xmlBlaster.engine.dispatch.CbDispatchConnectionsHandler;
import org.xmlBlaster.engine.distributor.plugins.MsgDistributorPluginManager;
import org.xmlBlaster.util.property.Property;
import org.xmlBlaster.util.queue.I_EntryFactory;
import org.xmlBlaster.engine.queuemsg.ServerEntryFactory;
import org.xmlBlaster.engine.msgstore.StoragePluginManager;
import org.xmlBlaster.engine.runlevel.I_RunlevelListener;
import org.xmlBlaster.engine.runlevel.RunlevelManager;
import org.xmlBlaster.engine.runlevel.PluginHolderSaxFactory;
import org.xmlBlaster.engine.runlevel.PluginHolder;
import org.xmlBlaster.util.queue.I_Queue;
import org.xmlBlaster.engine.queuemsg.ReferenceEntry;
import org.xmlBlaster.engine.MsgUnitWrapper;
import org.xmlBlaster.engine.persistence.MsgFileDumper;

import java.util.*;
import java.util.logging.Level;
import java.util.logging.Logger;


/**
 * This holds global needed data of one xmlBlaster instance.
 * <p>
 * @author <a href="mailto:[e-mail]">Marcel Ruff</a>
 */
public final class ServerScope extends org.xmlBlaster.util.Global implements I_RunlevelListener
{
   private static Logger log = Logger.getLogger(ServerScope.class.getName());

   private volatile RunlevelManager runlevelManager;
   private int currRunlevel = 0;

   /** the authentication service (a layer around it for security reasons) */
   private I_Authenticate authenticate;
   /** the xmlBlaster core class */
   private RequestBroker requestBroker;
   private NodeId nodeId;
   /** Unique id, even for each restart of a node */
   private volatile String instanceId;
   private volatile ClusterManager clusterManager;
   private volatile Timeout sessionTimer;
   private volatile Timeout topicTimer;
   private volatile Timeout telnetSessionTimer;

   private boolean useCluster;
   private boolean firstUseCluster = true; // to allow caching

   private volatile CbProtocolManager cbProtocolManager;

   private volatile StoragePluginManager topicStorePluginManager;

   private volatile CommandManager commandManager;
   private boolean useAdminManager = true;
   private boolean firstUseAdminManager = true; // to allow caching
   private MomClientGateway momClientGateway = null;

   private volatile MsgFileDumper msgFileDumper;

   private PluginHolder pluginHolder;

   private volatile MsgDistributorPluginManager msgDistributorPluginManager;

   private SubjectEntryShuffler subjectEntryShuffler;

   private SessionInfo internalSessionInfo;

   private TopicAccessor topicAccessor;

   public void finalize() {
      if (log.isLoggable(Level.FINE)) log.fine("Entering finalize");
      shutdown();
   }

   public void shutdown() {
      super.shutdown();
      if (log.isLoggable(Level.FINE)) log.fine("Destroying engine.Global handle");

      if (sessionTimer != null) {
         sessionTimer.shutdown();
         sessionTimer = null;
      }
      if (topicTimer != null) {
         topicTimer.shutdown();
         topicTimer = null;
      }
      removeTelnetSessionTimer();
   }

    public ServerScope() {
       this(null, true);
    }

   /**
    * One instance of this represents one xmlBlaster server.
    * @param args Environment arguments (key/value pairs)
    */
   public ServerScope(String[] args) {
      super(args);
      init(args);
      addObjectEntry(Constants.OBJECT_ENTRY_ServerScope, this); // registers itself in util.Global "ServerNodeScope"
   }

   public ServerScope(Properties p, boolean loadPropFile) {
      super(Property.propsToArgs(p), loadPropFile, false);
      initThis();
      addObjectEntry(Constants.OBJECT_ENTRY_ServerScope, this); // registers itself in util.Global
      // The util.Global base class can't initiliaze it, as this class is initialized later and overwrites with null
   }

   /**
    * Calls super.init and checks the environment for "cluster.node.id"
    * <p />
    * See private org.xmlBlaster.Main#createNodeId()
    * @return 1 Show usage, 0 OK, -1 error
    */
   public int init(String[] args) {
      int ret = super.init(args);
      initThis();
      return ret;
   }

   private void initThis() {
      String myId = getProperty().get("cluster.node.id", (String)null);
      if (myId == null && getBootstrapAddress().getRawAddress() != null && getBootstrapAddress().getRawAddress().length() > 0) {
         myId = getStrippedString(getBootstrapAddress().getRawAddress());
      }
      if (myId == null && getBootstrapAddress().getBootstrapPort() > 0) {
         myId = getStrippedString(getBootstrapAddress().getBootstrapHostname() + ":" + getBootstrapAddress().getBootstrapPort());
      }
      if (myId == null) {
         if (useCluster()) {
            log.severe("Can't determine server instance name, try to add '-cluster.node.id <aUniqueId>' on startup.");
            System.exit(1);
         }
         else {
            myId = "xmlBlaster";  // fallback
         }
      }
      if (myId != null) {
         setId(myId);
      }
      getRunlevelManager().addRunlevelListener(this);
      log.info("Setting xmlBlaster instance name (-cluster.node.id) to '" + getId() + "'");
   }

   /**
    * Check where we are, on client or on server side?
    * util.Global returns false
    * @return true As we are engine.Global and running server side
    */
   public boolean isServerSide() {
      return true;
   }

   /**
    * The unique name of this xmlBlaster server instance.
    * @return Can be null during startup
    */
   public final ContextNode getContextNode() {
      return super.contextNode;
   }

   /**
    * Unique id of the xmlBlaster server, changes on each restart.
    * If 'node/heron' is restarted, the instanceId changes.
    * @return nodeId + timestamp, '/node/heron/instanceId/33470080380'
    */
   public String getInstanceId() {
      if (this.instanceId == null) {
         synchronized(this) {
            if (this.instanceId == null) {
               // TODO: Two mirrored /node/heron: add IP:port to instanceId?
               ContextNode node = new ContextNode("instanceId", ""+System.currentTimeMillis(),
                                      getContextNode());
               this.instanceId = node.getAbsoluteName();
               //this.instanceId = getLogPrefix() + "/instanceId/" + System.currentTimeMillis();
            }
         }
      }
      return this.instanceId;
   }

   /**
    * The unique name of this xmlBlaster server instance.
    * @return Can be null during startup
    */
   public final NodeId getNodeId() {
      return this.nodeId;
   }

   /**
    * Access the unique cluster node id (as a String).
    * @return The name of this xmlBlaster instance, e.g. "heron.mycompany.com"
    *         or "http://mycomp:3412"
    *         Can be null during startup
    */
   public final String getId() {
      if (getNodeId() == null) return null;
      return getNodeId().getId();
   }

   public final void setId(String id) {
      super.setId(id);
      if (id == null) return;
      this.nodeId = new NodeId(id); // ContextNode should replace NodeId one day
      this.contextNode = new ContextNode(ContextNode.CLUSTER_MARKER_TAG, getStrippedId(), (ContextNode)null);
   }

   /**
    * Initialize runlevel manager used to start/stop xmlBlaster with different run levels.
    */
   public final RunlevelManager getRunlevelManager() {
      if (this.runlevelManager == null) {
         boolean initJmx = false;
         synchronized(this) {
            if (this.runlevelManager == null) {
               this.runlevelManager = new RunlevelManager(this);
               initJmx = true;
            }
         }
         if (initJmx) {
            /* To avoid dead lock do outside of sync:
             * "RMI TCP Connection(1)-127.0.0.2" daemon prio=1 tid=0x08602b10 nid=0x1ae1 waiting for monitor entry [0xa73b8000..0xa73b9040]
        at org.xmlBlaster.util.Global.initLog(Global.java:519)
        - waiting to lock <0xa8f06f40> (a org.xmlBlaster.engine.Global)
        at org.xmlBlaster.util.Global.addLogger(Global.java:624)
        at org.xmlBlaster.util.Global.getLog(Global.java:646)
        at org.xmlBlaster.util.log.XmlBlasterJdk14LoggingHandler.publish(XmlBlasterJdk14LoggingHandler.java:86)
        at java.util.logging.Logger.log(Logger.java:428)
        at java.util.logging.Logger.doLog(Logger.java:450)
        at java.util.logging.Logger.logp(Logger.java:566)
        at sun.rmi.runtime.Log$LoggerLog.log(Log.java:212)
        at sun.rmi.server.UnicastServerRef.logCall(UnicastServerRef.java:424)
        at sun.rmi.server.UnicastServerRef.oldDispatch(UnicastServerRef.java:372)
        at sun.rmi.server.UnicastServerRef.dispatch(UnicastServerRef.java:240)
        at sun.rmi.transport.Transport$1.run(Transport.java:153)
        at java.security.AccessController.doPrivileged(Native Method)
        at sun.rmi.transport.Transport.serviceCall(Transport.java:149)
        at sun.rmi.transport.tcp.TCPTransport.handleMessages(TCPTransport.java:460)
        at sun.rmi.transport.tcp.TCPTransport$ConnectionHandler.run(TCPTransport.java:701)
        at java.lang.Thread.run(Thread.java:595)

"main" prio=1 tid=0x0805f588 nid=0x1ad2 runnable [0xbfe20000..0xbfe20998]
        at java.net.SocketInputStream.socketRead0(Native Method)
        at java.net.SocketInputStream.read(SocketInputStream.java:129)
        at java.io.BufferedInputStream.fill(BufferedInputStream.java:218)
        at java.io.BufferedInputStream.read(BufferedInputStream.java:235)
        - locked <0xa8a5f1e8> (a java.io.BufferedInputStream)
        at java.io.DataInputStream.readByte(DataInputStream.java:241)
        at sun.rmi.transport.StreamRemoteCall.executeCall(StreamRemoteCall.java:189)
        at sun.rmi.server.UnicastRef.invoke(UnicastRef.java:343)
        at sun.rmi.registry.RegistryImpl_Stub.unbind(Unknown Source)
        at java.rmi.Naming.unbind(Naming.java:135)
        at org.xmlBlaster.util.admin.extern.JmxWrapper.init(JmxWrapper.java:325)
        - locked <0xa8f40a18> (a org.xmlBlaster.util.admin.extern.JmxWrapper)
        at org.xmlBlaster.util.admin.extern.JmxWrapper.<init>(JmxWrapper.java:133)
        at org.xmlBlaster.util.admin.extern.JmxWrapper.getInstance(JmxWrapper.java:117)
        - locked <0xacc982b8> (a java.lang.Class)
        at org.xmlBlaster.util.Global.getJmxWrapper(Global.java:327)
        at org.xmlBlaster.util.Global.registerMBean(Global.java:375)
        at org.xmlBlaster.engine.runlevel.RunlevelManager.<init>(RunlevelManager.java:81)
        at org.xmlBlaster.engine.Global.getRunlevelManager(Global.java:230)
        - locked <0xa8f06f40> (a org.xmlBlaster.engine.Global)
        at org.xmlBlaster.engine.Global.initThis(Global.java:156)
        at org.xmlBlaster.engine.Global.init(Global.java:132)
        at org.xmlBlaster.engine.Global.<init>(Global.java:113)
        at org.xmlBlaster.Main.<init>(Main.java:108)
        at org.xmlBlaster.Main.main(Main.java:524)

             */
            this.runlevelManager.initJmx();
         }
      }
      return this.runlevelManager;
   }

   public int getRunlevel() {
      return this.currRunlevel;
   }

   public void setUseCluster(boolean useCluster) {