addressbase.java

java开源的企业总线.xmlBlaster

/*------------------------------------------------------------------------------
Name:      AddressBase.java
Project:   xmlBlaster.org
Copyright: xmlBlaster.org, see xmlBlaster-LICENSE file
Comment:   Holding connect address and callback address string including protocol
------------------------------------------------------------------------------*/
package org.xmlBlaster.util.qos.address;

import java.util.Iterator;
import java.util.Properties;
import java.util.logging.Level;
import java.util.logging.Logger;

import org.xml.sax.Attributes;
import org.xmlBlaster.contrib.ClientPropertiesInfo;
import org.xmlBlaster.util.EncodableData;
import org.xmlBlaster.util.Global;
import org.xmlBlaster.util.def.Constants;
import org.xmlBlaster.util.plugin.PluginManagerBase;
import org.xmlBlaster.util.property.PropBoolean;
import org.xmlBlaster.util.property.PropEntry;
import org.xmlBlaster.util.property.PropInt;
import org.xmlBlaster.util.property.PropLong;
import org.xmlBlaster.util.property.PropString;
import org.xmlBlaster.util.qos.ClientProperty;


/**
 * Abstract helper class holding connect address and callback address string
 * and protocol string.
 * <p />
 * See examples in the implementing classes
 * @see Address
 * @see CallbackAddress
 * @see org.xmlBlaster.test.classtest.qos.AddressBaseTest
 */
public abstract class AddressBase implements Cloneable
{
   protected final Global glob;
   private static Logger log = Logger.getLogger(AddressBase.class.getName());

   // Are strongest, usually set by programmer
   private /*I_Info*/ClientPropertiesInfo pluginAttributes;
   //private Hashtable pluginAttributes;
   private Properties pluginInfoParameters = new Properties();

   /** The root xml element: &lt;callback> or &lt;address>, is set from the derived class */
   protected String rootTag = null;

   protected String instanceName;

   protected final String className = "dispatch";
   protected final String context = null;
   /** For example "plugin/socket/" */
   protected String envPrefix = "";

   /** The unique address, e.g. the CORBA IOR string */
   private PropString rawAddress = new PropString("");

   private PropString bootstrapHostname = new PropString(""); // initially not "localhost" to ask bootstrap hostname

   public static final int DEFAULT_bootstrapPort = Constants.XMLBLASTER_PORT; // 3412
   private PropInt bootstrapPort = new PropInt(DEFAULT_bootstrapPort);

   /** The unique protocol type, defaults to SOCKET */
   public static final String DEFAULT_type = "SOCKET";
   protected PropString type = new PropString(DEFAULT_type);
   
   /** The protocol version, e.g. "1.0" */
   public static final String DEFAULT_version = "1.0";
   protected PropString version = new PropString(DEFAULT_version);
   
   /** BurstMode: The time to collect messages for publish/update */
   public static final long DEFAULT_collectTime = 0L;
   protected PropLong collectTime = new PropLong(DEFAULT_collectTime);
   
   /** Ping interval: pinging every given milliseconds */
   abstract public long getDefaultPingInterval();
   protected PropLong pingInterval = new PropLong(getDefaultPingInterval());
   
   /** How often to retry if connection fails */
   abstract public int getDefaultRetries();
   protected PropInt retries = new PropInt(getDefaultRetries());
   
   /** Delay between connection retries in milliseconds */
   abstract public long getDefaultDelay();
   protected PropLong delay = new PropLong(getDefaultDelay());
   
   /**
    * Shall the update() or publish() messages be send oneway (no application level ACK). 
    * <p />
    * For more info read the CORBA spec. Only CORBA and our native SOCKET protocol support oneway.
    * Defaults to false (the update() or publish() has a return value and can throw an exception).
    */
   public static final boolean DEFAULT_oneway = false;
   protected PropBoolean oneway = new  PropBoolean(DEFAULT_oneway);
   
   public static final boolean DEFAULT_dispatcherActive = true;
   /**
    * Control if the dispatcher is activated on login, i.e. if it is 
    * able to deliver asynchronous messages from the callback queue.
    * defaults to true
    */
   protected PropBoolean dispatcherActive = new PropBoolean(DEFAULT_dispatcherActive);

   /** Compress messages if set to "gzip" or "zip" */
   public static final String DEFAULT_compressType = "";
   protected PropString compressType = new PropString("compressType", DEFAULT_compressType);
   
   /** Messages bigger this size in bytes are compressed */
   public static final long DEFAULT_minSize = 0L;
   protected PropLong minSize = new PropLong("minSize", DEFAULT_minSize);

   public static final int DEFAULT_burstModeMaxEntries = 1;
   protected PropInt burstModeMaxEntries = new PropInt(DEFAULT_burstModeMaxEntries);

