[hornetq-commits] JBoss hornetq SVN: r10008 - in trunk: src/main/org/hornetq/core/message/impl and 14 other directories.

Author: clebert.suconic(a)jboss.com Date: 2010-12-07 17:22:35 -0500 (Tue, 07 Dec 2010) New Revision: 10008 Added: trunk/src/main/org/hornetq/core/client/impl/DecompressedLargeMessageControllerImpl.java trunk/src/main/org/hornetq/core/client/impl/LargeMessageController.java trunk/src/main/org/hornetq/core/client/impl/LargeMessageControllerImpl.java Removed: trunk/src/main/org/hornetq/core/client/impl/LargeMessageBufferImpl.java trunk/src/main/org/hornetq/core/client/impl/LargeMessageBufferInternal.java trunk/src/main/org/hornetq/utils/DecompressedLargeMessageBuffer.java Modified: trunk/src/main/org/hornetq/core/client/impl/ClientConsumerImpl.java trunk/src/main/org/hornetq/core/client/impl/ClientMessageImpl.java trunk/src/main/org/hornetq/core/client/impl/ClientProducerImpl.java trunk/src/main/org/hornetq/core/message/impl/MessageImpl.java trunk/src/main/org/hornetq/core/message/impl/MessageInternal.java trunk/src/main/org/hornetq/core/persistence/StorageManager.java trunk/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java trunk/src/main/org/hornetq/core/persistence/impl/nullpm/NullStorageManager.java trunk/src/main/org/hornetq/core/protocol/core/ServerSessionPacketHandler.java trunk/src/main/org/hornetq/core/protocol/core/impl/CoreSessionCallback.java trunk/src/main/org/hornetq/core/protocol/core/impl/PacketDecoder.java trunk/src/main/org/hornetq/core/protocol/core/impl/wireformat/SessionReceiveLargeMessage.java trunk/src/main/org/hornetq/core/protocol/core/impl/wireformat/SessionSendLargeMessage.java trunk/src/main/org/hornetq/core/protocol/core/impl/wireformat/SessionSendMessage.java trunk/src/main/org/hornetq/core/protocol/stomp/StompSession.java trunk/src/main/org/hornetq/core/server/ServerSession.java trunk/src/main/org/hornetq/core/server/impl/ServerConsumerImpl.java trunk/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java trunk/src/main/org/hornetq/spi/core/protocol/SessionCallback.java trunk/tests/src/org/hornetq/tests/unit/core/client/impl/LargeMessageBufferTest.java trunk/tests/src/org/hornetq/tests/unit/core/paging/impl/PagingStoreImplTest.java trunk/tests/src/org/hornetq/tests/unit/core/postoffice/impl/BindingsImplTest.java Log: HORNETQ-446 - Improvements on Large message - first step (optimizing encoding - removing two phase encoding) Modified: trunk/src/main/org/hornetq/core/client/impl/ClientConsumerImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/ClientConsumerImpl.java 2010-12-07 18:05:17 UTC (rev 10007) +++ trunk/src/main/org/hornetq/core/client/impl/ClientConsumerImpl.java 2010-12-07 22:22:35 UTC (rev 10008) @@ -29,7 +29,6 @@ import org.hornetq.core.protocol.core.impl.wireformat.SessionQueueQueryResponseMessage; import org.hornetq.core.protocol.core.impl.wireformat.SessionReceiveContinuationMessage; import org.hornetq.core.protocol.core.impl.wireformat.SessionReceiveLargeMessage; -import org.hornetq.utils.DecompressedLargeMessageBuffer; import org.hornetq.utils.Future; import org.hornetq.utils.PriorityLinkedList; import org.hornetq.utils.PriorityLinkedListImpl; @@ -83,7 +82,7 @@ private final Runner runner = new Runner(); - private LargeMessageBufferImpl currentLargeMessageBuffer; + private LargeMessageControllerImpl currentLargeMessageController; // When receiving LargeMessages, the user may choose to not read the body, on this case we need to discard the body // before moving to the next message. @@ -539,14 +538,10 @@ flowControl(packet.getPacketSize(), false); - ClientMessageInternal currentChunkMessage = new ClientMessageImpl(); + ClientMessageInternal currentChunkMessage = (ClientMessageInternal)packet.getLargeMessage(); currentChunkMessage.setDeliveryCount(packet.getDeliveryCount()); - // FIXME - this is really inefficient - decoding from a buffer to a byte[] then from the byte[] to another buffer - // which is then decoded to form the message! Clebert, what were you thinking? - currentChunkMessage.decodeHeadersAndProperties(HornetQBuffers.wrappedBuffer(packet.getLargeMessageHeader())); - currentChunkMessage.setLargeMessage(true); File largeMessageCache = null; @@ -558,15 +553,15 @@ largeMessageCache.deleteOnExit(); } - currentLargeMessageBuffer = new LargeMessageBufferImpl(this, packet.getLargeMessageSize(), 60, largeMessageCache); + currentLargeMessageController = new LargeMessageControllerImpl(this, packet.getLargeMessageSize(), 60, largeMessageCache); if (currentChunkMessage.isCompressed()) { - currentChunkMessage.setBuffer(new DecompressedLargeMessageBuffer(currentLargeMessageBuffer)); + currentChunkMessage.setBuffer(new DecompressedLargeMessageControllerImpl(currentLargeMessageController)); } else { - currentChunkMessage.setBuffer(currentLargeMessageBuffer); + currentChunkMessage.setBuffer(currentLargeMessageController); } currentChunkMessage.setFlowControlSize(0); @@ -580,7 +575,7 @@ { return; } - currentLargeMessageBuffer.addPacket(chunk); + currentLargeMessageController.addPacket(chunk); } public void clear(boolean waitForOnMessage) throws HornetQException @@ -904,10 +899,10 @@ // Now we wait for any current handler runners to run. waitForOnMessageToComplete(true); - if (currentLargeMessageBuffer != null) + if (currentLargeMessageController != null) { - currentLargeMessageBuffer.cancel(); - currentLargeMessageBuffer = null; + currentLargeMessageController.cancel(); + currentLargeMessageController = null; } closed = true; Modified: trunk/src/main/org/hornetq/core/client/impl/ClientMessageImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/ClientMessageImpl.java 2010-12-07 18:05:17 UTC (rev 10007) +++ trunk/src/main/org/hornetq/core/client/impl/ClientMessageImpl.java 2010-12-07 22:22:35 UTC (rev 10008) @@ -47,6 +47,9 @@ private ClientConsumerInternal consumer; private boolean largeMessage; + + // Used only when receiving large messages + private LargeMessageController largeMessageController; private int flowControlSize = -1; @@ -159,7 +162,7 @@ { if (largeMessage) { - ((LargeMessageBufferInternal)getWholeBuffer()).saveBuffer(out); + ((LargeMessageController)getWholeBuffer()).saveBuffer(out); } else { @@ -184,7 +187,7 @@ { if (largeMessage) { - ((LargeMessageBufferInternal)getWholeBuffer()).setOutputStream(out); + ((LargeMessageController)getWholeBuffer()).setOutputStream(out); } else { @@ -200,7 +203,7 @@ { if (largeMessage) { - return ((LargeMessageBufferInternal)getWholeBuffer()).waitCompletion(timeMilliseconds); + return ((LargeMessageController)getWholeBuffer()).waitCompletion(timeMilliseconds); } else { @@ -215,7 +218,7 @@ { if (largeMessage) { - ((LargeMessageBufferInternal)getWholeBuffer()).discardUnusedPackets(); + ((LargeMessageController)getWholeBuffer()).discardUnusedPackets(); } } Modified: trunk/src/main/org/hornetq/core/client/impl/ClientProducerImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/ClientProducerImpl.java 2010-12-07 18:05:17 UTC (rev 10007) +++ trunk/src/main/org/hornetq/core/client/impl/ClientProducerImpl.java 2010-12-07 22:22:35 UTC (rev 10008) @@ -335,7 +335,7 @@ int headerSize = msgI.getHeadersAndPropertiesEncodeSize(); - if (headerSize >= minLargeMessageSize) + if (msgI.getHeadersAndPropertiesEncodeSize() >= minLargeMessageSize) { throw new HornetQException(HornetQException.ILLEGAL_STATE, "Header size (" + headerSize + ") is too big, use the messageBody for large data, or increase minLargeMessageSize"); @@ -347,11 +347,8 @@ msgI.getWholeBuffer().readerIndex(0); } - HornetQBuffer headerBuffer = HornetQBuffers.fixedBuffer(headerSize); + SessionSendLargeMessage initialChunk = new SessionSendLargeMessage(msgI); - msgI.encodeHeadersAndProperties(headerBuffer); - SessionSendLargeMessage initialChunk = new SessionSendLargeMessage(headerBuffer.toByteBuffer().array()); - channel.send(initialChunk); try Copied: trunk/src/main/org/hornetq/core/client/impl/DecompressedLargeMessageControllerImpl.java (from rev 10006, trunk/src/main/org/hornetq/utils/DecompressedLargeMessageBuffer.java) =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/DecompressedLargeMessageControllerImpl.java (rev 0) +++ trunk/src/main/org/hornetq/core/client/impl/DecompressedLargeMessageControllerImpl.java 2010-12-07 22:22:35 UTC (rev 10008) @@ -0,0 +1,1094 @@ +/* + * Copyright 2010 Red Hat, Inc. + * Red Hat licenses this file to you under the Apache License, version + * 2.0 (the "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + * implied. See the License for the specific language governing + * permissions and limitations under the License. + */ + +package org.hornetq.core.client.impl; + +import java.io.DataInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.nio.ByteBuffer; +import java.nio.channels.GatheringByteChannel; +import java.nio.channels.ScatteringByteChannel; + +import org.hornetq.api.core.HornetQBuffer; +import org.hornetq.api.core.HornetQBuffers; +import org.hornetq.api.core.HornetQException; +import org.hornetq.api.core.SimpleString; +import org.hornetq.core.protocol.core.impl.wireformat.SessionReceiveContinuationMessage; +import org.hornetq.utils.DataConstants; +import org.hornetq.utils.HornetQBufferInputStream; +import org.hornetq.utils.InflaterReader; +import org.hornetq.utils.InflaterWriter; +import org.hornetq.utils.UTF8Util; +import org.jboss.netty.buffer.ChannelBuffer; + +/** + * A DecompressedHornetQBuffer + * + * @author <a href="mailto:clebert.suconic@jboss.org">Clebert Suconic</a> + * + * + */ +public class DecompressedLargeMessageControllerImpl implements LargeMessageController +{ + + // Constants ----------------------------------------------------- + + private static final String OPERATION_NOT_SUPPORTED = "Operation not supported"; + + private static final String READ_ONLY_ERROR_MESSAGE = "This is a read-only buffer, setOperations are not supported"; + + // Attributes ---------------------------------------------------- + + final LargeMessageController bufferDelegate; + + // Static -------------------------------------------------------- + + // Constructors -------------------------------------------------- + + public DecompressedLargeMessageControllerImpl(final LargeMessageController bufferDelegate) + { + this.bufferDelegate = bufferDelegate; + } + + + // Public -------------------------------------------------------- + + /** + * + */ + public void discardUnusedPackets() + { + bufferDelegate.discardUnusedPackets(); + } + + /** + * Add a buff to the List, or save it to the OutputStream if set + * @param packet + */ + public void addPacket(final SessionReceiveContinuationMessage packet) + { + bufferDelegate.addPacket(packet); + } + + public synchronized void cancel() + { + bufferDelegate.cancel(); + } + + public synchronized void close() + { + bufferDelegate.cancel(); + } + + public void setOutputStream(final OutputStream output) throws HornetQException + { + bufferDelegate.setOutputStream(new InflaterWriter(output)); + } + + public synchronized void saveBuffer(final OutputStream output) throws HornetQException + { + setOutputStream(output); + waitCompletion(0); + } + + /** + * + * @param timeWait Milliseconds to Wait. 0 means forever + * @throws Exception + */ + public synchronized boolean waitCompletion(final long timeWait) throws HornetQException + { + return bufferDelegate.waitCompletion(timeWait); + } + + // Channel Buffer Implementation --------------------------------- + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#array() + */ + public byte[] array() + { + throw new IllegalAccessError("array not supported on LargeMessageBufferImpl"); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#capacity() + */ + public int capacity() + { + return -1; + } + + DataInputStream dataInput = null; + + private DataInputStream getStream() + { + if (dataInput == null) + { + try + { + InputStream input = new HornetQBufferInputStream(bufferDelegate); + + dataInput = new DataInputStream(new InflaterReader(input)); + } + catch (Exception e) + { + throw new RuntimeException (e.getMessage(), e); + } + + } + return dataInput; + } + + private void positioningNotSupported() + { + throw new IllegalStateException("Position not supported over compressed large messages"); + } + + public byte readByte() + { + try + { + return getStream().readByte(); + } + catch (Exception e) + { + throw new RuntimeException (e.getMessage(), e); + } + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#getByte(int) + */ + public byte getByte(final int index) + { + positioningNotSupported(); + return 0; + } + + private byte getByte(final long index) + { + positioningNotSupported(); + return 0; + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, org.hornetq.api.core.buffers.ChannelBuffer, int, int) + */ + public void getBytes(final int index, final HornetQBuffer dst, final int dstIndex, final int length) + { + positioningNotSupported(); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, org.hornetq.api.core.buffers.ChannelBuffer, int, int) + */ + public void getBytes(final long index, final HornetQBuffer dst, final int dstIndex, final int length) + { + positioningNotSupported(); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, byte[], int, int) + */ + public void getBytes(final int index, final byte[] dst, final int dstIndex, final int length) + { + positioningNotSupported(); + } + + public void getBytes(final long index, final byte[] dst, final int dstIndex, final int length) + { + positioningNotSupported(); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, java.nio.ByteBuffer) + */ + public void getBytes(final int index, final ByteBuffer dst) + { + positioningNotSupported(); + } + + public void getBytes(final long index, final ByteBuffer dst) + { + positioningNotSupported(); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, java.io.OutputStream, int) + */ + public void getBytes(final int index, final OutputStream out, final int length) throws IOException + { + positioningNotSupported(); + } + + public void getBytes(final long index, final OutputStream out, final int length) throws IOException + { + positioningNotSupported(); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, java.nio.channels.GatheringByteChannel, int) + */ + public int getBytes(final int index, final GatheringByteChannel out, final int length) throws IOException + { + positioningNotSupported(); + return 0; + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#getInt(int) + */ + public int getInt(final int index) + { + positioningNotSupported(); + return 0; + } + + public int getInt(final long index) + { + positioningNotSupported(); + return 0; + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#getLong(int) + */ + public long getLong(final int index) + { + positioningNotSupported(); + return 0; + } + + public long getLong(final long index) + { + positioningNotSupported(); + return 0; + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#getShort(int) + */ + public short getShort(final int index) + { + positioningNotSupported(); + return 0; + } + + public short getShort(final long index) + { + return (short)(getByte(index) << 8 | getByte(index + 1) & 0xFF); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#getUnsignedMedium(int) + */ + public int getUnsignedMedium(final int index) + { + positioningNotSupported(); + return 0; + } + + + + public int getUnsignedMedium(final long index) + { + positioningNotSupported(); + return 0; + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#setByte(int, byte) + */ + public void setByte(final int index, final byte value) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#setBytes(int, org.hornetq.api.core.buffers.ChannelBuffer, int, int) + */ + public void setBytes(final int index, final HornetQBuffer src, final int srcIndex, final int length) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#setBytes(int, byte[], int, int) + */ + public void setBytes(final int index, final byte[] src, final int srcIndex, final int length) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#setBytes(int, java.nio.ByteBuffer) + */ + public void setBytes(final int index, final ByteBuffer src) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#setBytes(int, java.io.InputStream, int) + */ + public int setBytes(final int index, final InputStream in, final int length) throws IOException + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#setBytes(int, java.nio.channels.ScatteringByteChannel, int) + */ + public int setBytes(final int index, final ScatteringByteChannel in, final int length) throws IOException + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#setInt(int, int) + */ + public void setInt(final int index, final int value) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#setLong(int, long) + */ + public void setLong(final int index, final long value) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#setMedium(int, int) + */ + public void setMedium(final int index, final int value) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#setShort(int, short) + */ + public void setShort(final int index, final short value) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#toByteBuffer(int, int) + */ + public ByteBuffer toByteBuffer(final int index, final int length) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#toString(int, int, java.lang.String) + */ + public String toString(final int index, final int length, final String charsetName) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public int readerIndex() + { + return 0; + } + + public void readerIndex(final int readerIndex) + { + // TODO + } + + public int writerIndex() + { + // TODO + return 0; + } + + public long getSize() + { + // TODO + return 0; + } + + public void writerIndex(final int writerIndex) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void setIndex(final int readerIndex, final int writerIndex) + { + positioningNotSupported(); + } + + public void clear() + { + } + + public boolean readable() + { + return true; + } + + public boolean writable() + { + return false; + } + + public int readableBytes() + { + return 1; + } + + public int writableBytes() + { + return 0; + } + + public void markReaderIndex() + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void resetReaderIndex() + { + // TODO: reset positioning if possible + } + + public void markWriterIndex() + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void resetWriterIndex() + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void discardReadBytes() + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public short getUnsignedByte(final int index) + { + return (short)(getByte(index) & 0xFF); + } + + public int getUnsignedShort(final int index) + { + return getShort(index) & 0xFFFF; + } + + public int getMedium(final int index) + { + int value = getUnsignedMedium(index); + if ((value & 0x800000) != 0) + { + value |= 0xff000000; + } + return value; + } + + public long getUnsignedInt(final int index) + { + return getInt(index) & 0xFFFFFFFFL; + } + + public void getBytes(int index, final byte[] dst) + { + // TODO: optimize this by using System.arraycopy + for (int i = 0; i < dst.length; i++) + { + dst[i] = getByte(index++); + } + } + + public void getBytes(long index, final byte[] dst) + { + // TODO: optimize this by using System.arraycopy + for (int i = 0; i < dst.length; i++) + { + dst[i] = getByte(index++); + } + } + + public void getBytes(final int index, final HornetQBuffer dst) + { + getBytes(index, dst, dst.writableBytes()); + } + + public void getBytes(final int index, final HornetQBuffer dst, final int length) + { + if (length > dst.writableBytes()) + { + throw new IndexOutOfBoundsException(); + } + getBytes(index, dst, dst.writerIndex(), length); + dst.writerIndex(dst.writerIndex() + length); + } + + public void setBytes(final int index, final byte[] src) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void setBytes(final int index, final HornetQBuffer src) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void setBytes(final int index, final HornetQBuffer src, final int length) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void setZero(final int index, final int length) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public short readUnsignedByte() + { + try + { + return (short)getStream().readUnsignedByte(); + } + catch (Exception e) + { + throw new IllegalStateException (e.getMessage(), e); + } + } + + public short readShort() + { + try + { + return (short)getStream().readShort(); + } + catch (Exception e) + { + throw new IllegalStateException (e.getMessage(), e); + } + } + + public int readUnsignedShort() + { + try + { + return (int)getStream().readUnsignedShort(); + } + catch (Exception e) + { + throw new IllegalStateException (e.getMessage(), e); + } + } + + public int readMedium() + { + int value = readUnsignedMedium(); + if ((value & 0x800000) != 0) + { + value |= 0xff000000; + } + return value; + } + + + public int readUnsignedMedium() + { + return (readByte() & 0xff) << 16 | (readByte() & 0xff) << 8 | (readByte() & 0xff) << 0; + } + + public int readInt() + { + try + { + return getStream().readInt(); + } + catch (Exception e) + { + throw new IllegalStateException(e.getMessage(), e); + } + } + + public int readInt(final int pos) + { + positioningNotSupported(); + return 0; + } + + public long readUnsignedInt() + { + return readInt() & 0xFFFFFFFFL; + } + + public long readLong() + { + try + { + return getStream().readLong(); + } + catch (Exception e) + { + throw new IllegalStateException(e.getMessage(), e); + } + } + + public void readBytes(final byte[] dst, final int dstIndex, final int length) + { + try + { + getStream().read(dst, dstIndex, length); + } + catch (Exception e) + { + throw new IllegalStateException(e.getMessage(), e); + } + } + + public void readBytes(final byte[] dst) + { + readBytes(dst, 0, dst.length); + } + + public void readBytes(final HornetQBuffer dst) + { + readBytes(dst, dst.writableBytes()); + } + + public void readBytes(final HornetQBuffer dst, final int length) + { + if (length > dst.writableBytes()) + { + throw new IndexOutOfBoundsException(); + } + readBytes(dst, dst.writerIndex(), length); + dst.writerIndex(dst.writerIndex() + length); + } + + public void readBytes(final HornetQBuffer dst, final int dstIndex, final int length) + { + byte[] destBytes = new byte[length]; + readBytes(destBytes); + dst.setBytes(dstIndex, destBytes); + } + + public void readBytes(final ByteBuffer dst) + { + byte bytesToGet[] = new byte[dst.remaining()]; + readBytes(bytesToGet); + dst.put(bytesToGet); + } + + public int readBytes(final GatheringByteChannel out, final int length) throws IOException + { + throw new IllegalStateException("Not implemented!"); + } + + public void readBytes(final OutputStream out, final int length) throws IOException + { + throw new IllegalStateException("Not implemented!"); + } + + public void skipBytes(final int length) + { + + try + { + for (int i = 0 ; i < length; i++) + { + getStream().read(); + } + } + catch (Exception e) + { + throw new IllegalStateException(e.getMessage(), e); + } + } + + public void writeByte(final byte value) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void writeShort(final short value) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void writeMedium(final int value) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void writeInt(final int value) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void writeLong(final long value) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void writeBytes(final byte[] src, final int srcIndex, final int length) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void writeBytes(final byte[] src) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void writeBytes(final HornetQBuffer src) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void writeBytes(final HornetQBuffer src, final int length) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void writeBytes(final ByteBuffer src) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public int writeBytes(final InputStream in, final int length) throws IOException + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public int writeBytes(final ScatteringByteChannel in, final int length) throws IOException + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void writeZero(final int length) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public ByteBuffer toByteBuffer() + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public ByteBuffer[] toByteBuffers() + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public ByteBuffer[] toByteBuffers(final int index, final int length) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public String toString(final String charsetName) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public Object getUnderlyingBuffer() + { + return this; + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#readBoolean() + */ + public boolean readBoolean() + { + return readByte() != 0; + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#readChar() + */ + public char readChar() + { + return (char)readShort(); + } + + public char getChar(final int index) + { + return (char)getShort(index); + } + + public double getDouble(final int index) + { + return Double.longBitsToDouble(getLong(index)); + } + + public float getFloat(final int index) + { + return Float.intBitsToFloat(getInt(index)); + } + + public HornetQBuffer readBytes(final int length) + { + byte bytesToGet[] = new byte[length]; + readBytes(bytesToGet); + return HornetQBuffers.wrappedBuffer(bytesToGet); + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#readDouble() + */ + public double readDouble() + { + return Double.longBitsToDouble(readLong()); + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#readFloat() + */ + public float readFloat() + { + return Float.intBitsToFloat(readInt()); + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#readNullableSimpleString() + */ + public SimpleString readNullableSimpleString() + { + int b = readByte(); + if (b == DataConstants.NULL) + { + return null; + } + else + { + return readSimpleString(); + } + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#readNullableString() + */ + public String readNullableString() + { + int b = readByte(); + if (b == DataConstants.NULL) + { + return null; + } + else + { + return readString(); + } + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#readSimpleString() + */ + public SimpleString readSimpleString() + { + int len = readInt(); + byte[] data = new byte[len]; + readBytes(data); + return new SimpleString(data); + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#readString() + */ + public String readString() + { + int len = readInt(); + + if (len < 9) + { + char[] chars = new char[len]; + for (int i = 0; i < len; i++) + { + chars[i] = (char)readShort(); + } + return new String(chars); + } + else if (len < 0xfff) + { + return readUTF(); + } + else + { + return readSimpleString().toString(); + } + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#readUTF() + */ + public String readUTF() + { + return UTF8Util.readUTF(this); + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeBoolean(boolean) + */ + public void writeBoolean(final boolean val) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeChar(char) + */ + public void writeChar(final char val) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeDouble(double) + */ + public void writeDouble(final double val) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeFloat(float) + */ + public void writeFloat(final float val) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeNullableSimpleString(org.hornetq.util.SimpleString) + */ + public void writeNullableSimpleString(final SimpleString val) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeNullableString(java.lang.String) + */ + public void writeNullableString(final String val) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeSimpleString(org.hornetq.util.SimpleString) + */ + public void writeSimpleString(final SimpleString val) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeString(java.lang.String) + */ + public void writeString(final String val) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeUTF(java.lang.String) + */ + public void writeUTF(final String utf) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#compareTo(org.hornetq.api.core.buffers.ChannelBuffer) + */ + public int compareTo(final HornetQBuffer buffer) + { + return -1; + } + + public HornetQBuffer copy() + { + throw new UnsupportedOperationException(); + } + + public HornetQBuffer slice(final int index, final int length) + { + throw new UnsupportedOperationException(); + } + + // Package protected --------------------------------------------- + + // Protected ----------------------------------------------------- + + // Private ------------------------------------------------------- + + /** + * @param body + */ + // Inner classes ------------------------------------------------- + + public ChannelBuffer channelBuffer() + { + return null; + } + + public HornetQBuffer copy(final int index, final int length) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public HornetQBuffer duplicate() + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public HornetQBuffer readSlice(final int length) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void setChar(final int index, final char value) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void setDouble(final int index, final double value) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void setFloat(final int index, final float value) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public HornetQBuffer slice() + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } + + public void writeBytes(final HornetQBuffer src, final int srcIndex, final int length) + { + throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); + } +} Deleted: trunk/src/main/org/hornetq/core/client/impl/LargeMessageBufferImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/LargeMessageBufferImpl.java 2010-12-07 18:05:17 UTC (rev 10007) +++ trunk/src/main/org/hornetq/core/client/impl/LargeMessageBufferImpl.java 2010-12-07 22:22:35 UTC (rev 10008) @@ -1,1535 +0,0 @@ -/* - * Copyright 2009 Red Hat, Inc. - * Red Hat licenses this file to you under the Apache License, version - * 2.0 (the "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * http://www.apache.org/licenses/LICENSE-2.0 - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or - * implied. See the License for the specific language governing - * permissions and limitations under the License. - */ - -package org.hornetq.core.client.impl; - -import java.io.File; -import java.io.FileNotFoundException; -import java.io.IOException; -import java.io.InputStream; -import java.io.OutputStream; -import java.io.RandomAccessFile; -import java.nio.ByteBuffer; -import java.nio.channels.FileChannel; -import java.nio.channels.GatheringByteChannel; -import java.nio.channels.ScatteringByteChannel; -import java.util.concurrent.LinkedBlockingQueue; -import java.util.concurrent.TimeUnit; - -import org.hornetq.api.core.HornetQBuffer; -import org.hornetq.api.core.HornetQBuffers; -import org.hornetq.api.core.HornetQException; -import org.hornetq.api.core.SimpleString; -import org.hornetq.core.logging.Logger; -import org.hornetq.core.protocol.core.impl.wireformat.SessionReceiveContinuationMessage; -import org.hornetq.utils.DataConstants; -import org.hornetq.utils.UTF8Util; -import org.jboss.netty.buffer.ChannelBuffer; - -/** - * This class aggregates several SessionReceiveContinuationMessages as it was being handled by a single buffer. - * This buffer can be consumed as messages are arriving, and it will hold the packets until they are read using the ChannelBuffer interface, or the setOutputStream or saveStream are called. - * - * @author <a href="mailto:clebert.suconic@jboss.org">Clebert Suconic</a> - * - * - */ -public class LargeMessageBufferImpl implements LargeMessageBufferInternal -{ - // Constants ----------------------------------------------------- - - private static final String READ_ONLY_ERROR_MESSAGE = "This is a read-only buffer, setOperations are not supported"; - - // Attributes ---------------------------------------------------- - - private static final Logger log = Logger.getLogger(LargeMessageBufferImpl.class); - - private final ClientConsumerInternal consumerInternal; - - private final LinkedBlockingQueue<SessionReceiveContinuationMessage> packets = new LinkedBlockingQueue<SessionReceiveContinuationMessage>(); - - private volatile SessionReceiveContinuationMessage currentPacket = null; - - private final long totalSize; - - private final int bufferSize; - - private boolean streamEnded = false; - - private boolean streamClosed = false; - - private final int readTimeout; - - private long readerIndex = 0; - - private long packetPosition = -1; - - private long lastIndex = 0; - - private long packetLastPosition = -1; - - private OutputStream outStream; - - private Exception handledException; - - private final FileCache fileCache; - - // Static -------------------------------------------------------- - - // Constructors -------------------------------------------------- - - public LargeMessageBufferImpl(final ClientConsumerInternal consumerInternal, - final long totalSize, - final int readTimeout) - { - this(consumerInternal, totalSize, readTimeout, null); - } - - public LargeMessageBufferImpl(final ClientConsumerInternal consumerInternal, - final long totalSize, - final int readTimeout, - final File cachedFile) - { - this(consumerInternal, totalSize, readTimeout, cachedFile, 10 * 1024); - } - - public LargeMessageBufferImpl(final ClientConsumerInternal consumerInternal, - final long totalSize, - final int readTimeout, - final File cachedFile, - final int bufferSize) - { - this.consumerInternal = consumerInternal; - this.readTimeout = readTimeout; - this.totalSize = totalSize; - if (cachedFile == null) - { - fileCache = null; - } - else - { - fileCache = new FileCache(cachedFile); - } - this.bufferSize = bufferSize; - } - - // Public -------------------------------------------------------- - - public synchronized Exception getHandledException() - { - return handledException; - } - - /** - * - */ - public void discardUnusedPackets() - { - if (outStream == null) - { - try - { - checkForPacket(totalSize - 1); - } - catch (Exception ignored) - { - } - } - } - - /** - * Add a buff to the List, or save it to the OutputStream if set - * @param packet - */ - public void addPacket(final SessionReceiveContinuationMessage packet) - { - int flowControlCredit = 0; - boolean continues = false; - - synchronized (this) - { - if (outStream != null) - { - try - { - if (!packet.isContinues()) - { - streamEnded = true; - } - - if (fileCache != null) - { - fileCache.cachePackage(packet.getBody()); - } - - outStream.write(packet.getBody()); - - flowControlCredit = packet.getPacketSize(); - - continues = packet.isContinues(); - - notifyAll(); - - if (streamEnded) - { - outStream.close(); - } - } - catch (Exception e) - { - LargeMessageBufferImpl.log.warn(e.getMessage(), e); - handledException = e; - } - } - else - { - if (fileCache != null) - { - try - { - fileCache.cachePackage(packet.getBody()); - } - catch (Exception e) - { - LargeMessageBufferImpl.log.warn(e.getMessage(), e); - handledException = e; - } - } - - - packets.offer(packet); - } - } - - if (flowControlCredit != 0) - { - try - { - consumerInternal.flowControl(flowControlCredit, !continues); - } - catch (Exception e) - { - LargeMessageBufferImpl.log.warn(e.getMessage(), e); - handledException = e; - } - } - } - - public synchronized void cancel() - { - packets.offer(new SessionReceiveContinuationMessage()); - streamEnded = true; - streamClosed = true; - - notifyAll(); - } - - public synchronized void close() - { - if (fileCache != null) - { - fileCache.close(); - } - } - - public void setOutputStream(final OutputStream output) throws HornetQException - { - - int totalFlowControl = 0; - boolean continues = false; - - synchronized (this) - { - if (currentPacket != null) - { - sendPacketToOutput(output, currentPacket); - currentPacket = null; - } - while (true) - { - SessionReceiveContinuationMessage packet = packets.poll(); - if (packet == null) - { - break; - } - totalFlowControl += packet.getPacketSize(); - - continues = packet.isContinues(); - sendPacketToOutput(output, packet); - } - - outStream = output; - } - - consumerInternal.flowControl(totalFlowControl, !continues); - } - - public synchronized void saveBuffer(final OutputStream output) throws HornetQException - { - setOutputStream(output); - waitCompletion(0); - } - - /** - * - * @param timeWait Milliseconds to Wait. 0 means forever - * @throws Exception - */ - public synchronized boolean waitCompletion(final long timeWait) throws HornetQException - { - if (outStream == null) - { - // There is no stream.. it will never achieve the end of streaming - return false; - } - - long timeOut = System.currentTimeMillis() + timeWait; - while (!streamEnded && handledException == null) - { - try - { - this.wait(readTimeout == 0 ? 1 : readTimeout * 1000); - } - catch (InterruptedException e) - { - throw new HornetQException(HornetQException.INTERNAL_ERROR, e.getMessage(), e); - } - - if (timeWait > 0 && System.currentTimeMillis() > timeOut) - { - throw new HornetQException(HornetQException.LARGE_MESSAGE_ERROR_BODY, - "Timeout waiting for LargeMessage Body"); - } - } - - if (handledException != null) - { - throw new HornetQException(HornetQException.LARGE_MESSAGE_ERROR_BODY, - "Error on saving LargeMessageBufferImpl", - handledException); - } - - return streamEnded; - - } - - // Channel Buffer Implementation --------------------------------- - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#array() - */ - public byte[] array() - { - throw new IllegalAccessError("array not supported on LargeMessageBufferImpl"); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#capacity() - */ - public int capacity() - { - return -1; - } - - public byte readByte() - { - return getByte(readerIndex++); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#getByte(int) - */ - public byte getByte(final int index) - { - return getByte((long)index); - } - - private byte getByte(final long index) - { - checkForPacket(index); - - if (fileCache != null && index < packetPosition) - { - return fileCache.getByteFromCache(index); - } - else - { - return currentPacket.getBody()[(int)(index - packetPosition)]; - } - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, org.hornetq.api.core.buffers.ChannelBuffer, int, int) - */ - public void getBytes(final int index, final HornetQBuffer dst, final int dstIndex, final int length) - { - byte[] destBytes = new byte[length]; - getBytes(index, destBytes); - dst.setBytes(dstIndex, destBytes); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, org.hornetq.api.core.buffers.ChannelBuffer, int, int) - */ - public void getBytes(final long index, final HornetQBuffer dst, final int dstIndex, final int length) - { - byte[] destBytes = new byte[length]; - getBytes(index, destBytes); - dst.setBytes(dstIndex, destBytes); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, byte[], int, int) - */ - public void getBytes(final int index, final byte[] dst, final int dstIndex, final int length) - { - byte bytesToGet[] = new byte[length]; - - getBytes(index, bytesToGet); - - System.arraycopy(bytesToGet, 0, dst, dstIndex, length); - } - - public void getBytes(final long index, final byte[] dst, final int dstIndex, final int length) - { - byte bytesToGet[] = new byte[length]; - - getBytes(index, bytesToGet); - - System.arraycopy(bytesToGet, 0, dst, dstIndex, length); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, java.nio.ByteBuffer) - */ - public void getBytes(final int index, final ByteBuffer dst) - { - byte bytesToGet[] = new byte[dst.remaining()]; - getBytes(index, bytesToGet); - dst.put(bytesToGet); - } - - public void getBytes(final long index, final ByteBuffer dst) - { - byte bytesToGet[] = new byte[dst.remaining()]; - getBytes(index, bytesToGet); - dst.put(bytesToGet); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, java.io.OutputStream, int) - */ - public void getBytes(final int index, final OutputStream out, final int length) throws IOException - { - byte bytesToGet[] = new byte[length]; - getBytes(index, bytesToGet); - out.write(bytesToGet); - } - - public void getBytes(final long index, final OutputStream out, final int length) throws IOException - { - byte bytesToGet[] = new byte[length]; - getBytes(index, bytesToGet); - out.write(bytesToGet); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, java.nio.channels.GatheringByteChannel, int) - */ - public int getBytes(final int index, final GatheringByteChannel out, final int length) throws IOException - { - byte bytesToGet[] = new byte[length]; - getBytes(index, bytesToGet); - return out.write(ByteBuffer.wrap(bytesToGet)); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#getInt(int) - */ - public int getInt(final int index) - { - return (getByte(index) & 0xff) << 24 | (getByte(index + 1) & 0xff) << 16 | - (getByte(index + 2) & 0xff) << 8 | - (getByte(index + 3) & 0xff) << 0; - } - - public int getInt(final long index) - { - return (getByte(index) & 0xff) << 24 | (getByte(index + 1) & 0xff) << 16 | - (getByte(index + 2) & 0xff) << 8 | - (getByte(index + 3) & 0xff) << 0; - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#getLong(int) - */ - public long getLong(final int index) - { - return ((long)getByte(index) & 0xff) << 56 | ((long)getByte(index + 1) & 0xff) << 48 | - ((long)getByte(index + 2) & 0xff) << 40 | - ((long)getByte(index + 3) & 0xff) << 32 | - ((long)getByte(index + 4) & 0xff) << 24 | - ((long)getByte(index + 5) & 0xff) << 16 | - ((long)getByte(index + 6) & 0xff) << 8 | - ((long)getByte(index + 7) & 0xff) << 0; - } - - public long getLong(final long index) - { - return ((long)getByte(index) & 0xff) << 56 | ((long)getByte(index + 1) & 0xff) << 48 | - ((long)getByte(index + 2) & 0xff) << 40 | - ((long)getByte(index + 3) & 0xff) << 32 | - ((long)getByte(index + 4) & 0xff) << 24 | - ((long)getByte(index + 5) & 0xff) << 16 | - ((long)getByte(index + 6) & 0xff) << 8 | - ((long)getByte(index + 7) & 0xff) << 0; - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#getShort(int) - */ - public short getShort(final int index) - { - return (short)(getByte(index) << 8 | getByte(index + 1) & 0xFF); - } - - public short getShort(final long index) - { - return (short)(getByte(index) << 8 | getByte(index + 1) & 0xFF); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#getUnsignedMedium(int) - */ - public int getUnsignedMedium(final int index) - { - return (getByte(index) & 0xff) << 16 | (getByte(index + 1) & 0xff) << 8 | (getByte(index + 2) & 0xff) << 0; - } - - public int getUnsignedMedium(final long index) - { - return (getByte(index) & 0xff) << 16 | (getByte(index + 1) & 0xff) << 8 | (getByte(index + 2) & 0xff) << 0; - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#setByte(int, byte) - */ - public void setByte(final int index, final byte value) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#setBytes(int, org.hornetq.api.core.buffers.ChannelBuffer, int, int) - */ - public void setBytes(final int index, final HornetQBuffer src, final int srcIndex, final int length) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#setBytes(int, byte[], int, int) - */ - public void setBytes(final int index, final byte[] src, final int srcIndex, final int length) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#setBytes(int, java.nio.ByteBuffer) - */ - public void setBytes(final int index, final ByteBuffer src) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#setBytes(int, java.io.InputStream, int) - */ - public int setBytes(final int index, final InputStream in, final int length) throws IOException - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#setBytes(int, java.nio.channels.ScatteringByteChannel, int) - */ - public int setBytes(final int index, final ScatteringByteChannel in, final int length) throws IOException - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#setInt(int, int) - */ - public void setInt(final int index, final int value) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#setLong(int, long) - */ - public void setLong(final int index, final long value) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#setMedium(int, int) - */ - public void setMedium(final int index, final int value) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#setShort(int, short) - */ - public void setShort(final int index, final short value) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#toByteBuffer(int, int) - */ - public ByteBuffer toByteBuffer(final int index, final int length) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#toString(int, int, java.lang.String) - */ - public String toString(final int index, final int length, final String charsetName) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public int readerIndex() - { - return (int)readerIndex; - } - - public void readerIndex(final int readerIndex) - { - try - { - checkForPacket(readerIndex); - } - catch (Exception e) - { - LargeMessageBufferImpl.log.warn(e.getMessage(), e); - throw new RuntimeException(e.getMessage(), e); - } - this.readerIndex = readerIndex; - } - - public int writerIndex() - { - return (int)totalSize; - } - - public long getSize() - { - return totalSize; - } - - public void writerIndex(final int writerIndex) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public void setIndex(final int readerIndex, final int writerIndex) - { - try - { - checkForPacket(readerIndex); - } - catch (Exception e) - { - LargeMessageBufferImpl.log.warn(e.getMessage(), e); - throw new RuntimeException(e.getMessage(), e); - } - this.readerIndex = readerIndex; - } - - public void clear() - { - } - - public boolean readable() - { - return true; - } - - public boolean writable() - { - return false; - } - - public int readableBytes() - { - long readableBytes = totalSize - readerIndex; - - if (readableBytes > Integer.MAX_VALUE) - { - return Integer.MAX_VALUE; - } - else - { - return (int)(totalSize - readerIndex); - } - } - - public int writableBytes() - { - return 0; - } - - public void markReaderIndex() - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public void resetReaderIndex() - { - try - { - checkForPacket(0); - } - catch (Exception e) - { - LargeMessageBufferImpl.log.warn(e.getMessage(), e); - throw new RuntimeException(e.getMessage(), e); - } - } - - public void markWriterIndex() - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public void resetWriterIndex() - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public void discardReadBytes() - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public short getUnsignedByte(final int index) - { - return (short)(getByte(index) & 0xFF); - } - - public int getUnsignedShort(final int index) - { - return getShort(index) & 0xFFFF; - } - - public int getMedium(final int index) - { - int value = getUnsignedMedium(index); - if ((value & 0x800000) != 0) - { - value |= 0xff000000; - } - return value; - } - - public long getUnsignedInt(final int index) - { - return getInt(index) & 0xFFFFFFFFL; - } - - public void getBytes(int index, final byte[] dst) - { - // TODO: optimize this by using System.arraycopy - for (int i = 0; i < dst.length; i++) - { - dst[i] = getByte(index++); - } - } - - public void getBytes(long index, final byte[] dst) - { - // TODO: optimize this by using System.arraycopy - for (int i = 0; i < dst.length; i++) - { - dst[i] = getByte(index++); - } - } - - public void getBytes(final int index, final HornetQBuffer dst) - { - getBytes(index, dst, dst.writableBytes()); - } - - public void getBytes(final int index, final HornetQBuffer dst, final int length) - { - if (length > dst.writableBytes()) - { - throw new IndexOutOfBoundsException(); - } - getBytes(index, dst, dst.writerIndex(), length); - dst.writerIndex(dst.writerIndex() + length); - } - - public void setBytes(final int index, final byte[] src) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public void setBytes(final int index, final HornetQBuffer src) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public void setBytes(final int index, final HornetQBuffer src, final int length) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public void setZero(final int index, final int length) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public short readUnsignedByte() - { - return (short)(readByte() & 0xFF); - } - - public short readShort() - { - short v = getShort(readerIndex); - readerIndex += 2; - return v; - } - - public int readUnsignedShort() - { - return readShort() & 0xFFFF; - } - - public int readMedium() - { - int value = readUnsignedMedium(); - if ((value & 0x800000) != 0) - { - value |= 0xff000000; - } - return value; - } - - public int readUnsignedMedium() - { - int v = getUnsignedMedium(readerIndex); - readerIndex += 3; - return v; - } - - public int readInt() - { - int v = getInt(readerIndex); - readerIndex += 4; - return v; - } - - public int readInt(final int pos) - { - int v = getInt(pos); - return v; - } - - public long readUnsignedInt() - { - return readInt() & 0xFFFFFFFFL; - } - - public long readLong() - { - long v = getLong(readerIndex); - readerIndex += 8; - return v; - } - - public void readBytes(final byte[] dst, final int dstIndex, final int length) - { - getBytes(readerIndex, dst, dstIndex, length); - readerIndex += length; - } - - public void readBytes(final byte[] dst) - { - readBytes(dst, 0, dst.length); - } - - public void readBytes(final HornetQBuffer dst) - { - readBytes(dst, dst.writableBytes()); - } - - public void readBytes(final HornetQBuffer dst, final int length) - { - if (length > dst.writableBytes()) - { - throw new IndexOutOfBoundsException(); - } - readBytes(dst, dst.writerIndex(), length); - dst.writerIndex(dst.writerIndex() + length); - } - - public void readBytes(final HornetQBuffer dst, final int dstIndex, final int length) - { - getBytes(readerIndex, dst, dstIndex, length); - readerIndex += length; - } - - public void readBytes(final ByteBuffer dst) - { - int length = dst.remaining(); - getBytes(readerIndex, dst); - readerIndex += length; - } - - public int readBytes(final GatheringByteChannel out, final int length) throws IOException - { - int readBytes = getBytes((int)readerIndex, out, length); - readerIndex += readBytes; - return readBytes; - } - - public void readBytes(final OutputStream out, final int length) throws IOException - { - getBytes(readerIndex, out, length); - readerIndex += length; - } - - public void skipBytes(final int length) - { - - long newReaderIndex = readerIndex + length; - checkForPacket(newReaderIndex); - readerIndex = newReaderIndex; - } - - public void writeByte(final byte value) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public void writeShort(final short value) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public void writeMedium(final int value) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public void writeInt(final int value) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public void writeLong(final long value) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public void writeBytes(final byte[] src, final int srcIndex, final int length) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public void writeBytes(final byte[] src) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public void writeBytes(final HornetQBuffer src) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public void writeBytes(final HornetQBuffer src, final int length) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public void writeBytes(final ByteBuffer src) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public int writeBytes(final InputStream in, final int length) throws IOException - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public int writeBytes(final ScatteringByteChannel in, final int length) throws IOException - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public void writeZero(final int length) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public ByteBuffer toByteBuffer() - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public ByteBuffer[] toByteBuffers() - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public ByteBuffer[] toByteBuffers(final int index, final int length) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public String toString(final String charsetName) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public Object getUnderlyingBuffer() - { - return this; - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#readBoolean() - */ - public boolean readBoolean() - { - return readByte() != 0; - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#readChar() - */ - public char readChar() - { - return (char)readShort(); - } - - public char getChar(final int index) - { - return (char)getShort(index); - } - - public double getDouble(final int index) - { - return Double.longBitsToDouble(getLong(index)); - } - - public float getFloat(final int index) - { - return Float.intBitsToFloat(getInt(index)); - } - - public HornetQBuffer readBytes(final int length) - { - byte bytesToGet[] = new byte[length]; - getBytes(readerIndex, bytesToGet); - readerIndex += length; - return HornetQBuffers.wrappedBuffer(bytesToGet); - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#readDouble() - */ - public double readDouble() - { - return Double.longBitsToDouble(readLong()); - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#readFloat() - */ - public float readFloat() - { - return Float.intBitsToFloat(readInt()); - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#readNullableSimpleString() - */ - public SimpleString readNullableSimpleString() - { - int b = readByte(); - if (b == DataConstants.NULL) - { - return null; - } - else - { - return readSimpleString(); - } - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#readNullableString() - */ - public String readNullableString() - { - int b = readByte(); - if (b == DataConstants.NULL) - { - return null; - } - else - { - return readString(); - } - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#readSimpleString() - */ - public SimpleString readSimpleString() - { - int len = readInt(); - byte[] data = new byte[len]; - readBytes(data); - return new SimpleString(data); - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#readString() - */ - public String readString() - { - int len = readInt(); - - if (len < 9) - { - char[] chars = new char[len]; - for (int i = 0; i < len; i++) - { - chars[i] = (char)readShort(); - } - return new String(chars); - } - else if (len < 0xfff) - { - return readUTF(); - } - else - { - return readSimpleString().toString(); - } - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#readUTF() - */ - public String readUTF() - { - return UTF8Util.readUTF(this); - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeBoolean(boolean) - */ - public void writeBoolean(final boolean val) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeChar(char) - */ - public void writeChar(final char val) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeDouble(double) - */ - public void writeDouble(final double val) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeFloat(float) - */ - public void writeFloat(final float val) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeNullableSimpleString(org.hornetq.util.SimpleString) - */ - public void writeNullableSimpleString(final SimpleString val) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeNullableString(java.lang.String) - */ - public void writeNullableString(final String val) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeSimpleString(org.hornetq.util.SimpleString) - */ - public void writeSimpleString(final SimpleString val) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeString(java.lang.String) - */ - public void writeString(final String val) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeUTF(java.lang.String) - */ - public void writeUTF(final String utf) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#compareTo(org.hornetq.api.core.buffers.ChannelBuffer) - */ - public int compareTo(final HornetQBuffer buffer) - { - return -1; - } - - public HornetQBuffer copy() - { - throw new UnsupportedOperationException(); - } - - public HornetQBuffer slice(final int index, final int length) - { - throw new UnsupportedOperationException(); - } - - // Package protected --------------------------------------------- - - // Protected ----------------------------------------------------- - - // Private ------------------------------------------------------- - - /** - * @param output - * @param packet - * @throws HornetQException - */ - private void sendPacketToOutput(final OutputStream output, final SessionReceiveContinuationMessage packet) throws HornetQException - { - try - { - output.write(packet.getBody()); - if (!packet.isContinues()) - { - streamEnded = true; - output.close(); - } - } - catch (IOException e) - { - throw new HornetQException(HornetQException.LARGE_MESSAGE_ERROR_BODY, "Error writing body of message", e); - } - } - - private void popPacket() - { - try - { - - if (streamEnded) - { - // no more packets, we are over the last one already - throw new IndexOutOfBoundsException(); - } - - int sizeToAdd = currentPacket != null ? currentPacket.getBody().length : 1; - currentPacket = packets.poll(readTimeout, TimeUnit.SECONDS); - if (currentPacket == null) - { - throw new IndexOutOfBoundsException(); - } - - if (currentPacket.getBody() == null) // Empty packet as a signal to interruption - { - currentPacket = null; - streamEnded = true; - throw new IndexOutOfBoundsException(); - } - - consumerInternal.flowControl(currentPacket.getPacketSize(), !currentPacket.isContinues()); - - packetPosition += sizeToAdd; - - packetLastPosition = packetPosition + currentPacket.getBody().length; - } - catch (IndexOutOfBoundsException e) - { - throw e; - } - catch (Exception e) - { - throw new RuntimeException(e); - } - } - - private void checkForPacket(final long index) - { - if (outStream != null) - { - throw new IllegalAccessError("Can't read the messageBody after setting outputStream"); - } - - if (index >= totalSize) - { - throw new IndexOutOfBoundsException(); - } - - if (streamClosed) - { - throw new IllegalAccessError("The consumer associated with this large message was closed before the body was read"); - } - - if (fileCache == null) - { - if (index < lastIndex) - { - throw new IllegalAccessError("LargeMessage have read-only and one-way buffers"); - } - lastIndex = index; - } - - while (index >= packetLastPosition && !streamEnded) - { - popPacket(); - } - } - - /** - * @param body - */ - // Inner classes ------------------------------------------------- - private class FileCache - { - - public FileCache(final File cachedFile) - { - this.cachedFile = cachedFile; - } - - ByteBuffer readCache; - - long readCachePositionStart = Integer.MAX_VALUE; - - long readCachePositionEnd = -1; - - private final File cachedFile; - - private volatile RandomAccessFile cachedRAFile; - - private volatile FileChannel cachedChannel; - - private synchronized void readCache(final long position) - { - - try - { - if (position < readCachePositionStart || position > readCachePositionEnd) - { - - checkOpen(); - - if (position > cachedChannel.size()) - { - throw new ArrayIndexOutOfBoundsException("position > " + cachedChannel.size()); - } - - readCachePositionStart = position / bufferSize * bufferSize; - - cachedChannel.position(readCachePositionStart); - - if (readCache == null) - { - readCache = ByteBuffer.allocate(bufferSize); - } - - readCache.clear(); - - readCachePositionEnd = readCachePositionStart + cachedChannel.read(readCache) - 1; - } - } - catch (Exception e) - { - LargeMessageBufferImpl.log.warn(e.getMessage(), e); - throw new RuntimeException(e.getMessage(), e); - } - finally - { - close(); - } - } - - public synchronized byte getByteFromCache(final long position) - { - readCache(position); - - return readCache.get((int)(position - readCachePositionStart)); - - } - - public void cachePackage(final byte[] body) throws Exception - { - checkOpen(); - - cachedChannel.position(cachedChannel.size()); - cachedChannel.write(ByteBuffer.wrap(body)); - - close(); - } - - /** - * @throws FileNotFoundException - */ - public void checkOpen() throws FileNotFoundException - { - if (cachedFile != null || !cachedChannel.isOpen()) - { - cachedRAFile = new RandomAccessFile(cachedFile, "rw"); - - cachedChannel = cachedRAFile.getChannel(); - } - } - - public void close() - { - if (cachedChannel != null && cachedChannel.isOpen()) - { - try - { - cachedChannel.close(); - } - catch (Exception e) - { - LargeMessageBufferImpl.log.warn(e.getMessage(), e); - } - cachedChannel = null; - } - - if (cachedRAFile != null) - { - try - { - cachedRAFile.close(); - } - catch (Exception e) - { - LargeMessageBufferImpl.log.warn(e.getMessage(), e); - } - cachedRAFile = null; - } - - } - - @Override - protected void finalize() - { - close(); - if (cachedFile != null && cachedFile.exists()) - { - try - { - cachedFile.delete(); - } - catch (Exception e) - { - LargeMessageBufferImpl.log.warn("Exception during finalization for LargeMessage file cache", e); - } - } - } - - } - - public ChannelBuffer channelBuffer() - { - return null; - } - - public HornetQBuffer copy(final int index, final int length) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public HornetQBuffer duplicate() - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public HornetQBuffer readSlice(final int length) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public void setChar(final int index, final char value) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public void setDouble(final int index, final double value) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public void setFloat(final int index, final float value) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public HornetQBuffer slice() - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - - public void writeBytes(final HornetQBuffer src, final int srcIndex, final int length) - { - throw new IllegalAccessError(LargeMessageBufferImpl.READ_ONLY_ERROR_MESSAGE); - } - -} Deleted: trunk/src/main/org/hornetq/core/client/impl/LargeMessageBufferInternal.java =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/LargeMessageBufferInternal.java 2010-12-07 18:05:17 UTC (rev 10007) +++ trunk/src/main/org/hornetq/core/client/impl/LargeMessageBufferInternal.java 2010-12-07 22:22:35 UTC (rev 10008) @@ -1,67 +0,0 @@ -/* - * Copyright 2009 Red Hat, Inc. - * Red Hat licenses this file to you under the Apache License, version - * 2.0 (the "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * http://www.apache.org/licenses/LICENSE-2.0 - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or - * implied. See the License for the specific language governing - * permissions and limitations under the License. - */ - -package org.hornetq.core.client.impl; - -import java.io.OutputStream; - -import org.hornetq.api.core.HornetQBuffer; -import org.hornetq.api.core.HornetQException; -import org.hornetq.core.protocol.core.impl.wireformat.SessionReceiveContinuationMessage; - -/** - * A LargeMessageBufferInternal - * - * @author <a href="mailto:jmesnil@redhat.com">Jeff Mesnil</a> - */ -public interface LargeMessageBufferInternal extends HornetQBuffer -{ - /** - * Returns the size of this buffer. - - */ - long getSize(); - - /** - * Discards packets unused by this buffer. - */ - void discardUnusedPackets(); - - /** - * Closes this buffer. - */ - void close(); - - /** - * Cancels this buffer. - */ - void cancel(); - - /** - * Sets the OutputStream of this buffer to the specified output. - */ - void setOutputStream(final OutputStream output) throws HornetQException; - - /** - * Saves this buffer to the specified output. - */ - void saveBuffer(final OutputStream output) throws HornetQException; - - public void addPacket(final SessionReceiveContinuationMessage packet); - - /** - * Waits for the completion for the specified waiting time (in milliseconds). - */ - boolean waitCompletion(long timeWait) throws HornetQException; - -} Copied: trunk/src/main/org/hornetq/core/client/impl/LargeMessageController.java (from rev 10006, trunk/src/main/org/hornetq/core/client/impl/LargeMessageBufferInternal.java) =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/LargeMessageController.java (rev 0) +++ trunk/src/main/org/hornetq/core/client/impl/LargeMessageController.java 2010-12-07 22:22:35 UTC (rev 10008) @@ -0,0 +1,67 @@ +/* + * Copyright 2009 Red Hat, Inc. + * Red Hat licenses this file to you under the Apache License, version + * 2.0 (the "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + * implied. See the License for the specific language governing + * permissions and limitations under the License. + */ + +package org.hornetq.core.client.impl; + +import java.io.OutputStream; + +import org.hornetq.api.core.HornetQBuffer; +import org.hornetq.api.core.HornetQException; +import org.hornetq.core.protocol.core.impl.wireformat.SessionReceiveContinuationMessage; + +/** + * A LargeMessageBufferInternal + * + * @author <a href="mailto:jmesnil@redhat.com">Jeff Mesnil</a> + */ +public interface LargeMessageController extends HornetQBuffer +{ + /** + * Returns the size of this buffer. + + */ + long getSize(); + + /** + * Discards packets unused by this buffer. + */ + void discardUnusedPackets(); + + /** + * Closes this buffer. + */ + void close(); + + /** + * Cancels this buffer. + */ + void cancel(); + + /** + * Sets the OutputStream of this buffer to the specified output. + */ + void setOutputStream(final OutputStream output) throws HornetQException; + + /** + * Saves this buffer to the specified output. + */ + void saveBuffer(final OutputStream output) throws HornetQException; + + public void addPacket(final SessionReceiveContinuationMessage packet); + + /** + * Waits for the completion for the specified waiting time (in milliseconds). + */ + boolean waitCompletion(long timeWait) throws HornetQException; + +} Copied: trunk/src/main/org/hornetq/core/client/impl/LargeMessageControllerImpl.java (from rev 10006, trunk/src/main/org/hornetq/core/client/impl/LargeMessageBufferImpl.java) =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/LargeMessageControllerImpl.java (rev 0) +++ trunk/src/main/org/hornetq/core/client/impl/LargeMessageControllerImpl.java 2010-12-07 22:22:35 UTC (rev 10008) @@ -0,0 +1,1535 @@ +/* + * Copyright 2009 Red Hat, Inc. + * Red Hat licenses this file to you under the Apache License, version + * 2.0 (the "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + * implied. See the License for the specific language governing + * permissions and limitations under the License. + */ + +package org.hornetq.core.client.impl; + +import java.io.File; +import java.io.FileNotFoundException; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.io.RandomAccessFile; +import java.nio.ByteBuffer; +import java.nio.channels.FileChannel; +import java.nio.channels.GatheringByteChannel; +import java.nio.channels.ScatteringByteChannel; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; + +import org.hornetq.api.core.HornetQBuffer; +import org.hornetq.api.core.HornetQBuffers; +import org.hornetq.api.core.HornetQException; +import org.hornetq.api.core.SimpleString; +import org.hornetq.core.logging.Logger; +import org.hornetq.core.protocol.core.impl.wireformat.SessionReceiveContinuationMessage; +import org.hornetq.utils.DataConstants; +import org.hornetq.utils.UTF8Util; +import org.jboss.netty.buffer.ChannelBuffer; + +/** + * This class aggregates several SessionReceiveContinuationMessages as it was being handled by a single buffer. + * This buffer can be consumed as messages are arriving, and it will hold the packets until they are read using the ChannelBuffer interface, or the setOutputStream or saveStream are called. + * + * @author <a href="mailto:clebert.suconic@jboss.org">Clebert Suconic</a> + * + * + */ +public class LargeMessageControllerImpl implements LargeMessageController +{ + // Constants ----------------------------------------------------- + + private static final String READ_ONLY_ERROR_MESSAGE = "This is a read-only buffer, setOperations are not supported"; + + // Attributes ---------------------------------------------------- + + private static final Logger log = Logger.getLogger(LargeMessageControllerImpl.class); + + private final ClientConsumerInternal consumerInternal; + + private final LinkedBlockingQueue<SessionReceiveContinuationMessage> packets = new LinkedBlockingQueue<SessionReceiveContinuationMessage>(); + + private volatile SessionReceiveContinuationMessage currentPacket = null; + + private final long totalSize; + + private final int bufferSize; + + private boolean streamEnded = false; + + private boolean streamClosed = false; + + private final int readTimeout; + + private long readerIndex = 0; + + private long packetPosition = -1; + + private long lastIndex = 0; + + private long packetLastPosition = -1; + + private OutputStream outStream; + + private Exception handledException; + + private final FileCache fileCache; + + // Static -------------------------------------------------------- + + // Constructors -------------------------------------------------- + + public LargeMessageControllerImpl(final ClientConsumerInternal consumerInternal, + final long totalSize, + final int readTimeout) + { + this(consumerInternal, totalSize, readTimeout, null); + } + + public LargeMessageControllerImpl(final ClientConsumerInternal consumerInternal, + final long totalSize, + final int readTimeout, + final File cachedFile) + { + this(consumerInternal, totalSize, readTimeout, cachedFile, 10 * 1024); + } + + public LargeMessageControllerImpl(final ClientConsumerInternal consumerInternal, + final long totalSize, + final int readTimeout, + final File cachedFile, + final int bufferSize) + { + this.consumerInternal = consumerInternal; + this.readTimeout = readTimeout; + this.totalSize = totalSize; + if (cachedFile == null) + { + fileCache = null; + } + else + { + fileCache = new FileCache(cachedFile); + } + this.bufferSize = bufferSize; + } + + // Public -------------------------------------------------------- + + public synchronized Exception getHandledException() + { + return handledException; + } + + /** + * + */ + public void discardUnusedPackets() + { + if (outStream == null) + { + try + { + checkForPacket(totalSize - 1); + } + catch (Exception ignored) + { + } + } + } + + /** + * Add a buff to the List, or save it to the OutputStream if set + * @param packet + */ + public void addPacket(final SessionReceiveContinuationMessage packet) + { + int flowControlCredit = 0; + boolean continues = false; + + synchronized (this) + { + if (outStream != null) + { + try + { + if (!packet.isContinues()) + { + streamEnded = true; + } + + if (fileCache != null) + { + fileCache.cachePackage(packet.getBody()); + } + + outStream.write(packet.getBody()); + + flowControlCredit = packet.getPacketSize(); + + continues = packet.isContinues(); + + notifyAll(); + + if (streamEnded) + { + outStream.close(); + } + } + catch (Exception e) + { + LargeMessageControllerImpl.log.warn(e.getMessage(), e); + handledException = e; + } + } + else + { + if (fileCache != null) + { + try + { + fileCache.cachePackage(packet.getBody()); + } + catch (Exception e) + { + LargeMessageControllerImpl.log.warn(e.getMessage(), e); + handledException = e; + } + } + + + packets.offer(packet); + } + } + + if (flowControlCredit != 0) + { + try + { + consumerInternal.flowControl(flowControlCredit, !continues); + } + catch (Exception e) + { + LargeMessageControllerImpl.log.warn(e.getMessage(), e); + handledException = e; + } + } + } + + public synchronized void cancel() + { + packets.offer(new SessionReceiveContinuationMessage()); + streamEnded = true; + streamClosed = true; + + notifyAll(); + } + + public synchronized void close() + { + if (fileCache != null) + { + fileCache.close(); + } + } + + public void setOutputStream(final OutputStream output) throws HornetQException + { + + int totalFlowControl = 0; + boolean continues = false; + + synchronized (this) + { + if (currentPacket != null) + { + sendPacketToOutput(output, currentPacket); + currentPacket = null; + } + while (true) + { + SessionReceiveContinuationMessage packet = packets.poll(); + if (packet == null) + { + break; + } + totalFlowControl += packet.getPacketSize(); + + continues = packet.isContinues(); + sendPacketToOutput(output, packet); + } + + outStream = output; + } + + consumerInternal.flowControl(totalFlowControl, !continues); + } + + public synchronized void saveBuffer(final OutputStream output) throws HornetQException + { + setOutputStream(output); + waitCompletion(0); + } + + /** + * + * @param timeWait Milliseconds to Wait. 0 means forever + * @throws Exception + */ + public synchronized boolean waitCompletion(final long timeWait) throws HornetQException + { + if (outStream == null) + { + // There is no stream.. it will never achieve the end of streaming + return false; + } + + long timeOut = System.currentTimeMillis() + timeWait; + while (!streamEnded && handledException == null) + { + try + { + this.wait(readTimeout == 0 ? 1 : readTimeout * 1000); + } + catch (InterruptedException e) + { + throw new HornetQException(HornetQException.INTERNAL_ERROR, e.getMessage(), e); + } + + if (timeWait > 0 && System.currentTimeMillis() > timeOut) + { + throw new HornetQException(HornetQException.LARGE_MESSAGE_ERROR_BODY, + "Timeout waiting for LargeMessage Body"); + } + } + + if (handledException != null) + { + throw new HornetQException(HornetQException.LARGE_MESSAGE_ERROR_BODY, + "Error on saving LargeMessageBufferImpl", + handledException); + } + + return streamEnded; + + } + + // Channel Buffer Implementation --------------------------------- + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#array() + */ + public byte[] array() + { + throw new IllegalAccessError("array not supported on LargeMessageBufferImpl"); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#capacity() + */ + public int capacity() + { + return -1; + } + + public byte readByte() + { + return getByte(readerIndex++); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#getByte(int) + */ + public byte getByte(final int index) + { + return getByte((long)index); + } + + private byte getByte(final long index) + { + checkForPacket(index); + + if (fileCache != null && index < packetPosition) + { + return fileCache.getByteFromCache(index); + } + else + { + return currentPacket.getBody()[(int)(index - packetPosition)]; + } + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, org.hornetq.api.core.buffers.ChannelBuffer, int, int) + */ + public void getBytes(final int index, final HornetQBuffer dst, final int dstIndex, final int length) + { + byte[] destBytes = new byte[length]; + getBytes(index, destBytes); + dst.setBytes(dstIndex, destBytes); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, org.hornetq.api.core.buffers.ChannelBuffer, int, int) + */ + public void getBytes(final long index, final HornetQBuffer dst, final int dstIndex, final int length) + { + byte[] destBytes = new byte[length]; + getBytes(index, destBytes); + dst.setBytes(dstIndex, destBytes); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, byte[], int, int) + */ + public void getBytes(final int index, final byte[] dst, final int dstIndex, final int length) + { + byte bytesToGet[] = new byte[length]; + + getBytes(index, bytesToGet); + + System.arraycopy(bytesToGet, 0, dst, dstIndex, length); + } + + public void getBytes(final long index, final byte[] dst, final int dstIndex, final int length) + { + byte bytesToGet[] = new byte[length]; + + getBytes(index, bytesToGet); + + System.arraycopy(bytesToGet, 0, dst, dstIndex, length); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, java.nio.ByteBuffer) + */ + public void getBytes(final int index, final ByteBuffer dst) + { + byte bytesToGet[] = new byte[dst.remaining()]; + getBytes(index, bytesToGet); + dst.put(bytesToGet); + } + + public void getBytes(final long index, final ByteBuffer dst) + { + byte bytesToGet[] = new byte[dst.remaining()]; + getBytes(index, bytesToGet); + dst.put(bytesToGet); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, java.io.OutputStream, int) + */ + public void getBytes(final int index, final OutputStream out, final int length) throws IOException + { + byte bytesToGet[] = new byte[length]; + getBytes(index, bytesToGet); + out.write(bytesToGet); + } + + public void getBytes(final long index, final OutputStream out, final int length) throws IOException + { + byte bytesToGet[] = new byte[length]; + getBytes(index, bytesToGet); + out.write(bytesToGet); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, java.nio.channels.GatheringByteChannel, int) + */ + public int getBytes(final int index, final GatheringByteChannel out, final int length) throws IOException + { + byte bytesToGet[] = new byte[length]; + getBytes(index, bytesToGet); + return out.write(ByteBuffer.wrap(bytesToGet)); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#getInt(int) + */ + public int getInt(final int index) + { + return (getByte(index) & 0xff) << 24 | (getByte(index + 1) & 0xff) << 16 | + (getByte(index + 2) & 0xff) << 8 | + (getByte(index + 3) & 0xff) << 0; + } + + public int getInt(final long index) + { + return (getByte(index) & 0xff) << 24 | (getByte(index + 1) & 0xff) << 16 | + (getByte(index + 2) & 0xff) << 8 | + (getByte(index + 3) & 0xff) << 0; + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#getLong(int) + */ + public long getLong(final int index) + { + return ((long)getByte(index) & 0xff) << 56 | ((long)getByte(index + 1) & 0xff) << 48 | + ((long)getByte(index + 2) & 0xff) << 40 | + ((long)getByte(index + 3) & 0xff) << 32 | + ((long)getByte(index + 4) & 0xff) << 24 | + ((long)getByte(index + 5) & 0xff) << 16 | + ((long)getByte(index + 6) & 0xff) << 8 | + ((long)getByte(index + 7) & 0xff) << 0; + } + + public long getLong(final long index) + { + return ((long)getByte(index) & 0xff) << 56 | ((long)getByte(index + 1) & 0xff) << 48 | + ((long)getByte(index + 2) & 0xff) << 40 | + ((long)getByte(index + 3) & 0xff) << 32 | + ((long)getByte(index + 4) & 0xff) << 24 | + ((long)getByte(index + 5) & 0xff) << 16 | + ((long)getByte(index + 6) & 0xff) << 8 | + ((long)getByte(index + 7) & 0xff) << 0; + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#getShort(int) + */ + public short getShort(final int index) + { + return (short)(getByte(index) << 8 | getByte(index + 1) & 0xFF); + } + + public short getShort(final long index) + { + return (short)(getByte(index) << 8 | getByte(index + 1) & 0xFF); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#getUnsignedMedium(int) + */ + public int getUnsignedMedium(final int index) + { + return (getByte(index) & 0xff) << 16 | (getByte(index + 1) & 0xff) << 8 | (getByte(index + 2) & 0xff) << 0; + } + + public int getUnsignedMedium(final long index) + { + return (getByte(index) & 0xff) << 16 | (getByte(index + 1) & 0xff) << 8 | (getByte(index + 2) & 0xff) << 0; + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#setByte(int, byte) + */ + public void setByte(final int index, final byte value) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#setBytes(int, org.hornetq.api.core.buffers.ChannelBuffer, int, int) + */ + public void setBytes(final int index, final HornetQBuffer src, final int srcIndex, final int length) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#setBytes(int, byte[], int, int) + */ + public void setBytes(final int index, final byte[] src, final int srcIndex, final int length) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#setBytes(int, java.nio.ByteBuffer) + */ + public void setBytes(final int index, final ByteBuffer src) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#setBytes(int, java.io.InputStream, int) + */ + public int setBytes(final int index, final InputStream in, final int length) throws IOException + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#setBytes(int, java.nio.channels.ScatteringByteChannel, int) + */ + public int setBytes(final int index, final ScatteringByteChannel in, final int length) throws IOException + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#setInt(int, int) + */ + public void setInt(final int index, final int value) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#setLong(int, long) + */ + public void setLong(final int index, final long value) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#setMedium(int, int) + */ + public void setMedium(final int index, final int value) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#setShort(int, short) + */ + public void setShort(final int index, final short value) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#toByteBuffer(int, int) + */ + public ByteBuffer toByteBuffer(final int index, final int length) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#toString(int, int, java.lang.String) + */ + public String toString(final int index, final int length, final String charsetName) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public int readerIndex() + { + return (int)readerIndex; + } + + public void readerIndex(final int readerIndex) + { + try + { + checkForPacket(readerIndex); + } + catch (Exception e) + { + LargeMessageControllerImpl.log.warn(e.getMessage(), e); + throw new RuntimeException(e.getMessage(), e); + } + this.readerIndex = readerIndex; + } + + public int writerIndex() + { + return (int)totalSize; + } + + public long getSize() + { + return totalSize; + } + + public void writerIndex(final int writerIndex) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public void setIndex(final int readerIndex, final int writerIndex) + { + try + { + checkForPacket(readerIndex); + } + catch (Exception e) + { + LargeMessageControllerImpl.log.warn(e.getMessage(), e); + throw new RuntimeException(e.getMessage(), e); + } + this.readerIndex = readerIndex; + } + + public void clear() + { + } + + public boolean readable() + { + return true; + } + + public boolean writable() + { + return false; + } + + public int readableBytes() + { + long readableBytes = totalSize - readerIndex; + + if (readableBytes > Integer.MAX_VALUE) + { + return Integer.MAX_VALUE; + } + else + { + return (int)(totalSize - readerIndex); + } + } + + public int writableBytes() + { + return 0; + } + + public void markReaderIndex() + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public void resetReaderIndex() + { + try + { + checkForPacket(0); + } + catch (Exception e) + { + LargeMessageControllerImpl.log.warn(e.getMessage(), e); + throw new RuntimeException(e.getMessage(), e); + } + } + + public void markWriterIndex() + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public void resetWriterIndex() + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public void discardReadBytes() + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public short getUnsignedByte(final int index) + { + return (short)(getByte(index) & 0xFF); + } + + public int getUnsignedShort(final int index) + { + return getShort(index) & 0xFFFF; + } + + public int getMedium(final int index) + { + int value = getUnsignedMedium(index); + if ((value & 0x800000) != 0) + { + value |= 0xff000000; + } + return value; + } + + public long getUnsignedInt(final int index) + { + return getInt(index) & 0xFFFFFFFFL; + } + + public void getBytes(int index, final byte[] dst) + { + // TODO: optimize this by using System.arraycopy + for (int i = 0; i < dst.length; i++) + { + dst[i] = getByte(index++); + } + } + + public void getBytes(long index, final byte[] dst) + { + // TODO: optimize this by using System.arraycopy + for (int i = 0; i < dst.length; i++) + { + dst[i] = getByte(index++); + } + } + + public void getBytes(final int index, final HornetQBuffer dst) + { + getBytes(index, dst, dst.writableBytes()); + } + + public void getBytes(final int index, final HornetQBuffer dst, final int length) + { + if (length > dst.writableBytes()) + { + throw new IndexOutOfBoundsException(); + } + getBytes(index, dst, dst.writerIndex(), length); + dst.writerIndex(dst.writerIndex() + length); + } + + public void setBytes(final int index, final byte[] src) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public void setBytes(final int index, final HornetQBuffer src) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public void setBytes(final int index, final HornetQBuffer src, final int length) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public void setZero(final int index, final int length) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public short readUnsignedByte() + { + return (short)(readByte() & 0xFF); + } + + public short readShort() + { + short v = getShort(readerIndex); + readerIndex += 2; + return v; + } + + public int readUnsignedShort() + { + return readShort() & 0xFFFF; + } + + public int readMedium() + { + int value = readUnsignedMedium(); + if ((value & 0x800000) != 0) + { + value |= 0xff000000; + } + return value; + } + + public int readUnsignedMedium() + { + int v = getUnsignedMedium(readerIndex); + readerIndex += 3; + return v; + } + + public int readInt() + { + int v = getInt(readerIndex); + readerIndex += 4; + return v; + } + + public int readInt(final int pos) + { + int v = getInt(pos); + return v; + } + + public long readUnsignedInt() + { + return readInt() & 0xFFFFFFFFL; + } + + public long readLong() + { + long v = getLong(readerIndex); + readerIndex += 8; + return v; + } + + public void readBytes(final byte[] dst, final int dstIndex, final int length) + { + getBytes(readerIndex, dst, dstIndex, length); + readerIndex += length; + } + + public void readBytes(final byte[] dst) + { + readBytes(dst, 0, dst.length); + } + + public void readBytes(final HornetQBuffer dst) + { + readBytes(dst, dst.writableBytes()); + } + + public void readBytes(final HornetQBuffer dst, final int length) + { + if (length > dst.writableBytes()) + { + throw new IndexOutOfBoundsException(); + } + readBytes(dst, dst.writerIndex(), length); + dst.writerIndex(dst.writerIndex() + length); + } + + public void readBytes(final HornetQBuffer dst, final int dstIndex, final int length) + { + getBytes(readerIndex, dst, dstIndex, length); + readerIndex += length; + } + + public void readBytes(final ByteBuffer dst) + { + int length = dst.remaining(); + getBytes(readerIndex, dst); + readerIndex += length; + } + + public int readBytes(final GatheringByteChannel out, final int length) throws IOException + { + int readBytes = getBytes((int)readerIndex, out, length); + readerIndex += readBytes; + return readBytes; + } + + public void readBytes(final OutputStream out, final int length) throws IOException + { + getBytes(readerIndex, out, length); + readerIndex += length; + } + + public void skipBytes(final int length) + { + + long newReaderIndex = readerIndex + length; + checkForPacket(newReaderIndex); + readerIndex = newReaderIndex; + } + + public void writeByte(final byte value) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public void writeShort(final short value) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public void writeMedium(final int value) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public void writeInt(final int value) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public void writeLong(final long value) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public void writeBytes(final byte[] src, final int srcIndex, final int length) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public void writeBytes(final byte[] src) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public void writeBytes(final HornetQBuffer src) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public void writeBytes(final HornetQBuffer src, final int length) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public void writeBytes(final ByteBuffer src) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public int writeBytes(final InputStream in, final int length) throws IOException + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public int writeBytes(final ScatteringByteChannel in, final int length) throws IOException + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public void writeZero(final int length) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public ByteBuffer toByteBuffer() + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public ByteBuffer[] toByteBuffers() + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public ByteBuffer[] toByteBuffers(final int index, final int length) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public String toString(final String charsetName) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public Object getUnderlyingBuffer() + { + return this; + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#readBoolean() + */ + public boolean readBoolean() + { + return readByte() != 0; + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#readChar() + */ + public char readChar() + { + return (char)readShort(); + } + + public char getChar(final int index) + { + return (char)getShort(index); + } + + public double getDouble(final int index) + { + return Double.longBitsToDouble(getLong(index)); + } + + public float getFloat(final int index) + { + return Float.intBitsToFloat(getInt(index)); + } + + public HornetQBuffer readBytes(final int length) + { + byte bytesToGet[] = new byte[length]; + getBytes(readerIndex, bytesToGet); + readerIndex += length; + return HornetQBuffers.wrappedBuffer(bytesToGet); + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#readDouble() + */ + public double readDouble() + { + return Double.longBitsToDouble(readLong()); + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#readFloat() + */ + public float readFloat() + { + return Float.intBitsToFloat(readInt()); + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#readNullableSimpleString() + */ + public SimpleString readNullableSimpleString() + { + int b = readByte(); + if (b == DataConstants.NULL) + { + return null; + } + else + { + return readSimpleString(); + } + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#readNullableString() + */ + public String readNullableString() + { + int b = readByte(); + if (b == DataConstants.NULL) + { + return null; + } + else + { + return readString(); + } + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#readSimpleString() + */ + public SimpleString readSimpleString() + { + int len = readInt(); + byte[] data = new byte[len]; + readBytes(data); + return new SimpleString(data); + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#readString() + */ + public String readString() + { + int len = readInt(); + + if (len < 9) + { + char[] chars = new char[len]; + for (int i = 0; i < len; i++) + { + chars[i] = (char)readShort(); + } + return new String(chars); + } + else if (len < 0xfff) + { + return readUTF(); + } + else + { + return readSimpleString().toString(); + } + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#readUTF() + */ + public String readUTF() + { + return UTF8Util.readUTF(this); + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeBoolean(boolean) + */ + public void writeBoolean(final boolean val) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeChar(char) + */ + public void writeChar(final char val) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeDouble(double) + */ + public void writeDouble(final double val) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeFloat(float) + */ + public void writeFloat(final float val) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeNullableSimpleString(org.hornetq.util.SimpleString) + */ + public void writeNullableSimpleString(final SimpleString val) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeNullableString(java.lang.String) + */ + public void writeNullableString(final String val) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeSimpleString(org.hornetq.util.SimpleString) + */ + public void writeSimpleString(final SimpleString val) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeString(java.lang.String) + */ + public void writeString(final String val) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + /* (non-Javadoc) + * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeUTF(java.lang.String) + */ + public void writeUTF(final String utf) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + /* (non-Javadoc) + * @see org.hornetq.api.core.buffers.ChannelBuffer#compareTo(org.hornetq.api.core.buffers.ChannelBuffer) + */ + public int compareTo(final HornetQBuffer buffer) + { + return -1; + } + + public HornetQBuffer copy() + { + throw new UnsupportedOperationException(); + } + + public HornetQBuffer slice(final int index, final int length) + { + throw new UnsupportedOperationException(); + } + + // Package protected --------------------------------------------- + + // Protected ----------------------------------------------------- + + // Private ------------------------------------------------------- + + /** + * @param output + * @param packet + * @throws HornetQException + */ + private void sendPacketToOutput(final OutputStream output, final SessionReceiveContinuationMessage packet) throws HornetQException + { + try + { + output.write(packet.getBody()); + if (!packet.isContinues()) + { + streamEnded = true; + output.close(); + } + } + catch (IOException e) + { + throw new HornetQException(HornetQException.LARGE_MESSAGE_ERROR_BODY, "Error writing body of message", e); + } + } + + private void popPacket() + { + try + { + + if (streamEnded) + { + // no more packets, we are over the last one already + throw new IndexOutOfBoundsException(); + } + + int sizeToAdd = currentPacket != null ? currentPacket.getBody().length : 1; + currentPacket = packets.poll(readTimeout, TimeUnit.SECONDS); + if (currentPacket == null) + { + throw new IndexOutOfBoundsException(); + } + + if (currentPacket.getBody() == null) // Empty packet as a signal to interruption + { + currentPacket = null; + streamEnded = true; + throw new IndexOutOfBoundsException(); + } + + consumerInternal.flowControl(currentPacket.getPacketSize(), !currentPacket.isContinues()); + + packetPosition += sizeToAdd; + + packetLastPosition = packetPosition + currentPacket.getBody().length; + } + catch (IndexOutOfBoundsException e) + { + throw e; + } + catch (Exception e) + { + throw new RuntimeException(e); + } + } + + private void checkForPacket(final long index) + { + if (outStream != null) + { + throw new IllegalAccessError("Can't read the messageBody after setting outputStream"); + } + + if (index >= totalSize) + { + throw new IndexOutOfBoundsException(); + } + + if (streamClosed) + { + throw new IllegalAccessError("The consumer associated with this large message was closed before the body was read"); + } + + if (fileCache == null) + { + if (index < lastIndex) + { + throw new IllegalAccessError("LargeMessage have read-only and one-way buffers"); + } + lastIndex = index; + } + + while (index >= packetLastPosition && !streamEnded) + { + popPacket(); + } + } + + /** + * @param body + */ + // Inner classes ------------------------------------------------- + private class FileCache + { + + public FileCache(final File cachedFile) + { + this.cachedFile = cachedFile; + } + + ByteBuffer readCache; + + long readCachePositionStart = Integer.MAX_VALUE; + + long readCachePositionEnd = -1; + + private final File cachedFile; + + private volatile RandomAccessFile cachedRAFile; + + private volatile FileChannel cachedChannel; + + private synchronized void readCache(final long position) + { + + try + { + if (position < readCachePositionStart || position > readCachePositionEnd) + { + + checkOpen(); + + if (position > cachedChannel.size()) + { + throw new ArrayIndexOutOfBoundsException("position > " + cachedChannel.size()); + } + + readCachePositionStart = position / bufferSize * bufferSize; + + cachedChannel.position(readCachePositionStart); + + if (readCache == null) + { + readCache = ByteBuffer.allocate(bufferSize); + } + + readCache.clear(); + + readCachePositionEnd = readCachePositionStart + cachedChannel.read(readCache) - 1; + } + } + catch (Exception e) + { + LargeMessageControllerImpl.log.warn(e.getMessage(), e); + throw new RuntimeException(e.getMessage(), e); + } + finally + { + close(); + } + } + + public synchronized byte getByteFromCache(final long position) + { + readCache(position); + + return readCache.get((int)(position - readCachePositionStart)); + + } + + public void cachePackage(final byte[] body) throws Exception + { + checkOpen(); + + cachedChannel.position(cachedChannel.size()); + cachedChannel.write(ByteBuffer.wrap(body)); + + close(); + } + + /** + * @throws FileNotFoundException + */ + public void checkOpen() throws FileNotFoundException + { + if (cachedFile != null || !cachedChannel.isOpen()) + { + cachedRAFile = new RandomAccessFile(cachedFile, "rw"); + + cachedChannel = cachedRAFile.getChannel(); + } + } + + public void close() + { + if (cachedChannel != null && cachedChannel.isOpen()) + { + try + { + cachedChannel.close(); + } + catch (Exception e) + { + LargeMessageControllerImpl.log.warn(e.getMessage(), e); + } + cachedChannel = null; + } + + if (cachedRAFile != null) + { + try + { + cachedRAFile.close(); + } + catch (Exception e) + { + LargeMessageControllerImpl.log.warn(e.getMessage(), e); + } + cachedRAFile = null; + } + + } + + @Override + protected void finalize() + { + close(); + if (cachedFile != null && cachedFile.exists()) + { + try + { + cachedFile.delete(); + } + catch (Exception e) + { + LargeMessageControllerImpl.log.warn("Exception during finalization for LargeMessage file cache", e); + } + } + } + + } + + public ChannelBuffer channelBuffer() + { + return null; + } + + public HornetQBuffer copy(final int index, final int length) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public HornetQBuffer duplicate() + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public HornetQBuffer readSlice(final int length) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public void setChar(final int index, final char value) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public void setDouble(final int index, final double value) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public void setFloat(final int index, final float value) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public HornetQBuffer slice() + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + + public void writeBytes(final HornetQBuffer src, final int srcIndex, final int length) + { + throw new IllegalAccessError(LargeMessageControllerImpl.READ_ONLY_ERROR_MESSAGE); + } + +} Modified: trunk/src/main/org/hornetq/core/message/impl/MessageImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/message/impl/MessageImpl.java 2010-12-07 18:05:17 UTC (rev 10007) +++ trunk/src/main/org/hornetq/core/message/impl/MessageImpl.java 2010-12-07 22:22:35 UTC (rev 10008) @@ -24,7 +24,7 @@ import org.hornetq.api.core.PropertyConversionException; import org.hornetq.api.core.SimpleString; import org.hornetq.core.buffers.impl.ResetLimitWrappedHornetQBuffer; -import org.hornetq.core.client.impl.LargeMessageBufferInternal; +import org.hornetq.core.client.impl.LargeMessageController; import org.hornetq.core.logging.Logger; import org.hornetq.core.message.BodyEncoder; import org.hornetq.core.protocol.core.impl.PacketImpl; @@ -245,11 +245,24 @@ properties.decode(buffer); } + public void copyHeadersAndProperties(final MessageInternal msg) + { + messageID = msg.getMessageID(); + address = msg.getAddress(); + userID = msg.getUserID(); + type = msg.getType(); + durable = msg.isDurable(); + expiration = msg.getExpiration(); + timestamp = msg.getTimestamp(); + priority = msg.getPriority(); + properties = msg.getTypedProperties(); + } + public HornetQBuffer getBodyBuffer() { if (bodyBuffer == null) { - if (buffer instanceof LargeMessageBufferInternal == false) + if (buffer instanceof LargeMessageController == false) { bodyBuffer = new ResetLimitWrappedHornetQBuffer(BODY_OFFSET, buffer, this); } @@ -845,6 +858,11 @@ { return new DecodingContext(); } + + public TypedProperties getTypedProperties() + { + return this.properties; + } // Public -------------------------------------------------------- Modified: trunk/src/main/org/hornetq/core/message/impl/MessageInternal.java =================================================================== --- trunk/src/main/org/hornetq/core/message/impl/MessageInternal.java 2010-12-07 18:05:17 UTC (rev 10007) +++ trunk/src/main/org/hornetq/core/message/impl/MessageInternal.java 2010-12-07 22:22:35 UTC (rev 10008) @@ -20,6 +20,7 @@ import org.hornetq.api.core.Message; import org.hornetq.api.core.SimpleString; import org.hornetq.core.message.BodyEncoder; +import org.hornetq.utils.TypedProperties; /** * A MessageInternal @@ -61,4 +62,6 @@ InputStream getBodyInputStream(); void setAddressTransient(SimpleString address); + + TypedProperties getTypedProperties(); } Modified: trunk/src/main/org/hornetq/core/persistence/StorageManager.java =================================================================== --- trunk/src/main/org/hornetq/core/persistence/StorageManager.java 2010-12-07 18:05:17 UTC (rev 10007) +++ trunk/src/main/org/hornetq/core/persistence/StorageManager.java 2010-12-07 22:22:35 UTC (rev 10008) @@ -23,6 +23,7 @@ import org.hornetq.api.core.SimpleString; import org.hornetq.core.journal.IOAsyncTask; import org.hornetq.core.journal.JournalLoadInformation; +import org.hornetq.core.message.impl.MessageInternal; import org.hornetq.core.paging.PageTransactionInfo; import org.hornetq.core.paging.PagedMessage; import org.hornetq.core.paging.PagingManager; @@ -127,7 +128,14 @@ LargeServerMessage createLargeMessage(); - LargeServerMessage createLargeMessage(long id, byte[] header); + /** + * + * @param id + * @param message This is a temporary message that holds the parsed properties. + * The remoting layer can't create a ServerMessage directly, then this will be replaced. + * @return + */ + LargeServerMessage createLargeMessage(long id, MessageInternal message); void prepare(long txID, Xid xid) throws Exception; Modified: trunk/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java =================================================================== --- trunk/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java 2010-12-07 18:05:17 UTC (rev 10007) +++ trunk/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java 2010-12-07 22:22:35 UTC (rev 10008) @@ -47,6 +47,7 @@ import org.hornetq.core.journal.impl.JournalImpl; import org.hornetq.core.journal.impl.NIOSequentialFileFactory; import org.hornetq.core.logging.Logger; +import org.hornetq.core.message.impl.MessageInternal; import org.hornetq.core.paging.PageTransactionInfo; import org.hornetq.core.paging.PagedMessage; import org.hornetq.core.paging.PagingManager; @@ -427,7 +428,7 @@ } } - public LargeServerMessage createLargeMessage(final long id, final byte[] header) + public LargeServerMessage createLargeMessage(final long id, final MessageInternal message) { if (isReplicated()) { @@ -435,11 +436,9 @@ } LargeServerMessageImpl largeMessage = (LargeServerMessageImpl)createLargeMessage(); + + largeMessage.copyHeadersAndProperties(message); - HornetQBuffer headerBuffer = HornetQBuffers.wrappedBuffer(header); - - largeMessage.decodeHeadersAndProperties(headerBuffer); - largeMessage.setMessageID(id); return largeMessage; Modified: trunk/src/main/org/hornetq/core/persistence/impl/nullpm/NullStorageManager.java =================================================================== --- trunk/src/main/org/hornetq/core/persistence/impl/nullpm/NullStorageManager.java 2010-12-07 18:05:17 UTC (rev 10007) +++ trunk/src/main/org/hornetq/core/persistence/impl/nullpm/NullStorageManager.java 2010-12-07 22:22:35 UTC (rev 10008) @@ -27,6 +27,7 @@ import org.hornetq.api.core.SimpleString; import org.hornetq.core.journal.IOAsyncTask; import org.hornetq.core.journal.JournalLoadInformation; +import org.hornetq.core.message.impl.MessageInternal; import org.hornetq.core.paging.PageTransactionInfo; import org.hornetq.core.paging.PagedMessage; import org.hornetq.core.paging.PagingManager; @@ -187,14 +188,12 @@ return new NullStorageLargeServerMessage(); } - public LargeServerMessage createLargeMessage(final long id, final byte[] header) + public LargeServerMessage createLargeMessage(final long id, final MessageInternal message) { NullStorageLargeServerMessage largeMessage = new NullStorageLargeServerMessage(); + + largeMessage.copyHeadersAndProperties(message); - HornetQBuffer headerBuffer = HornetQBuffers.wrappedBuffer(header); - - largeMessage.decodeHeadersAndProperties(headerBuffer); - largeMessage.setMessageID(id); return largeMessage; Modified: trunk/src/main/org/hornetq/core/protocol/core/ServerSessionPacketHandler.java =================================================================== --- trunk/src/main/org/hornetq/core/protocol/core/ServerSessionPacketHandler.java 2010-12-07 18:05:17 UTC (rev 10007) +++ trunk/src/main/org/hornetq/core/protocol/core/ServerSessionPacketHandler.java 2010-12-07 22:22:35 UTC (rev 10008) @@ -451,7 +451,7 @@ case SESS_SEND_LARGE: { SessionSendLargeMessage message = (SessionSendLargeMessage)packet; - session.sendLarge(message.getLargeMessageHeader()); + session.sendLarge(message.getLargeMessage()); break; } case SESS_SEND_CONTINUATION: Modified: trunk/src/main/org/hornetq/core/protocol/core/impl/CoreSessionCallback.java =================================================================== --- trunk/src/main/org/hornetq/core/protocol/core/impl/CoreSessionCallback.java 2010-12-07 18:05:17 UTC (rev 10007) +++ trunk/src/main/org/hornetq/core/protocol/core/impl/CoreSessionCallback.java 2010-12-07 22:22:35 UTC (rev 10008) @@ -49,9 +49,9 @@ this.channel = channel; } - public int sendLargeMessage(long consumerID, byte[] headerBuffer, long bodySize, int deliveryCount) + public int sendLargeMessage(ServerMessage message, long consumerID, long bodySize, int deliveryCount) { - Packet packet = new SessionReceiveLargeMessage(consumerID, headerBuffer, bodySize, deliveryCount); + Packet packet = new SessionReceiveLargeMessage(consumerID, message, bodySize, deliveryCount); channel.send(packet); Modified: trunk/src/main/org/hornetq/core/protocol/core/impl/PacketDecoder.java =================================================================== --- trunk/src/main/org/hornetq/core/protocol/core/impl/PacketDecoder.java 2010-12-07 18:05:17 UTC (rev 10007) +++ trunk/src/main/org/hornetq/core/protocol/core/impl/PacketDecoder.java 2010-12-07 22:22:35 UTC (rev 10008) @@ -84,7 +84,9 @@ import static org.hornetq.core.protocol.core.impl.PacketImpl.SESS_ADD_METADATA; import org.hornetq.api.core.HornetQBuffer; +import org.hornetq.core.client.impl.ClientMessageImpl; import org.hornetq.core.logging.Logger; +import org.hornetq.core.persistence.StorageManager; import org.hornetq.core.protocol.core.Packet; import org.hornetq.core.protocol.core.impl.wireformat.ClusterTopologyChangeMessage; import org.hornetq.core.protocol.core.impl.wireformat.CreateQueueMessage; @@ -161,7 +163,7 @@ public class PacketDecoder { private static final Logger log = Logger.getLogger(PacketDecoder.class); - + public Packet decode(final HornetQBuffer in) { final byte packetType = in.readByte(); @@ -372,6 +374,8 @@ } case SESS_SEND_LARGE: { + // Using a ClientMessage, but that will be replaced later.. + // This is just to avoid reading a byte[] to read the message packet = new SessionSendLargeMessage(); break; } Modified: trunk/src/main/org/hornetq/core/protocol/core/impl/wireformat/SessionReceiveLargeMessage.java =================================================================== --- trunk/src/main/org/hornetq/core/protocol/core/impl/wireformat/SessionReceiveLargeMessage.java 2010-12-07 18:05:17 UTC (rev 10007) +++ trunk/src/main/org/hornetq/core/protocol/core/impl/wireformat/SessionReceiveLargeMessage.java 2010-12-07 22:22:35 UTC (rev 10008) @@ -14,7 +14,9 @@ package org.hornetq.core.protocol.core.impl.wireformat; import org.hornetq.api.core.HornetQBuffer; +import org.hornetq.core.client.impl.ClientMessageImpl; import org.hornetq.core.logging.Logger; +import org.hornetq.core.message.impl.MessageInternal; import org.hornetq.core.protocol.core.impl.PacketImpl; /** @@ -28,7 +30,7 @@ { private static final Logger log = Logger.getLogger(SessionReceiveLargeMessage.class); - private byte[] largeMessageHeader; + private MessageInternal message; /** Since we receive the message before the entire message was received, */ private long largeMessageSize; @@ -37,13 +39,15 @@ private int deliveryCount; + // To be used on decoding at the client while receiving a large message public SessionReceiveLargeMessage() { super(PacketImpl.SESS_RECEIVE_LARGE_MSG); + this.message = new ClientMessageImpl(); } public SessionReceiveLargeMessage(final long consumerID, - final byte[] largeMessageHeader, + final MessageInternal message, final long largeMessageSize, final int deliveryCount) { @@ -51,16 +55,16 @@ this.consumerID = consumerID; - this.largeMessageHeader = largeMessageHeader; + this.message = message; this.deliveryCount = deliveryCount; this.largeMessageSize = largeMessageSize; } - public byte[] getLargeMessageHeader() + public MessageInternal getLargeMessage() { - return largeMessageHeader; + return message; } public long getConsumerID() @@ -87,8 +91,7 @@ buffer.writeLong(consumerID); buffer.writeInt(deliveryCount); buffer.writeLong(largeMessageSize); - buffer.writeInt(largeMessageHeader.length); - buffer.writeBytes(largeMessageHeader); + message.encodeHeadersAndProperties(buffer); } @Override @@ -97,9 +100,7 @@ consumerID = buffer.readLong(); deliveryCount = buffer.readInt(); largeMessageSize = buffer.readLong(); - int size = buffer.readInt(); - largeMessageHeader = new byte[size]; - buffer.readBytes(largeMessageHeader); + message.decodeHeadersAndProperties(buffer); } } Modified: trunk/src/main/org/hornetq/core/protocol/core/impl/wireformat/SessionSendLargeMessage.java =================================================================== --- trunk/src/main/org/hornetq/core/protocol/core/impl/wireformat/SessionSendLargeMessage.java 2010-12-07 18:05:17 UTC (rev 10007) +++ trunk/src/main/org/hornetq/core/protocol/core/impl/wireformat/SessionSendLargeMessage.java 2010-12-07 22:22:35 UTC (rev 10008) @@ -14,7 +14,9 @@ package org.hornetq.core.protocol.core.impl.wireformat; import org.hornetq.api.core.HornetQBuffer; +import org.hornetq.core.message.impl.MessageInternal; import org.hornetq.core.protocol.core.impl.PacketImpl; +import org.hornetq.core.server.impl.ServerMessageImpl; /** * @author <a href="mailto:tim.fox@jboss.com">Tim Fox</a> @@ -30,46 +32,42 @@ // Attributes ---------------------------------------------------- /** Used only if largeMessage */ - private byte[] largeMessageHeader; + private MessageInternal largeMessage; // Static -------------------------------------------------------- // Constructors -------------------------------------------------- - - public SessionSendLargeMessage(final byte[] largeMessageHeader) + + // To be used by the PacketDecoder + public SessionSendLargeMessage() { - super(PacketImpl.SESS_SEND_LARGE); - - this.largeMessageHeader = largeMessageHeader; + this(new ServerMessageImpl()); } - public SessionSendLargeMessage() + public SessionSendLargeMessage(final MessageInternal largeMessage) { super(PacketImpl.SESS_SEND_LARGE); + + this.largeMessage = largeMessage; } // Public -------------------------------------------------------- - public byte[] getLargeMessageHeader() + public MessageInternal getLargeMessage() { - return largeMessageHeader; + return largeMessage; } @Override public void encodeRest(final HornetQBuffer buffer) { - buffer.writeInt(largeMessageHeader.length); - buffer.writeBytes(largeMessageHeader); + largeMessage.encodeHeadersAndProperties(buffer); } @Override public void decodeRest(final HornetQBuffer buffer) { - int largeMessageLength = buffer.readInt(); - - largeMessageHeader = new byte[largeMessageLength]; - - buffer.readBytes(largeMessageHeader); + largeMessage.decodeHeadersAndProperties(buffer); } // Package protected --------------------------------------------- Modified: trunk/src/main/org/hornetq/core/protocol/core/impl/wireformat/SessionSendMessage.java =================================================================== --- trunk/src/main/org/hornetq/core/protocol/core/impl/wireformat/SessionSendMessage.java 2010-12-07 18:05:17 UTC (rev 10007) +++ trunk/src/main/org/hornetq/core/protocol/core/impl/wireformat/SessionSendMessage.java 2010-12-07 22:22:35 UTC (rev 10008) @@ -14,7 +14,6 @@ package org.hornetq.core.protocol.core.impl.wireformat; import org.hornetq.api.core.HornetQBuffer; -import org.hornetq.api.core.Message; import org.hornetq.core.logging.Logger; import org.hornetq.core.message.impl.MessageInternal; import org.hornetq.core.protocol.core.impl.PacketImpl; Modified: trunk/src/main/org/hornetq/core/protocol/stomp/StompSession.java =================================================================== --- trunk/src/main/org/hornetq/core/protocol/stomp/StompSession.java 2010-12-07 18:05:17 UTC (rev 10007) +++ trunk/src/main/org/hornetq/core/protocol/stomp/StompSession.java 2010-12-07 22:22:35 UTC (rev 10008) @@ -149,7 +149,7 @@ return 0; } - public int sendLargeMessage(long consumerID, byte[] headerBuffer, long bodySize, int deliveryCount) + public int sendLargeMessage(ServerMessage msg, long consumerID, long bodySize, int deliveryCount) { return 0; } Modified: trunk/src/main/org/hornetq/core/server/ServerSession.java =================================================================== --- trunk/src/main/org/hornetq/core/server/ServerSession.java 2010-12-07 18:05:17 UTC (rev 10007) +++ trunk/src/main/org/hornetq/core/server/ServerSession.java 2010-12-07 22:22:35 UTC (rev 10008) @@ -19,6 +19,7 @@ import javax.transaction.xa.Xid; import org.hornetq.api.core.SimpleString; +import org.hornetq.core.message.impl.MessageInternal; /** * @@ -102,7 +103,7 @@ void send(ServerMessage message, boolean direct) throws Exception; - void sendLarge(byte[] largeMessageHeader) throws Exception; + void sendLarge(MessageInternal msg) throws Exception; void forceConsumerDelivery(long consumerID, long sequence) throws Exception; Modified: trunk/src/main/org/hornetq/core/server/impl/ServerConsumerImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/ServerConsumerImpl.java 2010-12-07 18:05:17 UTC (rev 10007) +++ trunk/src/main/org/hornetq/core/server/impl/ServerConsumerImpl.java 2010-12-07 22:22:35 UTC (rev 10008) @@ -768,10 +768,6 @@ if (!sentInitialPacket) { - HornetQBuffer headerBuffer = HornetQBuffers.fixedBuffer(largeMessage.getHeadersAndPropertiesEncodeSize()); - - largeMessage.encodeHeadersAndProperties(headerBuffer); - context = largeMessage.getBodyEncoder(); sizePendingLargeMessage = context.getLargeBodySize(); @@ -780,8 +776,8 @@ sentInitialPacket = true; - int packetSize = callback.sendLargeMessage(id, - headerBuffer.toByteBuffer().array(), + int packetSize = callback.sendLargeMessage(largeMessage, + id, context.getLargeBodySize(), ref.getDeliveryCount()); Modified: trunk/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java 2010-12-07 18:05:17 UTC (rev 10007) +++ trunk/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java 2010-12-07 22:22:35 UTC (rev 10008) @@ -37,6 +37,7 @@ import org.hornetq.core.filter.impl.FilterImpl; import org.hornetq.core.journal.IOAsyncTask; import org.hornetq.core.logging.Logger; +import org.hornetq.core.message.impl.MessageInternal; import org.hornetq.core.paging.PagingStore; import org.hornetq.core.persistence.StorageManager; import org.hornetq.core.postoffice.Binding; @@ -965,19 +966,19 @@ consumer.receiveCredits(credits); } - public void sendLarge(final byte[] largeMessageHeader) throws Exception + public void sendLarge(final MessageInternal message) throws Exception { // need to create the LargeMessage before continue long id = storageManager.generateUniqueID(); - - LargeServerMessage msg = storageManager.createLargeMessage(id, largeMessageHeader); - + + LargeServerMessage largeMsg = storageManager.createLargeMessage(id, message); + if (currentLargeMessage != null) { ServerSessionImpl.log.warn("Replacing incomplete LargeMessage with ID=" + currentLargeMessage.getMessageID()); } - currentLargeMessage = msg; + currentLargeMessage = largeMsg; } public void send(final ServerMessage message, final boolean direct) throws Exception Modified: trunk/src/main/org/hornetq/spi/core/protocol/SessionCallback.java =================================================================== --- trunk/src/main/org/hornetq/spi/core/protocol/SessionCallback.java 2010-12-07 18:05:17 UTC (rev 10007) +++ trunk/src/main/org/hornetq/spi/core/protocol/SessionCallback.java 2010-12-07 22:22:35 UTC (rev 10008) @@ -30,7 +30,7 @@ int sendMessage(ServerMessage message, long consumerID, int deliveryCount); - int sendLargeMessage(long consumerID, byte[] headerBuffer, long bodySize, int deliveryCount); + int sendLargeMessage(ServerMessage message, long consumerID, long bodySize, int deliveryCount); int sendLargeMessageContinuation(long consumerID, byte[] body, boolean continues, boolean requiresResponse); Deleted: trunk/src/main/org/hornetq/utils/DecompressedLargeMessageBuffer.java =================================================================== --- trunk/src/main/org/hornetq/utils/DecompressedLargeMessageBuffer.java 2010-12-07 18:05:17 UTC (rev 10007) +++ trunk/src/main/org/hornetq/utils/DecompressedLargeMessageBuffer.java 2010-12-07 22:22:35 UTC (rev 10008) @@ -1,1091 +0,0 @@ -/* - * Copyright 2010 Red Hat, Inc. - * Red Hat licenses this file to you under the Apache License, version - * 2.0 (the "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * http://www.apache.org/licenses/LICENSE-2.0 - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or - * implied. See the License for the specific language governing - * permissions and limitations under the License. - */ - -package org.hornetq.utils; - -import java.io.DataInputStream; -import java.io.IOException; -import java.io.InputStream; -import java.io.OutputStream; -import java.nio.ByteBuffer; -import java.nio.channels.GatheringByteChannel; -import java.nio.channels.ScatteringByteChannel; -import java.util.concurrent.Executor; - -import org.hornetq.api.core.HornetQBuffer; -import org.hornetq.api.core.HornetQBuffers; -import org.hornetq.api.core.HornetQException; -import org.hornetq.api.core.SimpleString; -import org.hornetq.core.client.impl.LargeMessageBufferInternal; -import org.hornetq.core.protocol.core.impl.wireformat.SessionReceiveContinuationMessage; -import org.jboss.netty.buffer.ChannelBuffer; - -/** - * A DecompressedHornetQBuffer - * - * @author <a href="mailto:clebert.suconic@jboss.org">Clebert Suconic</a> - * - * - */ -public class DecompressedLargeMessageBuffer implements LargeMessageBufferInternal -{ - - // Constants ----------------------------------------------------- - - private static final String OPERATION_NOT_SUPPORTED = "Operation not supported"; - - private static final String READ_ONLY_ERROR_MESSAGE = "This is a read-only buffer, setOperations are not supported"; - - // Attributes ---------------------------------------------------- - - final LargeMessageBufferInternal bufferDelegate; - - // Static -------------------------------------------------------- - - // Constructors -------------------------------------------------- - - public DecompressedLargeMessageBuffer(final LargeMessageBufferInternal bufferDelegate) - { - this.bufferDelegate = bufferDelegate; - } - - - // Public -------------------------------------------------------- - - /** - * - */ - public void discardUnusedPackets() - { - bufferDelegate.discardUnusedPackets(); - } - - /** - * Add a buff to the List, or save it to the OutputStream if set - * @param packet - */ - public void addPacket(final SessionReceiveContinuationMessage packet) - { - bufferDelegate.addPacket(packet); - } - - public synchronized void cancel() - { - bufferDelegate.cancel(); - } - - public synchronized void close() - { - bufferDelegate.cancel(); - } - - public void setOutputStream(final OutputStream output) throws HornetQException - { - bufferDelegate.setOutputStream(new InflaterWriter(output)); - } - - public synchronized void saveBuffer(final OutputStream output) throws HornetQException - { - setOutputStream(output); - waitCompletion(0); - } - - /** - * - * @param timeWait Milliseconds to Wait. 0 means forever - * @throws Exception - */ - public synchronized boolean waitCompletion(final long timeWait) throws HornetQException - { - return bufferDelegate.waitCompletion(timeWait); - } - - // Channel Buffer Implementation --------------------------------- - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#array() - */ - public byte[] array() - { - throw new IllegalAccessError("array not supported on LargeMessageBufferImpl"); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#capacity() - */ - public int capacity() - { - return -1; - } - - DataInputStream dataInput = null; - - private DataInputStream getStream() - { - if (dataInput == null) - { - try - { - InputStream input = new HornetQBufferInputStream(bufferDelegate); - - dataInput = new DataInputStream(new InflaterReader(input)); - } - catch (Exception e) - { - throw new RuntimeException (e.getMessage(), e); - } - - } - return dataInput; - } - - private void positioningNotSupported() - { - throw new IllegalStateException("Position not supported over compressed large messages"); - } - - public byte readByte() - { - try - { - return getStream().readByte(); - } - catch (Exception e) - { - throw new RuntimeException (e.getMessage(), e); - } - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#getByte(int) - */ - public byte getByte(final int index) - { - positioningNotSupported(); - return 0; - } - - private byte getByte(final long index) - { - positioningNotSupported(); - return 0; - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, org.hornetq.api.core.buffers.ChannelBuffer, int, int) - */ - public void getBytes(final int index, final HornetQBuffer dst, final int dstIndex, final int length) - { - positioningNotSupported(); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, org.hornetq.api.core.buffers.ChannelBuffer, int, int) - */ - public void getBytes(final long index, final HornetQBuffer dst, final int dstIndex, final int length) - { - positioningNotSupported(); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, byte[], int, int) - */ - public void getBytes(final int index, final byte[] dst, final int dstIndex, final int length) - { - positioningNotSupported(); - } - - public void getBytes(final long index, final byte[] dst, final int dstIndex, final int length) - { - positioningNotSupported(); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, java.nio.ByteBuffer) - */ - public void getBytes(final int index, final ByteBuffer dst) - { - positioningNotSupported(); - } - - public void getBytes(final long index, final ByteBuffer dst) - { - positioningNotSupported(); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, java.io.OutputStream, int) - */ - public void getBytes(final int index, final OutputStream out, final int length) throws IOException - { - positioningNotSupported(); - } - - public void getBytes(final long index, final OutputStream out, final int length) throws IOException - { - positioningNotSupported(); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#getBytes(int, java.nio.channels.GatheringByteChannel, int) - */ - public int getBytes(final int index, final GatheringByteChannel out, final int length) throws IOException - { - positioningNotSupported(); - return 0; - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#getInt(int) - */ - public int getInt(final int index) - { - positioningNotSupported(); - return 0; - } - - public int getInt(final long index) - { - positioningNotSupported(); - return 0; - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#getLong(int) - */ - public long getLong(final int index) - { - positioningNotSupported(); - return 0; - } - - public long getLong(final long index) - { - positioningNotSupported(); - return 0; - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#getShort(int) - */ - public short getShort(final int index) - { - positioningNotSupported(); - return 0; - } - - public short getShort(final long index) - { - return (short)(getByte(index) << 8 | getByte(index + 1) & 0xFF); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#getUnsignedMedium(int) - */ - public int getUnsignedMedium(final int index) - { - positioningNotSupported(); - return 0; - } - - - - public int getUnsignedMedium(final long index) - { - positioningNotSupported(); - return 0; - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#setByte(int, byte) - */ - public void setByte(final int index, final byte value) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#setBytes(int, org.hornetq.api.core.buffers.ChannelBuffer, int, int) - */ - public void setBytes(final int index, final HornetQBuffer src, final int srcIndex, final int length) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#setBytes(int, byte[], int, int) - */ - public void setBytes(final int index, final byte[] src, final int srcIndex, final int length) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#setBytes(int, java.nio.ByteBuffer) - */ - public void setBytes(final int index, final ByteBuffer src) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#setBytes(int, java.io.InputStream, int) - */ - public int setBytes(final int index, final InputStream in, final int length) throws IOException - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#setBytes(int, java.nio.channels.ScatteringByteChannel, int) - */ - public int setBytes(final int index, final ScatteringByteChannel in, final int length) throws IOException - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#setInt(int, int) - */ - public void setInt(final int index, final int value) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#setLong(int, long) - */ - public void setLong(final int index, final long value) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#setMedium(int, int) - */ - public void setMedium(final int index, final int value) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#setShort(int, short) - */ - public void setShort(final int index, final short value) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#toByteBuffer(int, int) - */ - public ByteBuffer toByteBuffer(final int index, final int length) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#toString(int, int, java.lang.String) - */ - public String toString(final int index, final int length, final String charsetName) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public int readerIndex() - { - return 0; - } - - public void readerIndex(final int readerIndex) - { - // TODO - } - - public int writerIndex() - { - // TODO - return 0; - } - - public long getSize() - { - // TODO - return 0; - } - - public void writerIndex(final int writerIndex) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void setIndex(final int readerIndex, final int writerIndex) - { - positioningNotSupported(); - } - - public void clear() - { - } - - public boolean readable() - { - return true; - } - - public boolean writable() - { - return false; - } - - public int readableBytes() - { - return 1; - } - - public int writableBytes() - { - return 0; - } - - public void markReaderIndex() - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void resetReaderIndex() - { - // TODO: reset positioning if possible - } - - public void markWriterIndex() - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void resetWriterIndex() - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void discardReadBytes() - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public short getUnsignedByte(final int index) - { - return (short)(getByte(index) & 0xFF); - } - - public int getUnsignedShort(final int index) - { - return getShort(index) & 0xFFFF; - } - - public int getMedium(final int index) - { - int value = getUnsignedMedium(index); - if ((value & 0x800000) != 0) - { - value |= 0xff000000; - } - return value; - } - - public long getUnsignedInt(final int index) - { - return getInt(index) & 0xFFFFFFFFL; - } - - public void getBytes(int index, final byte[] dst) - { - // TODO: optimize this by using System.arraycopy - for (int i = 0; i < dst.length; i++) - { - dst[i] = getByte(index++); - } - } - - public void getBytes(long index, final byte[] dst) - { - // TODO: optimize this by using System.arraycopy - for (int i = 0; i < dst.length; i++) - { - dst[i] = getByte(index++); - } - } - - public void getBytes(final int index, final HornetQBuffer dst) - { - getBytes(index, dst, dst.writableBytes()); - } - - public void getBytes(final int index, final HornetQBuffer dst, final int length) - { - if (length > dst.writableBytes()) - { - throw new IndexOutOfBoundsException(); - } - getBytes(index, dst, dst.writerIndex(), length); - dst.writerIndex(dst.writerIndex() + length); - } - - public void setBytes(final int index, final byte[] src) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void setBytes(final int index, final HornetQBuffer src) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void setBytes(final int index, final HornetQBuffer src, final int length) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void setZero(final int index, final int length) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public short readUnsignedByte() - { - try - { - return (short)getStream().readUnsignedByte(); - } - catch (Exception e) - { - throw new IllegalStateException (e.getMessage(), e); - } - } - - public short readShort() - { - try - { - return (short)getStream().readShort(); - } - catch (Exception e) - { - throw new IllegalStateException (e.getMessage(), e); - } - } - - public int readUnsignedShort() - { - try - { - return (int)getStream().readUnsignedShort(); - } - catch (Exception e) - { - throw new IllegalStateException (e.getMessage(), e); - } - } - - public int readMedium() - { - int value = readUnsignedMedium(); - if ((value & 0x800000) != 0) - { - value |= 0xff000000; - } - return value; - } - - - public int readUnsignedMedium() - { - return (readByte() & 0xff) << 16 | (readByte() & 0xff) << 8 | (readByte() & 0xff) << 0; - } - - public int readInt() - { - try - { - return getStream().readInt(); - } - catch (Exception e) - { - throw new IllegalStateException(e.getMessage(), e); - } - } - - public int readInt(final int pos) - { - positioningNotSupported(); - return 0; - } - - public long readUnsignedInt() - { - return readInt() & 0xFFFFFFFFL; - } - - public long readLong() - { - try - { - return getStream().readLong(); - } - catch (Exception e) - { - throw new IllegalStateException(e.getMessage(), e); - } - } - - public void readBytes(final byte[] dst, final int dstIndex, final int length) - { - try - { - getStream().read(dst, dstIndex, length); - } - catch (Exception e) - { - throw new IllegalStateException(e.getMessage(), e); - } - } - - public void readBytes(final byte[] dst) - { - readBytes(dst, 0, dst.length); - } - - public void readBytes(final HornetQBuffer dst) - { - readBytes(dst, dst.writableBytes()); - } - - public void readBytes(final HornetQBuffer dst, final int length) - { - if (length > dst.writableBytes()) - { - throw new IndexOutOfBoundsException(); - } - readBytes(dst, dst.writerIndex(), length); - dst.writerIndex(dst.writerIndex() + length); - } - - public void readBytes(final HornetQBuffer dst, final int dstIndex, final int length) - { - byte[] destBytes = new byte[length]; - readBytes(destBytes); - dst.setBytes(dstIndex, destBytes); - } - - public void readBytes(final ByteBuffer dst) - { - byte bytesToGet[] = new byte[dst.remaining()]; - readBytes(bytesToGet); - dst.put(bytesToGet); - } - - public int readBytes(final GatheringByteChannel out, final int length) throws IOException - { - throw new IllegalStateException("Not implemented!"); - } - - public void readBytes(final OutputStream out, final int length) throws IOException - { - throw new IllegalStateException("Not implemented!"); - } - - public void skipBytes(final int length) - { - - try - { - for (int i = 0 ; i < length; i++) - { - getStream().read(); - } - } - catch (Exception e) - { - throw new IllegalStateException(e.getMessage(), e); - } - } - - public void writeByte(final byte value) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void writeShort(final short value) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void writeMedium(final int value) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void writeInt(final int value) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void writeLong(final long value) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void writeBytes(final byte[] src, final int srcIndex, final int length) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void writeBytes(final byte[] src) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void writeBytes(final HornetQBuffer src) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void writeBytes(final HornetQBuffer src, final int length) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void writeBytes(final ByteBuffer src) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public int writeBytes(final InputStream in, final int length) throws IOException - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public int writeBytes(final ScatteringByteChannel in, final int length) throws IOException - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void writeZero(final int length) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public ByteBuffer toByteBuffer() - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public ByteBuffer[] toByteBuffers() - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public ByteBuffer[] toByteBuffers(final int index, final int length) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public String toString(final String charsetName) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public Object getUnderlyingBuffer() - { - return this; - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#readBoolean() - */ - public boolean readBoolean() - { - return readByte() != 0; - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#readChar() - */ - public char readChar() - { - return (char)readShort(); - } - - public char getChar(final int index) - { - return (char)getShort(index); - } - - public double getDouble(final int index) - { - return Double.longBitsToDouble(getLong(index)); - } - - public float getFloat(final int index) - { - return Float.intBitsToFloat(getInt(index)); - } - - public HornetQBuffer readBytes(final int length) - { - byte bytesToGet[] = new byte[length]; - readBytes(bytesToGet); - return HornetQBuffers.wrappedBuffer(bytesToGet); - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#readDouble() - */ - public double readDouble() - { - return Double.longBitsToDouble(readLong()); - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#readFloat() - */ - public float readFloat() - { - return Float.intBitsToFloat(readInt()); - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#readNullableSimpleString() - */ - public SimpleString readNullableSimpleString() - { - int b = readByte(); - if (b == DataConstants.NULL) - { - return null; - } - else - { - return readSimpleString(); - } - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#readNullableString() - */ - public String readNullableString() - { - int b = readByte(); - if (b == DataConstants.NULL) - { - return null; - } - else - { - return readString(); - } - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#readSimpleString() - */ - public SimpleString readSimpleString() - { - int len = readInt(); - byte[] data = new byte[len]; - readBytes(data); - return new SimpleString(data); - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#readString() - */ - public String readString() - { - int len = readInt(); - - if (len < 9) - { - char[] chars = new char[len]; - for (int i = 0; i < len; i++) - { - chars[i] = (char)readShort(); - } - return new String(chars); - } - else if (len < 0xfff) - { - return readUTF(); - } - else - { - return readSimpleString().toString(); - } - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#readUTF() - */ - public String readUTF() - { - return UTF8Util.readUTF(this); - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeBoolean(boolean) - */ - public void writeBoolean(final boolean val) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeChar(char) - */ - public void writeChar(final char val) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeDouble(double) - */ - public void writeDouble(final double val) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeFloat(float) - */ - public void writeFloat(final float val) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeNullableSimpleString(org.hornetq.util.SimpleString) - */ - public void writeNullableSimpleString(final SimpleString val) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeNullableString(java.lang.String) - */ - public void writeNullableString(final String val) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeSimpleString(org.hornetq.util.SimpleString) - */ - public void writeSimpleString(final SimpleString val) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeString(java.lang.String) - */ - public void writeString(final String val) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - /* (non-Javadoc) - * @see org.hornetq.spi.core.remoting.HornetQBuffer#writeUTF(java.lang.String) - */ - public void writeUTF(final String utf) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - /* (non-Javadoc) - * @see org.hornetq.api.core.buffers.ChannelBuffer#compareTo(org.hornetq.api.core.buffers.ChannelBuffer) - */ - public int compareTo(final HornetQBuffer buffer) - { - return -1; - } - - public HornetQBuffer copy() - { - throw new UnsupportedOperationException(); - } - - public HornetQBuffer slice(final int index, final int length) - { - throw new UnsupportedOperationException(); - } - - // Package protected --------------------------------------------- - - // Protected ----------------------------------------------------- - - // Private ------------------------------------------------------- - - /** - * @param body - */ - // Inner classes ------------------------------------------------- - - public ChannelBuffer channelBuffer() - { - return null; - } - - public HornetQBuffer copy(final int index, final int length) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public HornetQBuffer duplicate() - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public HornetQBuffer readSlice(final int length) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void setChar(final int index, final char value) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void setDouble(final int index, final double value) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void setFloat(final int index, final float value) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public HornetQBuffer slice() - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } - - public void writeBytes(final HornetQBuffer src, final int srcIndex, final int length) - { - throw new IllegalAccessError(OPERATION_NOT_SUPPORTED); - } -} Modified: trunk/tests/src/org/hornetq/tests/unit/core/client/impl/LargeMessageBufferTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/unit/core/client/impl/LargeMessageBufferTest.java 2010-12-07 18:05:17 UTC (rev 10007) +++ trunk/tests/src/org/hornetq/tests/unit/core/client/impl/LargeMessageBufferTest.java 2010-12-07 22:22:35 UTC (rev 10008) @@ -37,7 +37,7 @@ import org.hornetq.core.client.impl.ClientConsumerInternal; import org.hornetq.core.client.impl.ClientMessageInternal; import org.hornetq.core.client.impl.ClientSessionInternal; -import org.hornetq.core.client.impl.LargeMessageBufferImpl; +import org.hornetq.core.client.impl.LargeMessageControllerImpl; import org.hornetq.core.protocol.core.impl.wireformat.SessionQueueQueryResponseMessage; import org.hornetq.core.protocol.core.impl.wireformat.SessionReceiveContinuationMessage; import org.hornetq.core.protocol.core.impl.wireformat.SessionReceiveLargeMessage; @@ -85,7 +85,7 @@ // Test Simple getBytes public void testGetBytes() throws Exception { - LargeMessageBufferImpl buffer = create15BytesSample(); + LargeMessageControllerImpl buffer = create15BytesSample(); for (int i = 1; i <= 15; i++) { @@ -112,7 +112,7 @@ // Test for void getBytes(final int index, final byte[] dst) public void testGetBytesIByteArray() throws Exception { - LargeMessageBufferImpl buffer = create15BytesSample(); + LargeMessageControllerImpl buffer = create15BytesSample(); byte[] bytes = new byte[15]; buffer.getBytes(0, bytes); @@ -135,7 +135,7 @@ // testing void getBytes(int index, ChannelBuffer dst, int dstIndex, int length) public void testGetBytesILChannelBufferII() throws Exception { - LargeMessageBufferImpl buffer = create15BytesSample(); + LargeMessageControllerImpl buffer = create15BytesSample(); HornetQBuffer dstBuffer = HornetQBuffers.fixedBuffer(20); @@ -152,7 +152,7 @@ // testing void getBytes(int index, ChannelBuffer dst, int dstIndex, int length) public void testReadIntegers() throws Exception { - LargeMessageBufferImpl buffer = createBufferWithIntegers(3, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15); + LargeMessageControllerImpl buffer = createBufferWithIntegers(3, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15); for (int i = 1; i <= 15; i++) { @@ -171,7 +171,7 @@ public void testReadIntegersOverStream() throws Exception { - LargeMessageBufferImpl buffer = createBufferWithIntegers(3, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15); + LargeMessageControllerImpl buffer = createBufferWithIntegers(3, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15); HornetQBufferInputStream is = new HornetQBufferInputStream(buffer); DataInputStream dataInput = new DataInputStream(is); @@ -186,7 +186,7 @@ // testing void getBytes(int index, ChannelBuffer dst, int dstIndex, int length) public void testReadLongs() throws Exception { - LargeMessageBufferImpl buffer = createBufferWithLongs(3, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15); + LargeMessageControllerImpl buffer = createBufferWithLongs(3, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15); for (int i = 1; i <= 15; i++) { @@ -205,7 +205,7 @@ public void testReadLongsOverStream() throws Exception { - LargeMessageBufferImpl buffer = createBufferWithLongs(3, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15); + LargeMessageControllerImpl buffer = createBufferWithLongs(3, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15); HornetQBufferInputStream is = new HornetQBufferInputStream(buffer); DataInputStream dataInput = new DataInputStream(is); @@ -231,7 +231,7 @@ dynamic.writeDouble(d1); dynamic.writeFloat(f1); - LargeMessageBufferImpl readBuffer = splitBuffer(3, dynamic.toByteBuffer().array()); + LargeMessageControllerImpl readBuffer = splitBuffer(3, dynamic.toByteBuffer().array()); Assert.assertEquals(str1, readBuffer.readUTF()); Assert.assertEquals(str2, readBuffer.readString()); @@ -260,7 +260,7 @@ dynamic.writeDouble(d1); dynamic.writeFloat(f1); - LargeMessageBufferImpl readBuffer = splitBuffer(3, dynamic.toByteBuffer().array(), getTestFile()); + LargeMessageControllerImpl readBuffer = splitBuffer(3, dynamic.toByteBuffer().array(), getTestFile()); Assert.assertEquals(str1, readBuffer.readUTF()); Assert.assertEquals(str2, readBuffer.readString()); @@ -280,7 +280,7 @@ public void testReadPartialData() throws Exception { - final LargeMessageBufferImpl buffer = new LargeMessageBufferImpl(new FakeConsumerInternal(), 10, 10); + final LargeMessageControllerImpl buffer = new LargeMessageControllerImpl(new FakeConsumerInternal(), 10, 10); buffer.addPacket(new FakePacket(-1, new byte[] { 0, 1, 2, 3, 4 }, true, true)); @@ -336,7 +336,7 @@ public void testInterruptData() throws Exception { - LargeMessageBufferImpl readBuffer = splitBuffer(3, new byte[] { 0, 1, 2, 3, 4 }); + LargeMessageControllerImpl readBuffer = splitBuffer(3, new byte[] { 0, 1, 2, 3, 4 }); byte bytes[] = new byte[30]; readBuffer.readBytes(bytes, 0, 5); @@ -349,7 +349,7 @@ public void testSplitBufferOnFile() throws Exception { - LargeMessageBufferImpl outBuffer = new LargeMessageBufferImpl(new FakeConsumerInternal(), + LargeMessageControllerImpl outBuffer = new LargeMessageControllerImpl(new FakeConsumerInternal(), 1024 * 1024, 1, getTestFile(), @@ -391,7 +391,7 @@ public void testStreamData() throws Exception { - final LargeMessageBufferImpl outBuffer = new LargeMessageBufferImpl(new FakeConsumerInternal(), + final LargeMessageControllerImpl outBuffer = new LargeMessageControllerImpl(new FakeConsumerInternal(), 1024 * 11 + 123, 1); @@ -509,7 +509,7 @@ public void testStreamDataWaitCompletionOnCompleteBuffer() throws Exception { - final LargeMessageBufferImpl outBuffer = create15BytesSample(); + final LargeMessageControllerImpl outBuffer = create15BytesSample(); outBuffer.saveBuffer(new OutputStream() { @@ -524,7 +524,7 @@ public void testErrorOnSetStreaming() throws Exception { long start = System.currentTimeMillis(); - final LargeMessageBufferImpl outBuffer = new LargeMessageBufferImpl(new FakeConsumerInternal(), 5, 30); + final LargeMessageControllerImpl outBuffer = new LargeMessageControllerImpl(new FakeConsumerInternal(), 5, 30); outBuffer.addPacket(new FakePacket(-1, new byte[] { 0, 1, 2, 3, 4 }, true, false)); @@ -589,12 +589,12 @@ /** * @return */ - private LargeMessageBufferImpl create15BytesSample() throws Exception + private LargeMessageControllerImpl create15BytesSample() throws Exception { return splitBuffer(5, new byte[] { 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15 }); } - private LargeMessageBufferImpl createBufferWithIntegers(final int splitFactor, final int... values) throws Exception + private LargeMessageControllerImpl createBufferWithIntegers(final int splitFactor, final int... values) throws Exception { ByteArrayOutputStream byteOut = new ByteArrayOutputStream(values.length * 4); DataOutputStream dataOut = new DataOutputStream(byteOut); @@ -607,7 +607,7 @@ return splitBuffer(splitFactor, byteOut.toByteArray()); } - private LargeMessageBufferImpl createBufferWithLongs(final int splitFactor, final long... values) throws Exception + private LargeMessageControllerImpl createBufferWithLongs(final int splitFactor, final long... values) throws Exception { ByteArrayOutputStream byteOut = new ByteArrayOutputStream(values.length * 8); DataOutputStream dataOut = new DataOutputStream(byteOut); @@ -620,14 +620,14 @@ return splitBuffer(splitFactor, byteOut.toByteArray()); } - private LargeMessageBufferImpl splitBuffer(final int splitFactor, final byte[] bytes) throws Exception + private LargeMessageControllerImpl splitBuffer(final int splitFactor, final byte[] bytes) throws Exception { return splitBuffer(splitFactor, bytes, null); } - private LargeMessageBufferImpl splitBuffer(final int splitFactor, final byte[] bytes, final File file) throws Exception + private LargeMessageControllerImpl splitBuffer(final int splitFactor, final byte[] bytes, final File file) throws Exception { - LargeMessageBufferImpl outBuffer = new LargeMessageBufferImpl(new FakeConsumerInternal(), bytes.length, 5, file); + LargeMessageControllerImpl outBuffer = new LargeMessageControllerImpl(new FakeConsumerInternal(), bytes.length, 5, file); ByteArrayInputStream input = new ByteArrayInputStream(bytes); Modified: trunk/tests/src/org/hornetq/tests/unit/core/paging/impl/PagingStoreImplTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/unit/core/paging/impl/PagingStoreImplTest.java 2010-12-07 18:05:17 UTC (rev 10007) +++ trunk/tests/src/org/hornetq/tests/unit/core/paging/impl/PagingStoreImplTest.java 2010-12-07 22:22:35 UTC (rev 10008) @@ -40,6 +40,7 @@ import org.hornetq.core.journal.SequentialFile; import org.hornetq.core.journal.SequentialFileFactory; import org.hornetq.core.journal.impl.NIOSequentialFileFactory; +import org.hornetq.core.message.impl.MessageInternal; import org.hornetq.core.paging.Page; import org.hornetq.core.paging.PageTransactionInfo; import org.hornetq.core.paging.PagedMessage; @@ -75,7 +76,6 @@ import org.hornetq.tests.util.RandomUtil; import org.hornetq.tests.util.UnitTestCase; import org.hornetq.utils.ExecutorFactory; -import org.hornetq.utils.UUID; /** * @@ -1387,7 +1387,7 @@ /* (non-Javadoc) * @see org.hornetq.core.persistence.StorageManager#createLargeMessage(byte[]) */ - public LargeServerMessage createLargeMessage(final long messageId, final byte[] header) + public LargeServerMessage createLargeMessage(final long messageId, final MessageInternal msg) { return null; Modified: trunk/tests/src/org/hornetq/tests/unit/core/postoffice/impl/BindingsImplTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/unit/core/postoffice/impl/BindingsImplTest.java 2010-12-07 18:05:17 UTC (rev 10007) +++ trunk/tests/src/org/hornetq/tests/unit/core/postoffice/impl/BindingsImplTest.java 2010-12-07 22:22:35 UTC (rev 10008) @@ -983,6 +983,15 @@ return false; } + /* (non-Javadoc) + * @see org.hornetq.core.message.impl.MessageInternal#getTypedProperties() + */ + public TypedProperties getTypedProperties() + { + // TODO Auto-generated method stub + return null; + } + } class FakeFilter implements Filter