streammessage.java

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

/*
 * JORAM: Java(TM) Open Reliable Asynchronous Messaging
 * Copyright (C) 2001 - ScalAgent Distributed Technologies
 * Copyright (C) 1996 - Dyade
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or any later version.
 * 
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 * 
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307
 * USA.
 *
 * Initial developer(s): Frederic Maistre (INRIA)
 * Contributor(s): Nicolas Tachker (Bull SA)
 */
package org.objectweb.joram.client.jms;

import java.io.*;

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

/**
 * Implements the <code>javax.jms.StreamMessage</code> interface.
 */
public class StreamMessage extends Message implements javax.jms.StreamMessage
{
  /** The array in which the written data is buffered. */
  private ByteArrayOutputStream outputBuffer = null;
  /** The stream in which body data is written. */
  private DataOutputStream outputStream = null;
  /** The stream for reading the data. */
  private DataInputStream inputStream = null;

  /** <code>true</code> if the message body is read-only. */
  private boolean RObody = false; 
  /** <code>true</code> if the message body is write-only. */
  private boolean WObody = true;

  /** Local bytes array. */
  private byte[] bytes = null;
  /** <code>true</code> if the message has been sent since its last modif. */
  private boolean prepared = false;

  private int available = 0;
  private boolean firstTimeBytesRead = true;

  private static final int SHORT = 1;
  private static final int CHAR = 2;
  private static final int INT = 3;
  private static final int LONG = 4;
  private static final int FLOAT = 5;
  private static final int DOUBLE = 6;
  private static final int BOOLEAN = 7;
  private static final int STRING = 8;
  private static final int BYTE = 9;
  private static final int BYTES = 10;
  private static final int NULL = 11;
  


  /**
   * Instanciates a bright new <code>StreamMessage</code>.
   *
   * @exception JMSException  In case of an error while creating the output
   *              stream.
   */
  StreamMessage() throws JMSException {
    super();
    outputBuffer = new ByteArrayOutputStream();
    outputStream = new DataOutputStream(outputBuffer);
    available = 0;
    firstTimeBytesRead = true;
  }

  /**
   * Instanciates a <code>StreamMessage</code> wrapping a consumed
   * MOM message containing a stream of bytes.
   *
   * @param sess  The consuming session.
   * @param momMsg  The MOM message to wrap.
   *
   * @exception JMSException  In case of an error while creating the input
   *              stream.
   */
  StreamMessage(Session sess, org.objectweb.joram.shared.messages.Message momMsg)
    throws JMSException {
    super(sess, momMsg);
    bytes = momMsg.getStream();
    
    try {
      inputStream = new DataInputStream(new ByteArrayInputStream(bytes));
    }
    catch (Exception exc) {
      JMSException jE =
        new JMSException("Error while creating the stream facility.");
      jE.setLinkedException(exc);
      throw jE;
    }
    RObody = true;
    WObody = false;
    available = 0;
    firstTimeBytesRead = true;
  }
  

  /** 
   * API method.
   *
   * @exception JMSException  In case of an error while closing the input or
   *              output streams.
   */
  public void clearBody() throws JMSException
  {
    super.clearBody();

    try {
      if (WObody) {
        outputStream.close();
        outputBuffer.close();
      }
      else
        inputStream.close();

      outputBuffer = new ByteArrayOutputStream();
      outputStream = new DataOutputStream(outputBuffer);
      bytes = null;
      RObody = false;
      WObody = true;
      prepared = false;
    }
    catch (IOException ioE) {
      JMSException jE = new JMSException("Error while closing the stream"
                                         + " facilities.");
      jE.setLinkedException(ioE);
      throw jE;
    }
  }


  /** 
   * API method.
   *
   * @exception MessageNotWriteableException  If the message body is read-only.
   * @exception JMSException  If the value could not be written on the stream.
   */
  public void writeBoolean(boolean value) 
    throws JMSException {
    prepareWrite();

    try {
      outputStream.writeByte(BOOLEAN);
      outputStream.writeBoolean(value);
    } catch (IOException ioE) {
      JMSException jE = new JMSException("Error while writing the value.");
      jE.setLinkedException(ioE);
      throw jE;
    }
  }
 
