xbbytesmessage.java

java开源的企业总线.xmlBlaster

/*------------------------------------------------------------------------------
 Name:      XBBytesMessage.java
 Project:   xmlBlaster.org
 Copyright: xmlBlaster.org, see xmlBlaster-LICENSE file
 ------------------------------------------------------------------------------*/
package org.xmlBlaster.jms;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.EOFException;
import java.io.IOException;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;

import javax.jms.BytesMessage;
import javax.jms.JMSException;
import javax.jms.MessageEOFException;
import javax.jms.MessageNotReadableException;
import javax.jms.MessageNotWriteableException;

import org.xmlBlaster.util.def.ErrorCode;

/**
 * XBBytesMessage
 * 
 * @author <a href="mailto:[e-mail]">Michele Laghi</a>
 * 
 */
public class XBBytesMessage extends XBMessage implements BytesMessage {
   private final static String ME = "XBBytesMessage";

   private DataInputStream dataIs;

   private DataOutputStream dataOs;

   private ByteArrayOutputStream baos;

   private ByteArrayInputStream bais;

   private boolean writeOnly;

   /**
    * true if the stream has reached its end. This is needed to return -1 when readBytes
    * is invoked on a passed stream.
    */
   private boolean streamEnded;
   
   public XBBytesMessage(XBSession session, byte[] content) throws JMSException {
      this(session, content, XBMessage.BYTES);
   }
   
   /**
    * If the content is empty it will be considered a message for a producer,
    * i.e. a message to be filled and sent. If the content is not null, then it
    * is assumed to be a message for a consumer
    * 
    * @param global
    * @param key
    * @param content
    * @param qos
    */
   XBBytesMessage(XBSession session, byte[] content, int type) throws JMSException {
      super(session, content, type);
      makeWriteable();
   }

   private void makeWriteable() throws JMSException {
      this.writeOnly = true;
      this.streamEnded = false;
      this.baos = new ByteArrayOutputStream(this.content == null ? 10000
            : this.content.length + 10000);
      try {
         this.dataOs = new DataOutputStream(baos);
         // this.objOs = new ObjectOutputStream(baos);
         if (this.content != null && content.length > 0)
            baos.write(content);
         this.dataIs = null;
         this.bais = null;
      } 
      catch (IOException ex) {
         throw new XBException(ex, "when creating a message");
      }
   }
   
   
   private void getterCheck(String methodName)
         throws MessageNotReadableException {
      if (this.writeOnly) {
         throw new MessageNotReadableException(ME
               + " writeonly message: not allowed to read on operation '"
               + methodName + "'");
      }
   }

   private void setterCheck(String methodName)
         throws MessageNotWriteableException {
      if (this.readOnly) {
         throw new MessageNotWriteableException("could not invoke '"
               + methodName + "' since the message is in readonly mode");
      }
   }

   public long getBodyLength() throws JMSException {
      getterCheck("getBodyLength");
      if (this.content == null) {
         if (this.dataOs != null)
            return this.dataOs.size();
         else
            return 0L;
      }
      return this.content.length;
   }

   private final void resetDataStream() throws JMSException {
      try {
         this.dataIs.reset();
      }
      catch (IOException ex) {
         throw new MessageEOFException(ME + ".resetDataStream: " + ex.getMessage(),
               ErrorCode.RESOURCE_UNAVAILABLE.getErrorCode());
      }
      
   }
   
   public boolean readBoolean() throws JMSException {
      getterCheck("readBoolean");
      try {
         this.dataIs.mark(10);
         return this.dataIs.readBoolean();
      }
      catch (EOFException ex) {
         throw new MessageEOFException(ME + ".readBoolean: " + ex.getMessage(),
               ErrorCode.USER_MESSAGE_INVALID.getErrorCode());
      }
      catch (IOException ex) {
         resetDataStream();
         throw new JMSException(ME + ".readBoolean: " + ex.getMessage(),
               ErrorCode.USER_ILLEGALARGUMENT.getErrorCode());
      }
   }

   public byte readByte() throws JMSException {
      getterCheck("readByte");
      try {
         this.dataIs.mark(10);
         return this.dataIs.readByte();
      }
      catch (EOFException ex) {
         throw new MessageEOFException(ME + ".readByte: " + ex.getMessage(),
               ErrorCode.USER_MESSAGE_INVALID.getErrorCode());
      }
      catch (IOException ex) {
         resetDataStream();
         throw new JMSException(ME + ".readByte: " + ex.getMessage(),
               ErrorCode.USER_ILLEGALARGUMENT.getErrorCode());
      }
   }

