msginfo.java

java开源的企业总线.xmlBlaster

/*------------------------------------------------------------------------------
 Name:      MsgInfo.java
 Project:   xmlBlaster.org
 Copyright: xmlBlaster.org, see xmlBlaster-LICENSE file
 Comment:   MsgInfo class for serialization/parsing of messages
 ------------------------------------------------------------------------------*/
package org.xmlBlaster.util.xbformat;

import java.util.logging.Logger;

import org.xmlBlaster.util.Global;
import org.xmlBlaster.util.Timestamp;
import org.xmlBlaster.util.XmlBlasterException;
import org.xmlBlaster.util.def.MethodName;
import org.xmlBlaster.util.def.ErrorCode;
import org.xmlBlaster.util.plugin.I_PluginConfig;
import org.xmlBlaster.util.protocol.email.AttachmentHolder;
import org.xmlBlaster.util.protocol.email.EmailExecutor;
import org.xmlBlaster.util.qos.StatusQosData;
import org.xmlBlaster.util.MsgUnitRaw;

import java.io.IOException;
import java.io.InputStream;
import java.io.ByteArrayInputStream;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Vector;

/**
 * Holds MsgUnits from socket or email protocol drivers with additional tranport
 * attributes. <br />
 * This class creates and parses raw byte[] messages which can be used to
 * transfer over a socket connection or as email attachment. <br />
 * MsgInfo instances may be reused, but are NOT reentrant (there are many
 * 'global' variables) <br />
 * 
 * @author [e-mail]
 * @see <a
 *      href="http://www.xmlBlaster.org/xmlBlaster/doc/requirements/protocol.socket.html">The
 *      protocol.socket requirement</a>
 */
public class MsgInfo {
   private static final String ME = "MsgInfo";

   private final Global glob;

   private static Logger log = Logger.getLogger(MsgInfo.class.getName());

   public static final byte INVOKE_BYTE = (byte) 73; // INVOKE_TYPE = "I";

   public static final byte RESPONSE_BYTE = (byte) 82; // RESPONSE_TYPE = "R";

   public static final byte EXCEPTION_BYTE = (byte) 69; // EXCEPTION_TYPE =
                                                         // "E";

   /** flag field number one */
   private boolean checksum;

   /** flag field 2 */
   private boolean compressed;

   /** flag field 3 */
   private byte type;

   /** flag field 4 */
   private byte byte4;

   /** flag field 5 */
   private byte byte5;

   /** flag field 6 */
   private int version;

   private String requestId;
   
   /** Remember if we got an explicit requestId or if we extracted it from the email-sentDate */
   protected boolean requestIdGuessed;

   private MethodName methodName;

   private String sessionId;

   private I_ProgressListener progressListener;

   /** Holding MsgUnitRaw objects which acts as a holder for the method arguments */
   private Vector msgVec;

   private I_MsgInfoParser msgInfoParser;
   
   private I_PluginConfig pluginConfig;

   /**
    * Transports information from receiver to response instance (Hack?)
    */
   private Map bounceObjects;

   /**
    * The same instance object may be reused. Ctor to parse messages using
    * I_MsgInfoParser implementations.
    */
   public MsgInfo(Global glob) {
      this(glob, INVOKE_BYTE, (String) null, (MethodName) null, (String) null,
            null);
   }

   /**
    * Ctor to parse messages with msgInfo.parse(iStream);
    */
   public MsgInfo(Global glob, I_ProgressListener progressListener) {
      this(glob, INVOKE_BYTE, (String) null, (MethodName) null, (String) null,
            progressListener);
   }

   /**
    * Create a raw message. msgInfo = new MsgInfo(glob, MsgInfo.INVOKE_BYTE,
    * MethodName.UPDATE, cbSessionId, progressListener);
    * msgInfo.addMessage(msgArr); byte[] rawMsg = msgInfo.createRawMsg();
    * 
    * @param glob
    * @param type
    * @param methodName
    * @param sessionId
    */
   public MsgInfo(Global glob, byte type, MethodName methodName,
         String sessionId) {
      this(glob, type, (String) null, methodName, sessionId, null);
   }