  /** 
   * API method.
   *
   * @exception MessageNotWriteableException  If the message body is read-only.
   * @exception JMSException  If the value could not be written on the stream.
   */ 
  public void writeByte(byte value) 
    throws JMSException {
    prepareWrite();

    try {
      outputStream.writeByte(BYTE);
      outputStream.writeByte((int) value);
    } catch (IOException ioE) {
      JMSException jE = new JMSException("Error while writing the value.");
      jE.setLinkedException(ioE);
      throw jE;
    }
  }
 
  /** 
   * API method.
   *
   * @exception MessageNotWriteableException  If the message body is read-only.
   * @exception JMSException  If the value could not be written on the stream.
   */   
  public void writeBytes(byte[] value) 
    throws JMSException {
    writeBytes(value, 0, value.length);
  }

  /** 
   * API method.
   *
   * @exception MessageNotWriteableException  If the message body is read-only.
   * @exception JMSException  If the value could not be written on the stream.
   */   
  public void writeBytes(byte[] value, int offset, int length)
    throws JMSException {
    prepareWrite();
    
    try {
      outputStream.writeByte(BYTES);
      if (value == null) {
        outputStream.writeInt(-1);
        return;
      } else
        outputStream.writeInt(length);
      outputStream.write(value, offset, length);
    } catch (IOException ioE) {
      JMSException jE = new JMSException("Error while writing the value.");
      jE.setLinkedException(ioE);
      throw jE;
    }
  }
 
  /** 
   * API method.
   *
   * @exception MessageNotWriteableException  If the message body is read-only.
   * @exception JMSException  If the value could not be written on the stream.
   */ 
  public void writeChar(char value) 
    throws JMSException {
    prepareWrite();

    try {
      outputStream.writeByte(CHAR);
      outputStream.writeChar((int) value);
    } catch (IOException ioE) {
      JMSException jE = new JMSException("Error while writing the value.");
      jE.setLinkedException(ioE);
      throw jE;
    }
  }
 
  /** 
   * API method.
   *
   * @exception MessageNotWriteableException  If the message body is read-only.
   * @exception JMSException  If the value could not be written on the stream.
   */ 
  public void writeDouble(double value) 
    throws JMSException {
    prepareWrite();

    try {
      outputStream.writeByte(DOUBLE);
      outputStream.writeDouble(value);
    } catch (IOException ioE) {
      JMSException jE = new JMSException("Error while writing the value.");
      jE.setLinkedException(ioE);
      throw jE;
    }
  }
 
  /** 
   * API method.
   *
   * @exception MessageNotWriteableException  If the message body is read-only.
   * @exception JMSException  If the value could not be written on the stream.
   */   
  public void writeFloat(float value) 
    throws JMSException {
    prepareWrite();

    try {
      outputStream.writeByte(FLOAT);
      outputStream.writeFloat(value);
    } catch (IOException ioE) {
      JMSException jE = new JMSException("Error while writing the value.");
      jE.setLinkedException(ioE);
      throw jE;
    }
  }
 
  /** 
   * API method.
   *
   * @exception MessageNotWriteableException  If the message body is read-only.
   * @exception JMSException  If the value could not be written on the stream.
   */  
  public void writeInt(int value) 
    throws JMSException {
    prepareWrite();

    try {
      outputStream.writeByte(INT);
      outputStream.writeInt(value);
    } catch (IOException ioE) {
      JMSException jE = new JMSException("Error while writing the value.");
      jE.setLinkedException(ioE);
      throw jE;
    }
  }
 
  /** 
   * API method.
   *
   * @exception MessageNotWriteableException  If the message body is read-only.
   * @exception JMSException  If the value could not be written on the stream.
   */ 
  public void writeLong(long value) 
    throws JMSException {
    prepareWrite();

    try {
      outputStream.writeByte(LONG);
      outputStream.writeLong(value);
    } catch (IOException ioE) {
      JMSException jE = new JMSException("Error while writing the value.");