   public int readUnsignedByte() throws JMSException {
      getterCheck("readUnsignedByte");
      try {
         this.dataIs.mark(10);
         return this.dataIs.readUnsignedByte();
      }
      catch (EOFException ex) {
         throw new MessageEOFException(ME + ".readUnsignedByte: " + ex.getMessage(),
               ErrorCode.USER_MESSAGE_INVALID.getErrorCode());
      }
      catch (IOException ex) {
         resetDataStream();
         throw new JMSException(ME + ".readUnsignedByte: " + ex.getMessage(),
               ErrorCode.USER_ILLEGALARGUMENT.getErrorCode());
      }
   }

   public short readShort() throws JMSException {
      getterCheck("readShort");
      try {
         this.dataIs.mark(10);
         return this.dataIs.readShort();
      } 
      catch (EOFException ex) {
         throw new MessageEOFException(ME + ".readShort: " + ex.getMessage(),
               ErrorCode.USER_MESSAGE_INVALID.getErrorCode());
      }
      catch (IOException ex) {
         resetDataStream();
         throw new JMSException(ME + ".readShort: " + ex.getMessage(),
               ErrorCode.USER_ILLEGALARGUMENT.getErrorCode());
      }
   }

   public int readUnsignedShort() throws JMSException {
      getterCheck("readUnsignedShort");
      try {
         this.dataIs.mark(10);
         return this.dataIs.readUnsignedShort();
      }
      catch (EOFException ex) {
         throw new MessageEOFException(ME + ".readUnsignedShort: " + ex.getMessage(),
               ErrorCode.USER_MESSAGE_INVALID.getErrorCode());
      }
      catch (IOException ex) {
         resetDataStream();
         throw new JMSException(ME + ".readUnsignedShort: " + ex.getMessage(),
               ErrorCode.USER_ILLEGALARGUMENT.getErrorCode());
      }
   }

   public char readChar() throws JMSException {
      getterCheck("readChar");
      try {
         this.dataIs.mark(10);
         return this.dataIs.readChar();
      }
      catch (EOFException ex) {
         throw new MessageEOFException(ME + ".readChar: " + ex.getMessage(),
               ErrorCode.USER_MESSAGE_INVALID.getErrorCode());
      }
      catch (IOException ex) {
         resetDataStream();
         throw new JMSException(ME + ".readChar: " + ex.getMessage(),
               ErrorCode.USER_ILLEGALARGUMENT.getErrorCode());
      }
   }

   public int readInt() throws JMSException {
      getterCheck("readInt");
      try {
         this.dataIs.mark(10);
         return this.dataIs.readInt();
      }
      catch (EOFException ex) {
         throw new MessageEOFException(ME + ".readInt: " + ex.getMessage(),
               ErrorCode.USER_MESSAGE_INVALID.getErrorCode());
      }
      catch (IOException ex) {
         resetDataStream();
         throw new JMSException(ME + ".readInt: " + ex.getMessage(),
               ErrorCode.USER_ILLEGALARGUMENT.getErrorCode());
      }
   }

   public long readLong() throws JMSException {
      getterCheck("readLong");
      try {
         this.dataIs.mark(10);
         return this.dataIs.readLong();
      }
      catch (EOFException ex) {
         throw new MessageEOFException(ME + ".readLong: " + ex.getMessage(),
               ErrorCode.USER_MESSAGE_INVALID.getErrorCode());
      }
      catch (IOException ex) {
         resetDataStream();
         throw new JMSException(ME + ".readLong: " + ex.getMessage(),
               ErrorCode.USER_ILLEGALARGUMENT.getErrorCode());
      }
   }

   public float readFloat() throws JMSException {
      getterCheck("readFloat");
      try {
         this.dataIs.mark(10);
         return this.dataIs.readFloat();
      }
      catch (EOFException ex) {
         throw new MessageEOFException(ME + ".readFloat: " + ex.getMessage(),
               ErrorCode.USER_MESSAGE_INVALID.getErrorCode());
      }
      catch (IOException ex) {
         resetDataStream();
         throw new JMSException(ME + ".readFloat: " + ex.getMessage(),
               ErrorCode.USER_ILLEGALARGUMENT.getErrorCode());