   public static final long DEFAULT_burstModeMaxBytes = -1L;
   protected PropLong burstModeMaxBytes = new PropLong(DEFAULT_burstModeMaxBytes);
   
   /** PtP messages wanted? Defaults to true, false prevents spamming */
   public static final boolean DEFAULT_ptpAllowed = true;
   protected PropBoolean ptpAllowed = new  PropBoolean(DEFAULT_ptpAllowed);
   
   /** The identifier sent to the callback client, the client can decide if he trusts this invocation */
   public static final String DEFAULT_sessionId = "unknown";
   protected PropString sessionId = new PropString(DEFAULT_sessionId);

   /** Shall this session callback be used for subjectQueue messages as well? For &lt;callback> only */
   public static final boolean DEFAULT_useForSubjectQueue = true;
   protected PropBoolean useForSubjectQueue = new  PropBoolean(DEFAULT_useForSubjectQueue);

   /**
    * Does client whish a dispatcher plugin. 
    * <p>
    * Set to "undef" forces to switch off, or e.g. "Priority,1.0" to access the PriorizedDispatchPlugin
    * </p>
    * <p>
    * Setting it to 'null' (which is the default) lets the server choose the plugin
    * </p>
    * @see <a href="http://www.xmlBlaster.org/xmlBlaster/doc/requirements/dispatch.control.plugin.html">The dispatch.control.plugin requirement</a>
    */
   public String DEFAULT_dispatchPlugin = PluginManagerBase.NO_PLUGIN_TYPE; // "undef";
   protected PropString dispatchPlugin = new PropString(DEFAULT_dispatchPlugin);

   /**
    * Setting this property will not throw an exception on pings when a ping response timeout is detected. Instead it
    * will just set the 'stalled' flag in the dispatch statistics.
    */
   public boolean DEFAULT_stallOnPingTimeout = false;
   protected PropBoolean stallOnPingTimeout = new PropBoolean(DEFAULT_stallOnPingTimeout);

   /** The node id to which we want to connect */
   protected String nodeId;

   /**
    */
   public AddressBase(Global glob, String rootTag) {
      this.glob = glob;

      setRootTag(rootTag);
   }

   /*
    * @throws IllegalArgumentException Not implemented. 
    
   public Object clone() {
      throw new IllegalArgumentException("AddressBase.clone() is not implemented");
   }
*/
   public AddressBase getClone() {
      try {
         AddressBase s = (AddressBase)super.clone();
          return s;
      } catch (CloneNotSupportedException e) {
          //This shouldn't happen because we implement Cloneable.
          throw new AssertionError();
      }
   }
   
