streammessageimpl.java

实现了Jms的服务器源码,支持多种适配器,DB,FTP,支持多种数据库

            getOutputStream();
            _out.writeByte(DOUBLE);
            _out.writeDouble(value);
        } catch (IOException exception) {
            raise(exception);
        }
    }

    /**
     * Write a string to the stream message
     *
     * @param value the <code>String</code> value to be written
     * @throws JMSException if JMS fails to write message due to
     * some internal JMS error
     * @throws MessageNotWriteableException if message in read-only mode
     * @throws NullPointerException if value is null
     */
    public final void writeString(String value) throws JMSException {
        checkWrite();
        if (value == null) {
            // could throw IllegalArgumentException, but this is in keeping
            // with that thrown by DataOutputStream
            throw new NullPointerException("Argument value is null");
        }
        try {
            getOutputStream();
            _out.writeByte(STRING);
            _out.writeUTF(value);
        } catch (IOException exception) {
            raise(exception);
        }
    }

    /**
     * Write a byte array field to the stream message
     * <p>
     * The byte array <code>value</code> is written as a byte array field
     * into the StreamMessage. Consecutively written byte array fields are
     * treated as two distinct fields when reading byte array fields.
     *
     * @param value the byte array to be written
     * @throws JMSException if JMS fails to write message due to
     * some internal JMS error
     * @throws MessageNotWriteableException if message in read-only mode
     * @throws NullPointerException if value is null
     */
    public final void writeBytes(byte[] value) throws JMSException {
        checkWrite();
        if (value == null) {
            // could throw IllegalArgumentException, but this is in keeping
            // with that thrown by DataOutputStream
            throw new NullPointerException("Argument value is null");
        }
        try {
            getOutputStream();
            _out.writeByte(BYTE_ARRAY);
            _out.writeInt(value.length);
            _out.write(value);
        } catch (IOException exception) {
            raise(exception);
        }
    }

    /**
     * Write a portion of a byte array as a byte array field to the stream
     * message
     * <p>
     * The a portion of the byte array <code>value</code> is written as a
     * byte array field into the StreamMessage. Consecutively written byte
     * array fields are treated as two distinct fields when reading byte
     * array fields.
     *
     * @param value the byte array value to be written
     * @param offset the initial offset within the byte array
     * @param length the number of bytes to write
     * @throws JMSException if JMS fails to write message due to
     * some internal JMS error
     * @throws MessageNotWriteableException if message in read-only mode
     * @throws NullPointerException if value is null
     */
    public void writeBytes(byte[] value, int offset, int length)
        throws JMSException {
        checkWrite();
        if (value == null) {
            // could throw IllegalArgumentException, but this is in keeping
            // with that thrown by DataOutputStream
            throw new NullPointerException("Argument value is null");
        }
        try {
            getOutputStream();
            _out.writeByte(BYTE_ARRAY);
            _out.writeInt(length);
            _out.write(value, offset, length);
        } catch (IOException exception) {
            raise(exception);
        }
    }

    /**
     * Write a Java object to the stream message
     * <p>
     * Note that this method only works for the objectified primitive
     * object types (Integer, Double, Long ...), String's and byte arrays.
     *
     * @param value the Java object to be written
     * @throws JMSException if JMS fails to write message due to
     * some internal JMS error
     * @throws MessageFormatException if the object is invalid
     * @throws MessageNotWriteableException if message in read-only mode
     */
    public void writeObject(Object value) throws JMSException {
        if (value == null) {
            try {
                checkWrite();
                getOutputStream();
                _out.writeByte(NULL);
            } catch (IOException exception) {
                raise(exception);
            }
        } else if (value instanceof Boolean) {
            writeBoolean(((Boolean) value).booleanValue());
        } else if (value instanceof Byte) {
            writeByte(((Byte) value).byteValue());
        } else if (value instanceof byte[]) {
            writeBytes((byte[]) value);
        } else if (value instanceof Short) {
            writeShort(((Short) value).shortValue());
        } else if (value instanceof Character) {
            writeChar(((Character) value).charValue());
        } else if (value instanceof Integer) {
            writeInt(((Integer) value).intValue());
        } else if (value instanceof Long) {
            writeLong(((Long) value).longValue());
        } else if (value instanceof Float) {
            writeFloat(((Float) value).floatValue());
        } else if (value instanceof Double) {
            writeDouble(((Double) value).doubleValue());
        } else if (value instanceof String) {
            writeString((String) value);
        } else {
            throw new MessageFormatException(
                "Objects of type " + value.getClass().getName()
                + " are not supported by StreamMessage");
        }
    }

    /**
     * Put the message body in read-only mode, and reposition the stream
     * to the beginning
     *
     * @throws JMSException if JMS fails to reset the message due to
     * some internal JMS error
     */
    public void reset() throws JMSException {
        try {
            if (!_bodyReadOnly) {
                _bodyReadOnly = true;
                if (_out != null) {
                    _out.flush();
                    _bytes = _byteOut.toByteArray();
                    _byteOut = null;
                    _out.close();
                    _out = null;
                }
            } else {
                if (_in != null) {
                    _byteIn = null;
                    _in.close();
                    _in = null;
                }
            }
            _readBytes = 0;
        } catch (IOException exception) {
            raise(exception);
        }
    }

    /**
     * Overide the super class method to reset the streams, and put the
     * message body in write only mode
     *
     * @throws JMSException if JMS fails to reset the message due to
     * some internal JMS error.
     */
    public void clearBody() throws JMSException {
        try {
            if (_bodyReadOnly) {
                // in read-only mode
                _bodyReadOnly = false;
                if (_in != null) {
                    _byteIn = null;
                    _in.close();
                    _in = null;
                    _offset = 0;
                }
            } else if (_out != null) {
                // already in write-only mode
                _byteOut = null;
                _out.close();
                _out = null;
            }
            _bytes = EMPTY;
            _readBytes = 0;
        } catch (IOException exception) {
            raise(exception);
        }
    }

    /**
     * Set the read-only mode of the message. If read-only, resets the message
     * for reading
     *
     * @param readOnly if true, make the message body and properties
     * @throws JMSException if the read-only mode cannot be changed
     */
    public final void setReadOnly(boolean readOnly) throws JMSException {
        if (readOnly) {
            reset();
        }
        super.setReadOnly(readOnly);
    }

    /**
     * Prepare to do a read
     *
     * @throws JMSException if the current position in the stream can't be
     * marked
     * @throws MessageNotReadableException if the message is in write-only mode
     */
    private final void prepare() throws JMSException {
        checkRead();
        getInputStream();
        try {
            _in.mark(_bytes.length - _in.available());
        } catch (IOException exception) {
            raise(exception);
        }
    }

    /**
     * Reverts the stream to its prior position if a MessageFormatException is
     * thrown, and propagates the exception.
     *
     * @param exception the exception that caused the reset
     * @throws MessageFormatException
     */
    private void revert(MessageFormatException exception)
        throws MessageFormatException {
        try {
            _in.reset();
        } catch (IOException ignore) {
            // can't reset the stream, but need to propagate the original
            // exception
        }
        throw exception;
    }

    /**
     * Reverts the stream to its prior position if a NumberFormatException or
     * NullPointerException is thrown, and propagates the exception.
     *
     * @param exception the exception that caused the reset
     * @throws NullPointerException
     * @throws NumberFormatException
     */
    private void revert(RuntimeException exception) {
        try {
            _in.reset();
        } catch (IOException ignore) {
            // can't reset the stream, but need to propagate the original
            // exception
        }
        throw exception;
    }

    /**
     * Read the next object from the stream message
     *
     * @return a Java object from the stream message, in objectified
     * format (eg. if it set as an int, then a Integer is returned).
     * @throws JMSException if JMS fails to read message due to some internal
     * JMS error
     * @throws MessageEOFException if end of message stream
     * @throws MessageFormatException if a byte array has not been fully read
     * by {@link #readBytes(byte[])}
     * @throws MessageNotReadableException if the message is in write-only mode
     */
    private Object readNext() throws JMSException {
        if (_readBytes != 0) {
            throw new MessageFormatException(
                "Cannot read the next field until the byte array is read");
        }

        byte type = 0;
        try {
            type = _in.readByte();
        } catch (IOException exception) {
            raise(exception);
        }
        if ((type & 0x0F) > TYPE_NAMES.length) {
            throw new JMSException("StreamMessage corrupted");
        }
        Object result = null;

        try {
            switch (type & 0x0F) {
                case BOOLEAN:
                    boolean value = ((type & 0xF0) != 0) ? true : false;
                    result = new Boolean(value);
                    break;
                case BYTE:
                    result = new Byte(_in.readByte());
                    break;
                case BYTE_ARRAY:
                    int length = _in.readInt();
                    byte[] bytes = new byte[length];
                    _in.readFully(bytes);
                    result = bytes;
                    break;
                case SHORT:
                    result = new Short(_in.readShort());
                    break;
                case CHAR:
                    result = new Character(_in.readChar());
                    break;
                case INT:
                    result = new Integer(_in.readInt());
                    break;
                case LONG:
                    result = new Long(_in.readLong());
                    break;
                case FLOAT:
                    result = new Float(_in.readFloat());
                    break;
                case DOUBLE:
                    result = new Double(_in.readDouble());
                    break;
                case STRING:
                    result = _in.readUTF();
                    break;
            }
        } catch (IOException exception) {
            raise(exception);
        }

        return result;
    }

    /**
     * Initialise the input stream if it hasn't been intialised
     *
     * @return the input stream
     */
    private DataInputStream getInputStream() {
        if (_in == null) {
            _byteIn = new ByteArrayInputStream(_bytes, _offset,
                _bytes.length - _offset);
            _in = new DataInputStream(_byteIn);
        }
        return _in;
    }

    /**
     * Initialise the output stream if it hasn't been intialised
     *
     * @return the output stream
     * @throws IOException if the output stream can't be created
     */
    private final DataOutputStream getOutputStream() throws IOException {
        if (_out == null) {
            _byteOut = new ByteArrayOutputStream();
            _out = new DataOutputStream(_byteOut);
            _out.write(_bytes);
        }
        return _out;
    }

    /**
     * Helper to raise a JMSException when an I/O error occurs
     *
     * @param exception the exception that caused the failure
     * @throws JMSException
     */
    private final void raise(IOException exception) throws JMSException {
        JMSException error = null;
        if (exception instanceof EOFException) {
            error = new MessageEOFException(exception.getMessage());
        } else {
            error = new JMSException(exception.getMessage());
        }
        error.setLinkedException(exception);
        throw error;
    }

}