streammessage.java

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

    try {
      byte type = inputStream.readByte();
      if (type == FLOAT)
        return inputStream.readFloat();
      else if (type == STRING)
        return Float.valueOf(inputStream.readUTF()).floatValue();
      else
        throw new MessageFormatException("type read: " + type + " is not float or String.");
    } catch (EOFException e1) {
      MessageEOFException exc = new MessageEOFException("end of message " + e1);
      exc.setLinkedException(e1);
      throw exc;
    } catch (IOException e2) {
      JMSException exc = new JMSException("IOException");
      exc.setLinkedException(e2);
      throw exc;
    } 
  }

  /**
   * API method.
   *
   * @exception MessageNotReadableException  If the message body is write-only.
   * @exception MessageFormatException       If reading the expected type is
   *                                         not possible.
   * @exception MessageEOFException          Unexpected end of bytes array.
   * @exception JMSException                 internal error
   */
  public double readDouble() throws JMSException
  {
    if (WObody)
      throw new MessageNotReadableException("Can't read the message body as"
                                            + " it is write-only.");
    try {
      byte type = inputStream.readByte();
      if (type == DOUBLE)
        return inputStream.readDouble();
      else if (type == FLOAT)
        return inputStream.readFloat();
      else if (type == STRING)
        return Double.valueOf(inputStream.readUTF()).doubleValue();
      else
        throw new MessageFormatException("type read: " + type + " is not a double, a float or a String.");
    } catch (EOFException e1) {
      MessageEOFException exc = new MessageEOFException("end of message " + e1);
      exc.setLinkedException(e1);
      throw exc;
    } catch (IOException e2) {
      JMSException exc = new JMSException("IOException");
      exc.setLinkedException(e2);
      throw exc;
    } 
  }

  /**
   * API method.
   *
   * @exception MessageNotReadableException  If the message body is write-only.
   * @exception MessageFormatException       If reading the expected type is
   *                                         not possible.
   * @exception MessageEOFException          Unexpected end of bytes array.
   * @exception JMSException                 internal error
   */
  public int readBytes(byte[] bytes) throws JMSException {
    if (WObody)
      throw new MessageNotReadableException("Can't read the message body as"
                                            + " it is write-only.");
    if (bytes == null)
      return -1;
    if (bytes.length == 0)
      return 0;
    
    int ret = 0; 
    try {
      byte type;
      int counter = 0; 
      if (firstTimeBytesRead) {
        type = inputStream.readByte();

        if (type == BYTES) {
          available = inputStream.readInt();
          
          if (available == 0 || available == -1)
            return available;
          
          int toread = 0;
          if (bytes.length >= available)
            toread = available;
          else
            toread = bytes.length;
          for (int i = 0; i < toread; i ++) {
            bytes[i] = inputStream.readByte();
            counter++;
          }
          ret = counter;
          available = toread - counter;
          firstTimeBytesRead = false;
          counter = 0;
        } else
          throw new MessageFormatException("type read: " + type + " is not a byte[].");
      } else {
        if (available > 0)
          type = BYTES;
        else {
          inputStream.mark(inputStream.available());
          type = inputStream.readByte();
        }
        
        if (type == BYTES) {
          if (available >= 0)
            available = inputStream.readInt();

          if (available == 0 || available == -1)
            return available;

          int toread = 0;          
          if (bytes.length >= available)
            toread = available;
          else
            toread = bytes.length;
          for (int i = 0; i < toread; i ++) {
            bytes[i] = inputStream.readByte();
            counter++;
          }
          ret = counter;
          available = toread - counter;
          firstTimeBytesRead = false;
          counter = 0;
        } else {
          inputStream.reset();
          firstTimeBytesRead = true;
          return -1;
        }
      }
    } catch (EOFException e1) {
      MessageEOFException exc = new MessageEOFException("end of message " + e1);
      exc.setLinkedException(e1);
      throw exc;
    } catch (IOException e2) {
      JMSException exc = new JMSException("IOException");
      exc.setLinkedException(e2);
      throw exc;
    } 

    if (ret == 0)
      return -1;
    return ret;
  }

  /**
   * API method.
   *
   * @exception MessageNotReadableException  If the message body is write-only.
   * @exception MessageFormatException       If reading the expected type is
   *                                         not possible.
   * @exception MessageEOFException          Unexpected end of bytes array.
   * @exception JMSException                 internal error
   */
  public String readString() throws JMSException {
    if (WObody)
      throw new MessageNotReadableException("Can't read the message body as"
                                            + " it is write-only.");
    try {
      byte type = inputStream.readByte();
      if (type == STRING)
        return inputStream.readUTF();
      else if (type == INT)
        return String.valueOf(inputStream.readInt());
      else if (type == SHORT)
        return String.valueOf(inputStream.readShort());
      else if (type == BYTE)
        return String.valueOf(inputStream.readByte());
      else if (type == FLOAT)
        return String.valueOf(inputStream.readFloat());
      else if (type == LONG)
        return String.valueOf(inputStream.readLong());
      else if (type == DOUBLE)
        return String.valueOf(inputStream.readDouble());
      else if (type == BOOLEAN)
        return String.valueOf(inputStream.readBoolean());
      else if (type == CHAR)
        return String.valueOf(inputStream.readChar());
      else if (type == NULL)
        return null;
      else
        throw new MessageFormatException("type read: " + type + 
                                         " is not a int, a short, a byte,... or a String.");
    } catch (EOFException e1) {
      MessageEOFException exc = new MessageEOFException("end of message " + e1);
      exc.setLinkedException(e1);
      throw exc;
    } catch (IOException e2) {
      JMSException exc = new JMSException("IOException");
      exc.setLinkedException(e2);
      throw exc;
    } 
  }

  /**
   * API method.
   *
   * @exception MessageNotReadableException  If the message body is write-only.
   * @exception MessageFormatException       If reading the body is
   *                                         not possible.
   * @exception MessageEOFException          Unexpected end of bytes array.
   * @exception JMSException                 internal error
   */
  public Object readObject() throws JMSException {
    if (WObody)
      throw new MessageNotReadableException("Can't read the message body as"
                                            + " it is write-only.");
    try {
      byte type = inputStream.readByte();

      if (type == BOOLEAN) {
        return new Boolean(inputStream.readBoolean());
      } else if (type == CHAR) {
        return new Character(inputStream.readChar());
      } else if (type == BYTE) {
        return new Byte(inputStream.readByte());
      } else if (type == SHORT) { 
        return new Short(inputStream.readShort());
      } else if (type == INT) { 
        return new Integer(inputStream.readInt());
      } else if (type == LONG) { 
        return new Long(inputStream.readLong());
      } else if (type == FLOAT) { 
        return new Float(inputStream.readFloat());
      } else if (type == DOUBLE) { 
        return new Double(inputStream.readDouble());
      } else if (type == STRING) {
        return inputStream.readUTF();
      } else if (type == NULL) {
        return null;
      } else if (type == BYTES) {
        byte[] b = new byte[inputStream.available()];
        inputStream.read(b);
        return b;
      } else
        throw new MessageFormatException("not a primitive object.");
    } catch (EOFException e1) {
      MessageEOFException exc = new MessageEOFException("end of message " + e1);
      exc.setLinkedException(e1);
      throw exc;
    } catch (IOException e2) {
      JMSException exc = new JMSException("IOException");
      exc.setLinkedException(e2);
      throw exc;
    } 
  }

  
  /** 
   * API method.
   *
   * @exception JMSException  If an error occurs while closing the output
   *              stream.
   */
  public void reset() throws JMSException {
    try {
      if (WObody) {
        outputStream.flush();
        bytes = outputBuffer.toByteArray();
      } else
        inputStream.close();
      
      inputStream = new DataInputStream(new ByteArrayInputStream(bytes));
      
      if (inputStream != null)
        inputStream.reset();

      RObody = true;
      WObody = false;
      firstTimeBytesRead = true;
    } catch (IOException iE) {
      JMSException jE =
        new JMSException("Error while manipulating the stream facilities.");
      jE.setLinkedException(iE);
      throw jE;
    }
  }

  /**
   * Method actually preparing the message for sending by transfering the
   * local body into the wrapped MOM message.
   *
   * @exception Exception  If an error occurs while serializing.
   */
  protected void prepare() throws Exception
  {
    super.prepare();

    if (WObody) {
      outputStream.flush();
      bytes = outputBuffer.toByteArray();
      prepared = true;
    }

    momMsg.clearBody();
    momMsg.setStream(bytes);
  } 

  /**
   * Internal method called before each writing operation.
   *
   * @exception MessageNotWriteableException  If the message body is READ only.
   * @exception JMSException  If the stream could not be prepared for the
   *              writing operation.
   */
  private void prepareWrite() throws JMSException
  {
    if (RObody)
      throw new MessageNotWriteableException("Can't write a value as the"
                                             + " message body is read-only.");
    if (prepared) {
      prepared = false;
      outputBuffer = new ByteArrayOutputStream();
      outputStream = new DataOutputStream(outputBuffer);
    }
  }
}