   /**
    * Configure property settings. 
    * "-/node/heron/dispatch/connection/delay 20" has precedence over "-delay 10"
    */
   protected void initialize()
   {
      /* This is always set on server side from ServerAddress.java
         but not always on client side
         Shall we switch it on always here?
      if (this.nodeId == null) {
         this.nodeId = glob.getId();
      }
      */
      // SOCKET, IOR, XMLRPC, RMI, ...
      this.type.setFromEnv(this.glob, this.nodeId, context, className, this.instanceName, "protocol");

      // dispatch/callback/plugin/socket/hostname
      // dispatch/connection/plugin/ior/localPort
      this.envPrefix = "plugin/"+this.type.getValue().toLowerCase()+"/";
      if (log.isLoggable(Level.FINE)) log.fine("type=" + this.type.getValue() + " nodeId=" + this.nodeId + " context=" + context +
                         " className=" + className + " instanceName=" + this.instanceName + " envPrefix=" + this.envPrefix);

      // On server side for SOCKET protocol we support compression types:
      // Constants.COMPRESS_ZLIB_STREAM="zlib:stream" or "zlib" with minSize=1234 bytes
      // This default setting comes from environment or protocol plugin property
      // None stream compressions can be overwritten by CallbackAddress for each client individually
      // Here follows the plugin initialization, further down we overwrite this with Address specific settings
      // Example on server side: "-plugin/socket/compress/type stream"
      this.compressType = getEnv("compress/type", this.compressType.getValue());
      this.minSize = getEnv("compress/minSize", this.minSize.getValue());

      this.bootstrapHostname.setFromEnv(this.glob, this.nodeId, context, className, this.instanceName, "bootstrapHostname");
      this.bootstrapPort.setFromEnv(this.glob, this.nodeId, context, className, this.instanceName, "bootstrapPort");

      //this.bootstrapHostname.setFromEnv(this.glob, this.nodeId, context, className, this.instanceName, envPrefix+"bootstrapHostname");
      //this.bootstrapPort.setFromEnv(this.glob, this.nodeId, context, className, this.instanceName, envPrefix+"bootstrapPort");
      //log.error(ME, "DEBUG ONLY: Checking " + this.instanceName + ": " + envPrefix+"port to result=" + this.bootstrapPort.getValue() );

      // These are protocol unspecific values
      this.burstModeMaxEntries.setFromEnv(this.glob, this.nodeId, context, className, this.instanceName, "burstMode/maxEntries");
      this.burstModeMaxBytes.setFromEnv(this.glob, this.nodeId, context, className, this.instanceName, "burstMode/maxBytes");
      this.collectTime.setFromEnv(this.glob, this.nodeId, context, className, this.instanceName, "burstMode/collectTime");
      this.pingInterval.setFromEnv(this.glob, this.nodeId, context, className, this.instanceName, "pingInterval");
      this.retries.setFromEnv(this.glob, this.nodeId, context, className, this.instanceName, "retries");
      this.delay.setFromEnv(this.glob, this.nodeId, context, className, this.instanceName, "delay");
      this.oneway.setFromEnv(this.glob, this.nodeId, context, className, this.instanceName, "oneway");
      this.dispatcherActive.setFromEnv(this.glob, this.nodeId, context, className, this.instanceName, "dispatcherActive");
      this.compressType.setFromEnv(this.glob, this.nodeId, context, className, this.instanceName, "compress/type");
      this.minSize.setFromEnv(this.glob, this.nodeId, context, className, this.instanceName, "compress/minSize");
      this.ptpAllowed.setFromEnv(this.glob, this.nodeId, context, className, this.instanceName, "ptpAllowed");
      this.sessionId.setFromEnv(this.glob, this.nodeId, context, className, this.instanceName, "sessionId");
      this.dispatchPlugin.setFromEnv(this.glob, this.nodeId, context, className, this.instanceName, "DispatchPlugin/defaultPlugin");
      this.stallOnPingTimeout.setFromEnv(this.glob, this.nodeId, context, className, this.instanceName, "stallOnPingTimeout");
      //log.error(ME, getType() + " " + "DEBUG ONLY " + this.compressType + " " + this.minSize + toXml());
   }

   /**
    * Set a protocol specific property. 
    * <p>
    * Setting a property here forces the setting in the plugin, it
    * has precedence over any environment, xmlBlaster.properties or command line setting.
    * <br />
    * You typically use this method in your client code to overwrite settings,.
    * please check the protocol specific documentation about the supported settings.
    * </p>
    * @param key    The property, e.g. "SOLingerTimeout" (WITHOUT any prefix like "plugin/socket/")
    *               The searched property is depending on the type (here "socket")
    *               and instance (here "connection") e.g. "plugin/socket/SOLingerTimeout"
    *               and with higher precedence "dispatch/connection/plugin/socket/SOLingerTimeout"
    * @param value  The value, e.g. "10000"
    */
   public void setPluginProperty(String key, String value) {
      if (this.pluginAttributes == null) this.pluginAttributes = new ClientPropertiesInfo(null);
      this.pluginAttributes.put(key, value);
      // refresh compressType or minSize: Those attributes are double used:
      // Once SOCKET specific and again as a common setting in <address ...>
      // TODO: clean up this mess: no SOCKET specific code in here!
      if ("compress/type".equals(key) || "compress/minSize".equals(key)) {
         initialize();
      }
   }
   
   public void addClientProperty(ClientProperty clientProperty) {
      if (this.pluginAttributes == null) this.pluginAttributes = new ClientPropertiesInfo(null);
      this.pluginAttributes.put(clientProperty.getName(), clientProperty);
   }

   /**
    * Set the PluginInfo parameters (derived from xmlBlasterPlugins.xml or xmlBlaster.properties). 
    * <br />
    * As a protocol plugin developer you should call this method if you have a PluginInfo instance
    * to use the default paramaters of the plugin.
    * <br />
    * Example from xmlBlasterPlugins.xml:
    * <br />
    *  &lt;plugin id='SOCKET_UDP' className='org.xmlBlaster.protocol.socket.SocketDriver'>
    *     ...
    *     &lt;attribute id='useUdpForOneway'>true</attribute>
    *  &lt;/plugin>
    * <p/>
    * These settings are used as default settings for the plugin with lowest priority
    * <p/>
    * Calls initialize() to reinitialize compression.
    */
   public void setPluginInfoParameters(Properties parameters) {
      if (parameters == null) {
         this.pluginInfoParameters = new Properties();
      }
      else {
         this.pluginInfoParameters = parameters;
      }
      initialize();
   }

   public String getEnvPrefix() {
      return this.envPrefix;