   public MsgInfo(Global glob, byte type, MethodName methodName,
         String sessionId, I_ProgressListener progressListener, String msgInfoParserClassName)
   throws XmlBlasterException {
      this(glob, type, methodName, sessionId, progressListener);
      if (msgInfoParserClassName != null)
         setMsgInfoParser(msgInfoParserClassName);
   }

   public MsgInfo(Global glob, byte type, MethodName methodName,
         String sessionId, I_ProgressListener progressListener) {
      this(glob, type, (String) null, methodName, sessionId, progressListener);
   }

   public MsgInfo(Global glob, byte type, String requestId,
         MethodName methodName, String sessionId,
         I_ProgressListener progressListener) {
      this.glob = glob;

      this.progressListener = progressListener;
      this.msgVec = new Vector();
      initialize();
      setType(type);
      setRequestId(requestId);
      setMethodName(methodName);
      setSecretSessionId(sessionId);
   }
   
   /**
    * Creates a new instance with the meta info from source.
    * 
    * @param type
    *           Please choose RESPONSE_BYTE or EXCEPTION_BYTE
    * @return The type is set to RESPONSE_BYTE, the content is empty
    */
   public MsgInfo createReturner(byte type) {
      MsgInfo returner = new MsgInfo(glob, type, getRequestId(),
            getMethodName(), getSecretSessionId(), getProgressListener());
            
      // returner.setBounceObjects(getBounceObjects());
      if (this.bounceObjects != null) {
         Map clone = (Map)((HashMap)this.bounceObjects).clone(); // Implicit expecting HashMap, fix me
         returner.setBounceObjects(clone);
         Object mailFrom = getBounceObject(EmailExecutor.BOUNCE_MAILFROM_KEY);
         Object mailTo = getBounceObject(EmailExecutor.BOUNCE_MAILTO_KEY);
         if (mailFrom != null && mailTo == null)
            returner.setBounceObject(EmailExecutor.BOUNCE_MAILTO_KEY, mailFrom);
         if (mailTo != null)
            returner.setBounceObject(EmailExecutor.BOUNCE_MAILFROM_KEY, mailTo);
      }
      
      returner.setRequestIdGuessed(isRequestIdGuessed());
      return returner;
   }

   /**
    * This method allows to reuse a MsgInfo instance.
    */
   public void initialize() {
      // msgLength = -1;
      checksum = false;
      compressed = false;
      type = INVOKE_BYTE; // request
      byte4 = 0;
      byte5 = 0;
      version = 1;
      requestId = null;
      methodName = null;
      sessionId = "";
      msgVec.clear();
   }

   /**
    * Guessing if attachment is compressed. TODO: Ask formatting plugins
    * 
    * @param fileName
    * @param mimeType
    * @return true if one of parameters ends with 'z'
    */
   public static boolean isCompressed(String fileName, String mimeType) {
      if (mimeType != null && mimeType.endsWith("z"))
         return true;
      if (fileName != null && fileName.endsWith("z"))
         return true;
      return false;
   }

   public int getNumMessages() {
      return msgVec.size();
   }

   /**
    * @param type
    *           The method type, e.g. MsgInfo.INVOKE_BYTE
    */
   public final void setType(byte type) {
      this.type = type;
   }

   /**
    * @return The method type, e.g. MsgInfo.INVOKE_BYTE
    */
   public final byte getType() {
      return this.type;
   }
   
   /**
    * Access the char representation to send over the wire. 
    * @param type
    * @return 'I', 'R' or 'E'
    */
   public static char getTypeChar(byte type) {
      if (type == INVOKE_BYTE)
         return 'I';
      else if (type == RESPONSE_BYTE)
         return 'R';
      else if (type == EXCEPTION_BYTE)
         return 'E';
      return 0;
   }

   /**
    * Similar to getType() but returns a nice human readable string for logging
    * output
    */
   public final String getTypeStr() {
      if (isInvoke())
         return "INVOKE";
      else if (isResponse())
         return "RESPONSE";
      else if (isException())
         return "EXCEPTION";
      return "UNKNOWN_TYPE";
   }

   /**
    * Similar to getType() but returns a nice human readable string for logging
    * output
    */
   public final static String getTypeStr(byte type) {
      if (type == INVOKE_BYTE)
         return "INVOKE";
      else if (type == RESPONSE_BYTE)
         return "RESPONSE";
      else if (type == EXCEPTION_BYTE)
         return "EXCEPTION";
      return "UNKNOWN_TYPE";
   }

   public final boolean isInvoke() {
      return (INVOKE_BYTE == type);
   }

   public final boolean isResponse() {
      return (RESPONSE_BYTE == type);
   }

   public final boolean isException() {
      return (EXCEPTION_BYTE == type);
   }

   /**
    * Set a unique ID (unique for this client), it will be bounced back with the
    * return value or with an exception occurred during this request. <br />
    * Note that you usually shouldn't set this value, this class generates a
    * unique requestId which you can access with getRequestId()
    */
   public final void setRequestId(String requestId) {
      this.requestId = requestId;
   }

   /**
    * Use this when sending a message.
    * <p>
    * Get a unique ID (unique for this client), it will be bounced back with the
    * return value or with an exception occurred during this request
    * </p>
    * 
    * @param prefix
    *           If desired you can specify a prefix for the request ID, e.g.
    *           "joe:"
    * @return An ID (unique in this JVM scope), e.g. "joe:3400" or "3400" if
    *         prefix is null
    */
   public final String createRequestId(String prefix) {
      if (this.requestId == null || this.requestId.length() < 1) {
         if (prefix == null)
            prefix = "";
         /* We need a unique timestamp over time
          * otherwise on client restart the timestamp
          * starts at 0 again and will be rejected by the
          * server (it thinks it is older than the last mail)
         synchronized (MsgInfo.class) {
            if (counter >= (Long.MAX_VALUE - 1L))
               counter = 0L;
            counter++;
            this.requestId = prefix + counter;
         }
         */
         Timestamp ts = new Timestamp();
         this.requestId = prefix + ts.getTimestamp();
      }
      return this.requestId;
   }

   /**
    * Use this when receiving a message.
    * 
    * @return The received request ID, is never null
    */
   public final String getRequestId() {
      if (this.requestId == null)
         return ""; // throw new IllegalArgumentException(ME + ": getRequestId
                     // returns null");
      return this.requestId;
   }

   /** For example MethodName.PUBLISH */
   public final void setMethodName(MethodName methodName) {
      this.methodName = methodName;
   }

   /**
    * For example MethodName.PUBLISH
    * 
    * @return Can be null
    */
   public final MethodName getMethodName() {
      return this.methodName;
   }

   /**
    * Access the method name as a string
    * 
    * @return Never null, for example "update"
    */
   public final String getMethodNameStr() {
      return (this.methodName == null) ? "" : this.methodName.toString();
   }

   /** The authentication sessionId */
   public final void setSecretSessionId(String sessionId) {
      this.sessionId = sessionId;
   }

   /**
    * The authentication sessionId
    * @return never null
    */
   public final String getSecretSessionId() {
      if (sessionId == null)
         return "";
      return this.sessionId;
   }

   /** Enable checksum? */
   public final void setChecksum(boolean checksum) {
      if (checksum == true) {
         log.warning("Checksum for raw socket message is not supported");
         return;
      }
      this.checksum = checksum;
   }

   /**
    * Compress message? NOTE: This compressed flag is set if the SOCKET header
    * is plain text and the MsgUnit[] is compressed. This mode is not
    * implemented, as we have "zlib:stream" compression which compresses the
    * whole socket input/output stream (there is no need to set this flag as it
    * is compressed as well).
    */
   public final void setCompressed(boolean compressed) {
      if (compressed == true) {
         log.warning("Compression for raw socket message is not supported");
         return;
      }