You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by lq...@apache.org on 2017/10/18 15:21:41 UTC
[07/10] qpid-broker-j git commit: QPID-7832: [Java Broker] Refactor
store/protocol API using Collection
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/660c206d/broker-core/src/test/java/org/apache/qpid/server/bytebuffer/QpidByteBufferTest.java
----------------------------------------------------------------------
diff --git a/broker-core/src/test/java/org/apache/qpid/server/bytebuffer/QpidByteBufferTest.java b/broker-core/src/test/java/org/apache/qpid/server/bytebuffer/QpidByteBufferTest.java
index d98c4f1..3653c05 100644
--- a/broker-core/src/test/java/org/apache/qpid/server/bytebuffer/QpidByteBufferTest.java
+++ b/broker-core/src/test/java/org/apache/qpid/server/bytebuffer/QpidByteBufferTest.java
@@ -29,23 +29,19 @@ import java.lang.reflect.Method;
import java.nio.BufferOverflowException;
import java.nio.BufferUnderflowException;
import java.nio.ByteBuffer;
-import java.nio.CharBuffer;
+import java.nio.InvalidMarkException;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.Iterator;
-import java.util.List;
import com.google.common.io.ByteStreams;
import org.junit.Assert;
import org.mockito.internal.util.Primitives;
import org.apache.qpid.test.utils.QpidTestCase;
-import org.apache.qpid.server.util.ByteBufferUtils;
public class QpidByteBufferTest extends QpidTestCase
{
+ private static final int BUFFER_FRAGMENT_SIZE = 5;
private static final int BUFFER_SIZE = 10;
private static final int POOL_SIZE = 20;
private static final double SPARSITY_FRACTION = 0.5;
@@ -59,20 +55,29 @@ public class QpidByteBufferTest extends QpidTestCase
{
super.setUp();
QpidByteBuffer.deinitialisePool();
- QpidByteBuffer.initialisePool(BUFFER_SIZE, POOL_SIZE, SPARSITY_FRACTION);
+ QpidByteBuffer.initialisePool(BUFFER_FRAGMENT_SIZE, POOL_SIZE, SPARSITY_FRACTION);
_parent = QpidByteBuffer.allocateDirect(BUFFER_SIZE);
}
@Override
public void tearDown() throws Exception
{
- super.tearDown();
- _parent.dispose();
- if (_slicedBuffer != null)
+ try
{
- _slicedBuffer.dispose();
+ super.tearDown();
+ }
+ finally
+ {
+ if (_parent != null)
+ {
+ _parent.dispose();
+ }
+ if (_slicedBuffer != null)
+ {
+ _slicedBuffer.dispose();
+ }
+ QpidByteBuffer.deinitialisePool();
}
- QpidByteBuffer.deinitialisePool();
}
public void testPutGetByIndex() throws Exception
@@ -113,6 +118,25 @@ public class QpidByteBufferTest extends QpidTestCase
assertEquals("Unexpected position after reset", 0, _slicedBuffer.position());
}
+ public void testMarkResetAcrossFragmentBoundary() throws Exception
+ {
+ for (int i = 0; i < BUFFER_SIZE; ++i)
+ {
+ _parent.put((byte) i);
+ }
+ _parent.flip();
+ _parent.mark();
+ for (int i = 0; i < BUFFER_FRAGMENT_SIZE + 2; ++i)
+ {
+ assertEquals("Unexpected value", i, _parent.get());
+ }
+ _parent.reset();
+ for (int i = 0; i < BUFFER_SIZE; ++i)
+ {
+ assertEquals("Unexpected value", i, _parent.get());
+ }
+ }
+
public void testPosition() throws Exception
{
_slicedBuffer = createSlice();
@@ -135,155 +159,102 @@ public class QpidByteBufferTest extends QpidTestCase
}
}
- public void testBulkPutGet() throws Exception
+ public void testSettingPositionBackwardsResetsMark()
{
- _slicedBuffer = createSlice();
-
- final byte[] source = getTestBytes(_slicedBuffer.remaining());
-
- QpidByteBuffer rv = _slicedBuffer.put(source, 0, source.length);
- assertEquals("Unexpected builder return value", _slicedBuffer, rv);
-
- _slicedBuffer.flip();
- byte[] target = new byte[_slicedBuffer.remaining()];
- rv = _slicedBuffer.get(target, 0, target.length);
- assertEquals("Unexpected builder return value", _slicedBuffer, rv);
-
- Assert.assertArrayEquals("Unexpected bulk put/get result", source, target);
-
-
- _slicedBuffer.clear();
- _slicedBuffer.position(1);
-
+ _parent.position(8);
+ _parent.mark();
+ _parent.position(7);
try
{
- _slicedBuffer.put(source, 0, source.length);
- fail("Exception not thrown");
+ _parent.reset();
+ fail("Expected exception not thrown");
}
- catch (BufferOverflowException e)
+ catch (InvalidMarkException e)
{
// pass
}
+ }
- assertEquals("Position should be unchanged after failed put", 1, _slicedBuffer.position());
+ public void testSettingPositionForwardDoeNotResetMark()
+ {
+ final int originalPosition = 3;
+ _parent.position(originalPosition);
+ _parent.mark();
+ _parent.position(9);
+
+ _parent.reset();
+
+ assertEquals("Unexpected position", originalPosition, _parent.position());
+ }
+
+ public void testRewind() throws Exception
+ {
+ final int expectedLimit = 7;
+ _parent.position(1);
+ _parent.limit(expectedLimit);
+ _parent.mark();
+ _parent.position(3);
+
+ _parent.rewind();
+ assertEquals("Unexpected position", 0, _parent.position());
+ assertEquals("Unexpected limit", expectedLimit, _parent.limit());
try
{
- _slicedBuffer.get(target, 0, target.length);
- fail("Exception not thrown");
+ _parent.reset();
+ fail("Expected exception not thrown");
}
- catch (BufferUnderflowException e)
+ catch (InvalidMarkException e)
{
// pass
}
-
- assertEquals("Position should be unchanged after failed get", 1, _slicedBuffer.position());
-
-
}
- public void testByteBufferPutGet()
+ public void testBulkPutGet() throws Exception
{
_slicedBuffer = createSlice();
- final byte[] source = getTestBytes(_slicedBuffer.remaining());
- ByteBuffer sourceByteBuffer = ByteBuffer.wrap(source);
+ final byte[] source = getTestBytes(_slicedBuffer.remaining());
- QpidByteBuffer rv = _slicedBuffer.put(sourceByteBuffer);
+ QpidByteBuffer rv = _slicedBuffer.put(source, 0, source.length);
assertEquals("Unexpected builder return value", _slicedBuffer, rv);
- assertEquals("Unexpected position", _slicedBuffer.capacity(), _slicedBuffer.position());
- assertEquals("Unexpected remaining", 0, _slicedBuffer.remaining());
-
- assertEquals("Unexpected remaining in source ByteBuffer", 0, sourceByteBuffer.remaining());
-
_slicedBuffer.flip();
+ byte[] target = new byte[_slicedBuffer.remaining()];
+ rv = _slicedBuffer.get(target, 0, target.length);
+ assertEquals("Unexpected builder return value", _slicedBuffer, rv);
- ByteBuffer destinationByteBuffer = ByteBuffer.allocate(source.length);
- _slicedBuffer.get(destinationByteBuffer);
-
-
- assertEquals("Unexpected remaining", 0, _slicedBuffer.remaining());
-
- assertEquals("Unexpected remaining in destination ByteBuffer", 0, destinationByteBuffer.remaining());
- assertEquals("Unexpected position in destination ByteBuffer", source.length, destinationByteBuffer.position());
+ Assert.assertArrayEquals("Unexpected bulk put/get result", source, target);
- Assert.assertArrayEquals("Unexpected ByteBuffer put/get result", source, destinationByteBuffer.array());
_slicedBuffer.clear();
_slicedBuffer.position(1);
- sourceByteBuffer.clear();
try
{
- _slicedBuffer.put(sourceByteBuffer);
- fail("Exception should be thrown");
+ _slicedBuffer.put(source, 0, source.length);
+ fail("Exception not thrown");
}
- catch(BufferOverflowException e)
+ catch (BufferOverflowException e)
{
// pass
}
- assertEquals("Position should not be changed after failed put", 1, _slicedBuffer.position());
- assertEquals("Source position should not changed after failed put", source.length, sourceByteBuffer.remaining());
-
- _slicedBuffer.clear();
- destinationByteBuffer.position(1);
+ assertEquals("Position should be unchanged after failed put", 1, _slicedBuffer.position());
try
{
- _slicedBuffer.get(destinationByteBuffer);
- fail("Exception should be thrown");
+ _slicedBuffer.get(target, 0, target.length);
+ fail("Exception not thrown");
}
- catch(BufferUnderflowException e )
+ catch (BufferUnderflowException e)
{
// pass
}
- }
-
- public void testQpidByteBufferPutGet()
- {
- _slicedBuffer = createSlice();
- final byte[] source = getTestBytes(_slicedBuffer.remaining());
-
- QpidByteBuffer sourceQpidByteBuffer = QpidByteBuffer.wrap(source);
- QpidByteBuffer rv = _slicedBuffer.put(sourceQpidByteBuffer);
- assertEquals("Unexpected builder return value", _slicedBuffer, rv);
-
- assertEquals("Unexpected position", _slicedBuffer.capacity(), _slicedBuffer.position());
- assertEquals("Unexpected remaining", 0, _slicedBuffer.remaining());
-
- assertEquals("Unexpected remaining in source QpidByteBuffer", 0, sourceQpidByteBuffer.remaining());
-
- _slicedBuffer.flip();
-
- ByteBuffer destinationByteBuffer = ByteBuffer.allocate(source.length);
- _slicedBuffer.get(destinationByteBuffer);
-
- assertEquals("Unexpected remaining", 0, _slicedBuffer.remaining());
-
- assertEquals("Unexpected remaining in destination ByteBuffer", 0, destinationByteBuffer.remaining());
- assertEquals("Unexpected position in destination ByteBuffer", source.length, destinationByteBuffer.position());
-
- Assert.assertArrayEquals("Unexpected ByteBuffer put/get result", source, destinationByteBuffer.array());
-
- _slicedBuffer.clear();
- _slicedBuffer.position(1);
+ assertEquals("Position should be unchanged after failed get", 1, _slicedBuffer.position());
- sourceQpidByteBuffer.clear();
- try
- {
- _slicedBuffer.put(sourceQpidByteBuffer);
- fail("Exception should be thrown");
- }
- catch(BufferOverflowException e)
- {
- // pass
- }
- assertEquals("Position should not be changed after failed put", 1, _slicedBuffer.position());
- assertEquals("Source position should not changed after failed put", source.length, sourceQpidByteBuffer.remaining());
}
public void testDuplicate()
@@ -293,8 +264,7 @@ public class QpidByteBufferTest extends QpidTestCase
int originalLimit = _slicedBuffer.limit();
_slicedBuffer.limit(originalLimit - 1);
- QpidByteBuffer duplicate = _slicedBuffer.duplicate();
- try
+ try (QpidByteBuffer duplicate = _slicedBuffer.duplicate())
{
assertEquals("Unexpected position", _slicedBuffer.position(), duplicate.position() );
assertEquals("Unexpected limit", _slicedBuffer.limit(), duplicate.limit() );
@@ -306,10 +276,6 @@ public class QpidByteBufferTest extends QpidTestCase
assertEquals("Unexpected position in the original", 1, _slicedBuffer.position());
assertEquals("Unexpected limit in the original", originalLimit -1, _slicedBuffer.limit());
}
- finally
- {
- duplicate.dispose();
- }
}
public void testCopyToByteBuffer()
@@ -400,8 +366,8 @@ public class QpidByteBufferTest extends QpidTestCase
_slicedBuffer.limit(_slicedBuffer.limit() - 1);
int remaining = _slicedBuffer.remaining();
- QpidByteBuffer newSlice = _slicedBuffer.slice();
- try
+
+ try (QpidByteBuffer newSlice = _slicedBuffer.slice())
{
assertEquals("Unexpected position in original", 1, _slicedBuffer.position());
assertEquals("Unexpected limit in original", source.length - 1, _slicedBuffer.limit());
@@ -416,10 +382,6 @@ public class QpidByteBufferTest extends QpidTestCase
System.arraycopy(source, 1, expected, 0, expected.length);
Assert.assertArrayEquals("Unexpected slice result", expected, destination);
}
- finally
- {
- newSlice.dispose();
- }
}
public void testViewOfSlice()
@@ -431,8 +393,7 @@ public class QpidByteBufferTest extends QpidTestCase
_slicedBuffer.position(1);
_slicedBuffer.limit(_slicedBuffer.limit() - 1);
- QpidByteBuffer view = _slicedBuffer.view(0, _slicedBuffer.remaining());
- try
+ try (QpidByteBuffer view = _slicedBuffer.view(0, _slicedBuffer.remaining()))
{
assertEquals("Unexpected position in original", 1, _slicedBuffer.position());
assertEquals("Unexpected limit in original", source.length - 1, _slicedBuffer.limit());
@@ -448,13 +409,8 @@ public class QpidByteBufferTest extends QpidTestCase
System.arraycopy(source, 1, expected, 0, expected.length);
Assert.assertArrayEquals("Unexpected view result", expected, destination);
}
- finally
- {
- view.dispose();
- }
- view = _slicedBuffer.view(1, _slicedBuffer.remaining() - 2);
- try
+ try (QpidByteBuffer view = _slicedBuffer.view(1, _slicedBuffer.remaining() - 2))
{
assertEquals("Unexpected position in original", 1, _slicedBuffer.position());
assertEquals("Unexpected limit in original", source.length - 1, _slicedBuffer.limit());
@@ -470,10 +426,6 @@ public class QpidByteBufferTest extends QpidTestCase
System.arraycopy(source, 2, expected, 0, expected.length);
Assert.assertArrayEquals("Unexpected view result", expected, destination);
}
- finally
- {
- view.dispose();
- }
}
public void testAsInputStream() throws Exception
@@ -496,40 +448,6 @@ public class QpidByteBufferTest extends QpidTestCase
Assert.assertArrayEquals("Unexpected view result", expected, destination.toByteArray());
}
- public void testAsByteBuffer() throws Exception
- {
- _slicedBuffer = createSlice();
-
- _slicedBuffer.position(1);
- _slicedBuffer.limit(_slicedBuffer.limit() - 1);
-
- _slicedBuffer.mark();
- int remaining = _slicedBuffer.remaining();
- byte[] source = getTestBytes(remaining);
- _slicedBuffer.put(source);
- _slicedBuffer.reset();
-
- ByteBuffer buffer = _slicedBuffer.asByteBuffer();
- assertEquals("Unexpected remaining", remaining, buffer.remaining());
-
- byte[] target = new byte[remaining];
- buffer.get(target);
- Assert.assertArrayEquals("Unexpected asByteBuffer result", source, target);
- }
-
- public void testDecode()
- {
- _slicedBuffer = createSlice();
- final String input = "ABC";
- _slicedBuffer.put(input.getBytes());
- _slicedBuffer.flip();
-
- final CharBuffer charBuffer = _slicedBuffer.decode(StandardCharsets.US_ASCII);
- final char[] destination = new char[charBuffer.remaining()];
- charBuffer.get(destination);
- Assert.assertArrayEquals("Unexpected char buffer", input.toCharArray(), destination);
- }
-
private byte[] getTestBytes(final int length)
{
final byte[] source = new byte[length];
@@ -598,6 +516,8 @@ public class QpidByteBufferTest extends QpidTestCase
{
// pass
}
+ _slicedBuffer.dispose();
+ _slicedBuffer = null;
}
private void testPutGetByIndex(final Class<?> primitiveTargetClass, Object value) throws Exception
@@ -659,6 +579,8 @@ public class QpidByteBufferTest extends QpidTestCase
{
// pass
}
+ _slicedBuffer.dispose();
+ _slicedBuffer = null;
}
private void invokeMethod(final Method method, final Object... value)
@@ -734,41 +656,49 @@ public class QpidByteBufferTest extends QpidTestCase
public void testPooledBufferIsZeroedLoan() throws Exception
{
- QpidByteBuffer buffer = QpidByteBuffer.allocateDirect(BUFFER_SIZE);
-
- buffer.put((byte) 0xFF);
- buffer.dispose();
+ try (QpidByteBuffer buffer = QpidByteBuffer.allocateDirect(BUFFER_SIZE))
+ {
+ buffer.put((byte) 0xFF);
+ }
- buffer = QpidByteBuffer.allocateDirect(BUFFER_SIZE);
- buffer.limit(1);
- assertEquals("Pooled QpidByteBuffer is not zeroed.", (byte) 0x0, buffer.get());
+ try (QpidByteBuffer buffer = QpidByteBuffer.allocateDirect(BUFFER_SIZE))
+ {
+ buffer.limit(1);
+ assertEquals("Pooled QpidByteBuffer is not zeroed.", (byte) 0x0, buffer.get());
+ }
}
public void testAllocateDirectOfSameSize() throws Exception
{
int bufferSize = BUFFER_SIZE;
- QpidByteBuffer buffer = QpidByteBuffer.allocateDirect(bufferSize);
- assertEquals("Unexpected buffer size", bufferSize, buffer.capacity());
- assertEquals("Unexpected position on newly created buffer", 0, buffer.position());
- assertEquals("Unexpected limit on newly created buffer", bufferSize, buffer.limit());
+ try (QpidByteBuffer buffer = QpidByteBuffer.allocateDirect(bufferSize))
+ {
+ assertEquals("Unexpected buffer size", bufferSize, buffer.capacity());
+ assertEquals("Unexpected position on newly created buffer", 0, buffer.position());
+ assertEquals("Unexpected limit on newly created buffer", bufferSize, buffer.limit());
+ }
}
public void testAllocateDirectOfSmallerSize() throws Exception
{
int bufferSize = BUFFER_SIZE - 1;
- QpidByteBuffer buffer = QpidByteBuffer.allocateDirect(bufferSize);
- assertEquals("Unexpected buffer size", bufferSize, buffer.capacity());
- assertEquals("Unexpected position on newly created buffer", 0, buffer.position());
- assertEquals("Unexpected limit on newly created buffer", bufferSize, buffer.limit());
+ try (QpidByteBuffer buffer = QpidByteBuffer.allocateDirect(bufferSize))
+ {
+ assertEquals("Unexpected buffer size", bufferSize, buffer.capacity());
+ assertEquals("Unexpected position on newly created buffer", 0, buffer.position());
+ assertEquals("Unexpected limit on newly created buffer", bufferSize, buffer.limit());
+ }
}
public void testAllocateDirectOfLargerSize() throws Exception
{
int bufferSize = BUFFER_SIZE + 1;
- QpidByteBuffer buffer = QpidByteBuffer.allocateDirect(bufferSize);
- assertEquals("Unexpected buffer size", bufferSize, buffer.capacity());
- assertEquals("Unexpected position on newly created buffer", 0, buffer.position());
- assertEquals("Unexpected limit on newly created buffer", bufferSize, buffer.limit());
+ try (QpidByteBuffer buffer = QpidByteBuffer.allocateDirect(bufferSize))
+ {
+ assertEquals("Unexpected buffer size", bufferSize, buffer.capacity());
+ assertEquals("Unexpected position on newly created buffer", 0, buffer.position());
+ assertEquals("Unexpected limit on newly created buffer", bufferSize, buffer.limit());
+ }
}
public void testAllocateDirectWithNegativeSize() throws Exception
@@ -800,37 +730,34 @@ public class QpidByteBufferTest extends QpidTestCase
public void testDeflateInflateDirect() throws Exception
{
byte[] input = "aaabbbcccddddeeeffff".getBytes();
- Collection<QpidByteBuffer> inputBufs = QpidByteBuffer.allocateDirectCollection(input.length);
-
- int offset = 0;
- for (QpidByteBuffer buf : inputBufs)
+ try (QpidByteBuffer inputBuf = QpidByteBuffer.allocateDirect(input.length))
{
- int len = buf.remaining();
- buf.put(input, offset, len);
- buf.flip();
- offset += len;
- }
- assertEquals(input.length, ByteBufferUtils.remaining(inputBufs));
+ inputBuf.put(input);
+ inputBuf.flip();
+
+ assertEquals(input.length, inputBuf.remaining());
- doDeflateInflate(input, inputBufs, true);
+ doDeflateInflate(input, inputBuf, true);
+ }
}
public void testDeflateInflateHeap() throws Exception
{
byte[] input = "aaabbbcccddddeeeffff".getBytes();
- Collection<QpidByteBuffer> inputBufs = Collections.singleton(QpidByteBuffer.wrap(input));
- doDeflateInflate(input, inputBufs, false);
+ try (QpidByteBuffer buffer = QpidByteBuffer.wrap(input))
+ {
+ doDeflateInflate(input, buffer, false);
+ }
}
public void testInflatingUncompressedBytes_ThrowsZipException() throws Exception
{
byte[] input = "not_a_compressed_stream".getBytes();
- QpidByteBuffer original = QpidByteBuffer.wrap(input);
- try
+ try (QpidByteBuffer original = QpidByteBuffer.wrap(input))
{
- QpidByteBuffer.inflate(Collections.singleton(original));
+ QpidByteBuffer.inflate(original);
fail("Exception not thrown");
}
catch(java.util.zip.ZipException ze)
@@ -841,109 +768,127 @@ public class QpidByteBufferTest extends QpidTestCase
public void testSlice() throws Exception
{
- QpidByteBuffer directBuffer = QpidByteBuffer.allocate(true, 6);
- directBuffer.position(2);
- directBuffer.limit(5);
- QpidByteBuffer directSlice = directBuffer.slice();
-
- assertTrue("Direct slice should be direct too", directSlice.isDirect());
- assertEquals("Unexpected capacity", 3, directSlice.capacity());
- assertEquals("Unexpected limit", 3, directSlice.limit());
- assertEquals("Unexpected position", 0, directSlice.position());
-
- directBuffer.dispose();
- directSlice.dispose();
+ try (QpidByteBuffer directBuffer = QpidByteBuffer.allocate(true, 6))
+ {
+ directBuffer.position(2);
+ directBuffer.limit(5);
+ try (QpidByteBuffer directSlice = directBuffer.slice())
+ {
+ assertTrue("Direct slice should be direct too", directSlice.isDirect());
+ assertEquals("Unexpected capacity", 3, directSlice.capacity());
+ assertEquals("Unexpected limit", 3, directSlice.limit());
+ assertEquals("Unexpected position", 0, directSlice.position());
+ }
+ }
}
public void testView() throws Exception
{
byte[] content = "ABCDEF".getBytes();
- QpidByteBuffer buffer = QpidByteBuffer.allocate(true, content.length);
- buffer.put(content);
- buffer.position(2);
- buffer.limit(5);
-
- QpidByteBuffer view = buffer.view(0, buffer.remaining());
-
- assertTrue("Unexpected view direct", view.isDirect());
-
- assertEquals("Unexpected capacity", 3, view.capacity());
- assertEquals("Unexpected limit", 3, view.limit());
- assertEquals("Unexpected position", 0, view.position());
-
- byte[] destination = new byte[view.remaining()];
- view.get(destination);
-
- Assert.assertArrayEquals("CDE".getBytes(), destination);
-
- QpidByteBuffer viewWithOffset = buffer.view(1, 1);
- destination = new byte[viewWithOffset.remaining()];
- viewWithOffset.get(destination);
+ try (QpidByteBuffer buffer = QpidByteBuffer.allocate(true, content.length))
+ {
+ buffer.put(content);
+ buffer.position(2);
+ buffer.limit(5);
- Assert.assertArrayEquals("D".getBytes(), destination);
+ try (QpidByteBuffer view = buffer.view(0, buffer.remaining()))
+ {
+ assertTrue("Unexpected view direct", view.isDirect());
+ assertEquals("Unexpected capacity", 3, view.capacity());
+ assertEquals("Unexpected limit", 3, view.limit());
+ assertEquals("Unexpected position", 0, view.position());
+
+ final byte[] destination = new byte[view.remaining()];
+ view.get(destination);
+ Assert.assertArrayEquals("CDE".getBytes(), destination);
+ }
- buffer.dispose();
- view.dispose();
- viewWithOffset.dispose();
+ try (QpidByteBuffer viewWithOffset = buffer.view(1, 1))
+ {
+ final byte[] destination = new byte[viewWithOffset.remaining()];
+ viewWithOffset.get(destination);
+ Assert.assertArrayEquals("D".getBytes(), destination);
+ }
+ }
}
public void testSparsity()
{
assertFalse("Unexpected sparsity after creation", _parent.isSparse());
- QpidByteBuffer child = _parent.view(0, 6);
+ QpidByteBuffer child = _parent.view(0, 8);
QpidByteBuffer grandChild = child.view(0, 2);
assertFalse("Unexpected sparsity after child creation", _parent.isSparse());
_parent.dispose();
+ _parent = null;
assertFalse("Unexpected sparsity after parent disposal", child.isSparse());
child.dispose();
assertTrue("Buffer should be sparse", grandChild.isSparse());
+ grandChild.dispose();
}
public void testAsQpidByteBuffers() throws IOException
{
byte[] dataForTwoBufs = "01234567890".getBytes(StandardCharsets.US_ASCII);
- Collection<QpidByteBuffer> qpidByteBuffers = QpidByteBuffer.asQpidByteBuffers(new ByteArrayInputStream(dataForTwoBufs));
- assertEquals("Unexpected number of bufs", 2, qpidByteBuffers.size());
- Iterator<QpidByteBuffer> itr = qpidByteBuffers.iterator();
- assertEquals("Unexpected remaining in first buf", 10, itr.next().remaining());
- assertEquals("Unexpected remaining in second buf", 1, itr.next().remaining());
+ try (QpidByteBuffer qpidByteBuffer = QpidByteBuffer.asQpidByteBuffer(new ByteArrayInputStream(dataForTwoBufs)))
+ {
+ assertEquals("Unexpected remaining in buf", 11, qpidByteBuffer.remaining());
+ }
+
+ try (QpidByteBuffer bufsForEmptyBytes = QpidByteBuffer.asQpidByteBuffer(new ByteArrayInputStream(new byte[]{})))
+ {
+ assertEquals("Unexpected remaining in buf for empty buffer", 0, bufsForEmptyBytes.remaining());
+ }
+ }
+
+ public void testConcatenate() throws Exception
+ {
+ try (QpidByteBuffer buffer1 = QpidByteBuffer.allocateDirect(10);
+ QpidByteBuffer buffer2 = QpidByteBuffer.allocateDirect(10))
+ {
+ final short buffer1Value = (short) (1 << 15);
+ buffer1.putShort(buffer1Value);
+ buffer1.flip();
+ final char buffer2Value = 'x';
+ buffer2.putChar(2, buffer2Value);
+ buffer2.position(4);
+ buffer2.flip();
- Collection<QpidByteBuffer> bufsForEmptyBytes = QpidByteBuffer.asQpidByteBuffers(new ByteArrayInputStream(new byte[]{}));
- assertEquals("Unexpected number of bufs for empty buffer", 0, bufsForEmptyBytes.size());
+ try (QpidByteBuffer concatenate = QpidByteBuffer.concatenate(buffer1, buffer2))
+ {
+ assertEquals("Unexpected capacity", 6, concatenate.capacity());
+ assertEquals("Unexpected position", 0, concatenate.position());
+ assertEquals("Unexpected limit", 6, concatenate.limit());
+ assertEquals("Unexpected value 1", buffer1Value, concatenate.getShort());
+ assertEquals("Unexpected value 2", buffer2Value, concatenate.getChar(4));
+ }
+ }
}
private void doDeflateInflate(byte[] input,
- Collection<QpidByteBuffer> inputBufs,
+ QpidByteBuffer inputBuf,
boolean direct) throws IOException
{
- Collection<QpidByteBuffer> deflatedBufs = QpidByteBuffer.deflate(inputBufs);
- assertNotNull(deflatedBufs);
+ try (QpidByteBuffer deflatedBuf = QpidByteBuffer.deflate(inputBuf))
+ {
+ assertNotNull(deflatedBuf);
- Collection<QpidByteBuffer> inflatedBufs = QpidByteBuffer.inflate(deflatedBufs);
- assertNotNull(inflatedBufs);
- assertTrue("Expected at least on buffer", inflatedBufs.size() >= 1);
+ try (QpidByteBuffer inflatedBuf = QpidByteBuffer.inflate(deflatedBuf))
+ {
+ assertNotNull(inflatedBuf);
- int bufNum = 1;
- int inputOffset = 0;
- int inflatedBytesTotal = 0;
- for(QpidByteBuffer inflatedBuf : inflatedBufs)
- {
- int inflatedBytesCount = inflatedBuf.remaining();
- inflatedBytesTotal += inflatedBytesCount;
+ int inflatedBytesCount = inflatedBuf.remaining();
- byte[] inflatedBytes = new byte[inflatedBytesCount];
- inflatedBuf.get(inflatedBytes);
- byte[] expectedBytes = Arrays.copyOfRange(input, inputOffset, inputOffset + inflatedBytes.length);
- Assert.assertArrayEquals("Inflated buf" + bufNum + " has unexpected content", expectedBytes, inflatedBytes);
+ byte[] inflatedBytes = new byte[inflatedBytesCount];
+ inflatedBuf.get(inflatedBytes);
+ byte[] expectedBytes = Arrays.copyOfRange(input, 0, inflatedBytes.length);
+ Assert.assertArrayEquals("Inflated buf has unexpected content", expectedBytes, inflatedBytes);
- inputOffset += inflatedBytes.length;
- bufNum++;
+ assertEquals("Unexpected number of inflated bytes", input.length, inflatedBytesCount);
+ }
}
-
- assertEquals("Unexpected number of inflated bytes", input.length, inflatedBytesTotal);
}
}
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/660c206d/broker-core/src/test/java/org/apache/qpid/server/protocol/v0_8/FieldTableTest.java
----------------------------------------------------------------------
diff --git a/broker-core/src/test/java/org/apache/qpid/server/protocol/v0_8/FieldTableTest.java b/broker-core/src/test/java/org/apache/qpid/server/protocol/v0_8/FieldTableTest.java
index 8a85614..40241c9 100644
--- a/broker-core/src/test/java/org/apache/qpid/server/protocol/v0_8/FieldTableTest.java
+++ b/broker-core/src/test/java/org/apache/qpid/server/protocol/v0_8/FieldTableTest.java
@@ -598,11 +598,11 @@ public class FieldTableTest extends QpidTestCase
buf.flip();
long length = buf.getInt() & 0xFFFFFFFFL;
- buf = buf.slice();
- buf.limit((int)length);
+ QpidByteBuffer bufSlice = buf.slice();
+ bufSlice.limit((int)length);
- FieldTable table2 = new FieldTable(buf);
+ FieldTable table2 = new FieldTable(bufSlice);
Assert.assertEquals((Boolean) true, table2.getBoolean("bool"));
Assert.assertEquals((Byte) Byte.MAX_VALUE, table2.getByte("byte"));
@@ -615,6 +615,10 @@ public class FieldTableTest extends QpidTestCase
Assert.assertEquals(Short.valueOf(Short.MAX_VALUE), table2.getShort("short"));
Assert.assertEquals("hello", table2.getString("string"));
Assert.assertNull(table2.getString("null-string"));
+ buf.dispose();
+ bufSlice.dispose();
+ table.dispose();
+ table2.dispose();
}
public void testEncodingSize()
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/660c206d/broker-core/src/test/java/org/apache/qpid/server/security/TrustStoreMessageSourceTest.java
----------------------------------------------------------------------
diff --git a/broker-core/src/test/java/org/apache/qpid/server/security/TrustStoreMessageSourceTest.java b/broker-core/src/test/java/org/apache/qpid/server/security/TrustStoreMessageSourceTest.java
index 19c16e4..978008e 100644
--- a/broker-core/src/test/java/org/apache/qpid/server/security/TrustStoreMessageSourceTest.java
+++ b/broker-core/src/test/java/org/apache/qpid/server/security/TrustStoreMessageSourceTest.java
@@ -19,10 +19,8 @@
package org.apache.qpid.server.security;
import static org.junit.Assert.assertArrayEquals;
-import static org.junit.Assert.assertNotNull;
import static org.mockito.Matchers.any;
import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.io.ByteArrayInputStream;
@@ -31,19 +29,18 @@ import java.io.ObjectInputStream;
import java.security.cert.Certificate;
import java.security.cert.CertificateEncodingException;
import java.util.ArrayList;
-import java.util.Collection;
import java.util.EnumSet;
import java.util.List;
import org.apache.qpid.server.bytebuffer.QpidByteBuffer;
import org.apache.qpid.server.consumer.ConsumerOption;
import org.apache.qpid.server.consumer.ConsumerTarget;
+import org.apache.qpid.server.message.MessageContainer;
import org.apache.qpid.server.message.MessageInstanceConsumer;
import org.apache.qpid.server.message.ServerMessage;
import org.apache.qpid.server.model.State;
import org.apache.qpid.server.model.TrustStore;
import org.apache.qpid.server.model.VirtualHost;
-import org.apache.qpid.server.message.MessageContainer;
import org.apache.qpid.server.store.MessageStore;
import org.apache.qpid.server.store.TestMemoryMessageStore;
import org.apache.qpid.test.utils.QpidTestCase;
@@ -96,19 +93,16 @@ public class TrustStoreMessageSourceTest extends QpidTestCase
{
final int bodySize = (int) message.getSize();
byte[] msgContent = new byte[bodySize];
- final Collection<QpidByteBuffer> allData = message.getStoredMessage().getContent(0, bodySize);
- int total = 0;
- for(QpidByteBuffer b : allData)
+ final List<String> certificates;
+ final ByteArrayInputStream bytesIn;
+ try (QpidByteBuffer allData = message.getStoredMessage().getContent(0, bodySize))
{
- int len = b.remaining();
- b.get(msgContent, total, len);
- b.dispose();
- total += len;
+ assertEquals("Unexpected message size was retrieved", bodySize, allData.remaining());
+ allData.get(msgContent);
}
- assertEquals("Unexpected message size was retrieved", bodySize, total);
- List<String> certificates = new ArrayList<>();
- ByteArrayInputStream bytesIn = new ByteArrayInputStream(msgContent);
+ certificates = new ArrayList<>();
+ bytesIn = new ByteArrayInputStream(msgContent);
try (ObjectInputStream is = new ObjectInputStream(bytesIn))
{
ArrayList<byte[]> encodedCertificates = (ArrayList<byte[]>) is.readObject();
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/660c206d/broker-core/src/test/java/org/apache/qpid/server/store/MessageStoreTestCase.java
----------------------------------------------------------------------
diff --git a/broker-core/src/test/java/org/apache/qpid/server/store/MessageStoreTestCase.java b/broker-core/src/test/java/org/apache/qpid/server/store/MessageStoreTestCase.java
index f2750f4..7f7f16a 100644
--- a/broker-core/src/test/java/org/apache/qpid/server/store/MessageStoreTestCase.java
+++ b/broker-core/src/test/java/org/apache/qpid/server/store/MessageStoreTestCase.java
@@ -366,18 +366,6 @@ public abstract class MessageStoreTestCase extends QpidTestCase
assertTrue("Message with id " + messageId3 + " is not found", enqueuedIds.contains(messageId3));
}
- public void testStoreIgnoresTransientMessage() throws Exception
- {
- long messageId = 1;
- int contentSize = 0;
- final StoredMessage<TestMessageMetaData> message = _store.addMessage(new TestMessageMetaData(messageId, contentSize, false)).allContentAdded();
-
- MessageHandler handler = mock(MessageHandler.class);
- _storeReader.visitMessages(handler);
-
- verify(handler, times(0)).handle(argThat(new MessageMetaDataMatcher(messageId)));
- }
-
public void testAddAndRemoveMessageWithoutContent() throws Exception
{
long messageId = 1;
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/660c206d/broker-core/src/test/java/org/apache/qpid/server/store/TestMessageMetaDataFactory.java
----------------------------------------------------------------------
diff --git a/broker-core/src/test/java/org/apache/qpid/server/store/TestMessageMetaDataFactory.java b/broker-core/src/test/java/org/apache/qpid/server/store/TestMessageMetaDataFactory.java
index ee67dd2..1abf0c0 100644
--- a/broker-core/src/test/java/org/apache/qpid/server/store/TestMessageMetaDataFactory.java
+++ b/broker-core/src/test/java/org/apache/qpid/server/store/TestMessageMetaDataFactory.java
@@ -20,19 +20,16 @@
*/
package org.apache.qpid.server.store;
-import java.util.List;
-
import org.apache.qpid.server.bytebuffer.QpidByteBuffer;
-import org.apache.qpid.server.bytebuffer.QpidByteBufferUtils;
import org.apache.qpid.server.plugin.MessageMetaDataType;
public class TestMessageMetaDataFactory implements MessageMetaDataType.Factory<TestMessageMetaData>
{
@Override
- public TestMessageMetaData createMetaData(List<QpidByteBuffer> bufs)
+ public TestMessageMetaData createMetaData(QpidByteBuffer buf)
{
- long id = QpidByteBufferUtils.getLong(bufs);
- int size = QpidByteBufferUtils.getInt(bufs);
+ long id = buf.getLong();
+ int size = buf.getInt();
return new TestMessageMetaData(id, size);
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/660c206d/broker-core/src/test/java/org/apache/qpid/server/store/TestMessageMetaDataType.java
----------------------------------------------------------------------
diff --git a/broker-core/src/test/java/org/apache/qpid/server/store/TestMessageMetaDataType.java b/broker-core/src/test/java/org/apache/qpid/server/store/TestMessageMetaDataType.java
index 50d9428..231dd35 100644
--- a/broker-core/src/test/java/org/apache/qpid/server/store/TestMessageMetaDataType.java
+++ b/broker-core/src/test/java/org/apache/qpid/server/store/TestMessageMetaDataType.java
@@ -20,9 +20,6 @@
*/
package org.apache.qpid.server.store;
-import java.util.Collection;
-import java.util.List;
-
import org.apache.qpid.server.bytebuffer.QpidByteBuffer;
import org.apache.qpid.server.message.AMQMessageHeader;
import org.apache.qpid.server.message.MessageReference;
@@ -42,9 +39,9 @@ public class TestMessageMetaDataType implements MessageMetaDataType<TestMessageM
}
@Override
- public TestMessageMetaData createMetaData(List<QpidByteBuffer> bufs)
+ public TestMessageMetaData createMetaData(QpidByteBuffer buf)
{
- return TestMessageMetaData.FACTORY.createMetaData(bufs);
+ return TestMessageMetaData.FACTORY.createMetaData(buf);
}
@Override
@@ -108,7 +105,13 @@ public class TestMessageMetaDataType implements MessageMetaDataType<TestMessageM
}
@Override
- public Collection<QpidByteBuffer> getContent(int offset, int length)
+ public QpidByteBuffer getContent()
+ {
+ return null;
+ }
+
+ @Override
+ public QpidByteBuffer getContent(int offset, int length)
{
return null;
}
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/660c206d/broker-core/src/test/java/org/apache/qpid/server/txn/MockServerMessage.java
----------------------------------------------------------------------
diff --git a/broker-core/src/test/java/org/apache/qpid/server/txn/MockServerMessage.java b/broker-core/src/test/java/org/apache/qpid/server/txn/MockServerMessage.java
index da61a64..3ccc94b 100644
--- a/broker-core/src/test/java/org/apache/qpid/server/txn/MockServerMessage.java
+++ b/broker-core/src/test/java/org/apache/qpid/server/txn/MockServerMessage.java
@@ -20,8 +20,6 @@
*/
package org.apache.qpid.server.txn;
-import java.util.Collection;
-
import org.apache.qpid.server.bytebuffer.QpidByteBuffer;
import org.apache.qpid.server.message.AMQMessageHeader;
@@ -127,7 +125,13 @@ class MockServerMessage implements ServerMessage
}
@Override
- public Collection<QpidByteBuffer> getContent(int offset, int length)
+ public QpidByteBuffer getContent()
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public QpidByteBuffer getContent(int offset, int length)
{
throw new UnsupportedOperationException();
}
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/660c206d/broker-core/src/test/java/org/apache/qpid/server/virtualhost/VirtualHostPropertiesNodeTest.java
----------------------------------------------------------------------
diff --git a/broker-core/src/test/java/org/apache/qpid/server/virtualhost/VirtualHostPropertiesNodeTest.java b/broker-core/src/test/java/org/apache/qpid/server/virtualhost/VirtualHostPropertiesNodeTest.java
index d5e184d..9af3300 100644
--- a/broker-core/src/test/java/org/apache/qpid/server/virtualhost/VirtualHostPropertiesNodeTest.java
+++ b/broker-core/src/test/java/org/apache/qpid/server/virtualhost/VirtualHostPropertiesNodeTest.java
@@ -58,5 +58,9 @@ public class VirtualHostPropertiesNodeTest extends QpidTestCase
MessageInstanceConsumer consumer = _virtualHostPropertiesNode.addConsumer(target, null, ServerMessage.class, getTestName(), options, 0);
final MessageContainer messageContainer = consumer.pullMessage();
assertNotNull("Could not pull message from VirtualHostPropertyNode", messageContainer);
+ if (messageContainer.getMessageReference() != null)
+ {
+ messageContainer.getMessageReference().release();
+ }
}
}
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/660c206d/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ConsumerTarget_0_10.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ConsumerTarget_0_10.java b/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ConsumerTarget_0_10.java
index e342db7..ce5551f 100644
--- a/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ConsumerTarget_0_10.java
+++ b/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ConsumerTarget_0_10.java
@@ -21,10 +21,8 @@
package org.apache.qpid.server.protocol.v0_10;
import java.io.IOException;
-import java.util.Collection;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicLong;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -39,16 +37,9 @@ import org.apache.qpid.server.message.MessageInstance.ConsumerAcquiredState;
import org.apache.qpid.server.message.MessageInstance.EntryState;
import org.apache.qpid.server.message.MessageInstanceConsumer;
import org.apache.qpid.server.message.ServerMessage;
-import org.apache.qpid.server.model.Exchange;
import org.apache.qpid.server.model.Queue;
import org.apache.qpid.server.plugin.MessageConverter;
import org.apache.qpid.server.protocol.MessageConverterRegistry;
-import org.apache.qpid.server.store.TransactionLogResource;
-import org.apache.qpid.server.txn.AutoCommitTransaction;
-import org.apache.qpid.server.txn.ServerTransaction;
-import org.apache.qpid.server.util.Action;
-import org.apache.qpid.server.util.ConnectionScopedRuntimeException;
-import org.apache.qpid.server.util.StateChangeListener;
import org.apache.qpid.server.protocol.v0_10.transport.DeliveryProperties;
import org.apache.qpid.server.protocol.v0_10.transport.Header;
import org.apache.qpid.server.protocol.v0_10.transport.MessageAcceptMode;
@@ -59,8 +50,13 @@ import org.apache.qpid.server.protocol.v0_10.transport.MessageProperties;
import org.apache.qpid.server.protocol.v0_10.transport.MessageTransfer;
import org.apache.qpid.server.protocol.v0_10.transport.Method;
import org.apache.qpid.server.protocol.v0_10.transport.Option;
-import org.apache.qpid.server.util.ByteBufferUtils;
+import org.apache.qpid.server.store.TransactionLogResource;
+import org.apache.qpid.server.txn.AutoCommitTransaction;
+import org.apache.qpid.server.txn.ServerTransaction;
+import org.apache.qpid.server.util.Action;
+import org.apache.qpid.server.util.ConnectionScopedRuntimeException;
import org.apache.qpid.server.util.GZIPUtils;
+import org.apache.qpid.server.util.StateChangeListener;
public class ConsumerTarget_0_10 extends AbstractConsumerTarget<ConsumerTarget_0_10>
{
@@ -254,50 +250,41 @@ public class ConsumerTarget_0_10 extends AbstractConsumerTarget<ConsumerTarget_0
boolean msgCompressed = messageProps != null && GZIPUtils.GZIP_CONTENT_ENCODING.equals(messageProps.getContentEncoding());
- Collection<QpidByteBuffer> bodyBuffers = msg.getBody();
+ QpidByteBuffer bodyBuffer = msg.getBody();
boolean compressionSupported = _session.getConnection().getConnectionDelegate().isCompressionSupported();
- if(msgCompressed && !compressionSupported && bodyBuffers != null)
+ if(msgCompressed && !compressionSupported && bodyBuffer != null)
{
- Collection<QpidByteBuffer> uncompressedBuffers = inflateIfPossible(bodyBuffers);
+ QpidByteBuffer uncompressedBuffer = inflateIfPossible(bodyBuffer);
messageProps.setContentEncoding(null);
- for (QpidByteBuffer buf : bodyBuffers)
- {
- buf.dispose();
- }
- bodyBuffers = uncompressedBuffers;
+ bodyBuffer.dispose();
+ bodyBuffer = uncompressedBuffer;
}
else if(!msgCompressed
&& compressionSupported
&& (messageProps == null || messageProps.getContentEncoding() == null)
- && bodyBuffers != null
- && ByteBufferUtils.remaining(bodyBuffers) > _session.getConnection().getMessageCompressionThreshold())
+ && bodyBuffer != null
+ && bodyBuffer.remaining() > _session.getConnection().getMessageCompressionThreshold())
{
- Collection<QpidByteBuffer> compressedBuffers = deflateIfPossible(bodyBuffers);
+ QpidByteBuffer compressedBuffers = deflateIfPossible(bodyBuffer);
if(messageProps == null)
{
messageProps = new MessageProperties();
}
messageProps.setContentEncoding(GZIPUtils.GZIP_CONTENT_ENCODING);
- for (QpidByteBuffer buf : bodyBuffers)
- {
- buf.dispose();
- }
- bodyBuffers = compressedBuffers;
+ bodyBuffer.dispose();
+ bodyBuffer = compressedBuffers;
}
Header header = new Header(deliveryProps, messageProps, msg.getHeader() == null ? null : msg.getHeader().getNonStandardProperties());
- xfr = batch ? new MessageTransfer(_name,_acceptMode,_acquireMode,header, bodyBuffers, BATCHED)
- : new MessageTransfer(_name,_acceptMode,_acquireMode,header, bodyBuffers);
- if (bodyBuffers != null)
+ xfr = batch ? new MessageTransfer(_name, _acceptMode, _acquireMode, header, bodyBuffer, BATCHED)
+ : new MessageTransfer(_name, _acceptMode, _acquireMode, header, bodyBuffer);
+ if (bodyBuffer != null)
{
- for (QpidByteBuffer buf : bodyBuffers)
- {
- buf.dispose();
- }
- bodyBuffers = null;
+ bodyBuffer.dispose();
+ bodyBuffer = null;
}
if(_acceptMode == MessageAcceptMode.NONE && _acquireMode != MessageAcquireMode.PRE_ACQUIRED)
{
@@ -612,11 +599,11 @@ public class ConsumerTarget_0_10 extends AbstractConsumerTarget<ConsumerTarget_0
}
- private Collection<QpidByteBuffer> deflateIfPossible(final Collection<QpidByteBuffer> buffers)
+ private QpidByteBuffer deflateIfPossible(final QpidByteBuffer buffer)
{
try
{
- return QpidByteBuffer.deflate(buffers);
+ return QpidByteBuffer.deflate(buffer);
}
catch (IOException e)
{
@@ -625,11 +612,11 @@ public class ConsumerTarget_0_10 extends AbstractConsumerTarget<ConsumerTarget_0
}
}
- private Collection<QpidByteBuffer> inflateIfPossible(final Collection<QpidByteBuffer> buffers)
+ private QpidByteBuffer inflateIfPossible(final QpidByteBuffer buffer)
{
try
{
- return QpidByteBuffer.inflate(buffers);
+ return QpidByteBuffer.inflate(buffer);
}
catch (IOException e)
{
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/660c206d/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageConverter_Internal_to_v0_10.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageConverter_Internal_to_v0_10.java b/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageConverter_Internal_to_v0_10.java
index 4ffbd3f..114b0f2 100644
--- a/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageConverter_Internal_to_v0_10.java
+++ b/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageConverter_Internal_to_v0_10.java
@@ -22,8 +22,6 @@ package org.apache.qpid.server.protocol.v0_10;
import static java.nio.charset.StandardCharsets.UTF_8;
-import java.util.Collection;
-import java.util.Collections;
import java.util.UUID;
import org.apache.qpid.server.bytebuffer.QpidByteBuffer;
@@ -107,9 +105,9 @@ public class MessageConverter_Internal_to_v0_10 implements MessageConverter<Inte
}
@Override
- public Collection<QpidByteBuffer> getContent(final int offset, final int length)
+ public QpidByteBuffer getContent(final int offset, final int length)
{
- return Collections.singleton(QpidByteBuffer.wrap(messageContent, offset, length));
+ return QpidByteBuffer.wrap(messageContent, offset, length);
}
@Override
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/660c206d/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageConverter_v0_10.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageConverter_v0_10.java b/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageConverter_v0_10.java
index dfdbc9a..8c9db87 100644
--- a/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageConverter_v0_10.java
+++ b/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageConverter_v0_10.java
@@ -22,8 +22,6 @@ package org.apache.qpid.server.protocol.v0_10;
import static java.nio.charset.StandardCharsets.UTF_8;
-import java.util.Collection;
-
import org.apache.qpid.server.bytebuffer.QpidByteBuffer;
import org.apache.qpid.server.message.ServerMessage;
import org.apache.qpid.server.model.NamedAddressSpace;
@@ -83,7 +81,7 @@ public class MessageConverter_v0_10 implements MessageConverter<ServerMessage, M
}
@Override
- public Collection<QpidByteBuffer> getContent(final int offset, final int length)
+ public QpidByteBuffer getContent(final int offset, final int length)
{
return serverMsg.getContent(offset, length);
}
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/660c206d/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageConverter_v0_10_to_Internal.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageConverter_v0_10_to_Internal.java b/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageConverter_v0_10_to_Internal.java
index 0610d09..8121dd6 100644
--- a/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageConverter_v0_10_to_Internal.java
+++ b/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageConverter_v0_10_to_Internal.java
@@ -67,13 +67,9 @@ public class MessageConverter_v0_10_to_Internal implements MessageConverter<Mess
{
final String mimeType = serverMessage.getMessageHeader().getMimeType();
byte[] data = new byte[(int) serverMessage.getSize()];
- int total = 0;
- for(QpidByteBuffer b : serverMessage.getContent(0, (int) serverMessage.getSize()))
+ try (QpidByteBuffer content = serverMessage.getContent())
{
- int len = b.remaining();
- b.get(data, total, len);
- b.dispose();
- total += len;
+ content.get(data);
}
String encoding = serverMessage.getMessageHeader().getEncoding();
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/660c206d/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageFormat_0_10.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageFormat_0_10.java b/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageFormat_0_10.java
index f74e291..de9afcc 100644
--- a/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageFormat_0_10.java
+++ b/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageFormat_0_10.java
@@ -27,14 +27,14 @@ import java.util.List;
import org.apache.qpid.server.bytebuffer.QpidByteBuffer;
import org.apache.qpid.server.plugin.MessageFormat;
import org.apache.qpid.server.plugin.PluggableService;
-import org.apache.qpid.server.store.MessageHandle;
-import org.apache.qpid.server.store.MessageStore;
-import org.apache.qpid.server.store.StoredMessage;
-import org.apache.qpid.server.util.ConnectionScopedRuntimeException;
import org.apache.qpid.server.protocol.v0_10.transport.DeliveryProperties;
import org.apache.qpid.server.protocol.v0_10.transport.Header;
import org.apache.qpid.server.protocol.v0_10.transport.MessageProperties;
import org.apache.qpid.server.protocol.v0_10.transport.Struct;
+import org.apache.qpid.server.store.MessageHandle;
+import org.apache.qpid.server.store.MessageStore;
+import org.apache.qpid.server.store.StoredMessage;
+import org.apache.qpid.server.util.ConnectionScopedRuntimeException;
@PluggableService
public class MessageFormat_0_10 implements MessageFormat<MessageTransferMessage>
@@ -63,7 +63,7 @@ public class MessageFormat_0_10 implements MessageFormat<MessageTransferMessage>
// format: <int header count> <headers> <body>
@Override
- public List<QpidByteBuffer> convertToMessageFormat(final MessageTransferMessage message)
+ public QpidByteBuffer convertToMessageFormat(final MessageTransferMessage message)
{
ServerEncoder encoder = new ServerEncoder(4096, true);
Struct[] structs = message.getHeader().getStructs();
@@ -72,22 +72,21 @@ public class MessageFormat_0_10 implements MessageFormat<MessageTransferMessage>
{
encoder.writeStruct32(struct);
}
- final QpidByteBuffer headerBuf = encoder.getBuffer();
- List<QpidByteBuffer> bufs = new ArrayList<>();
- bufs.add(headerBuf);
- bufs.addAll(message.getContent(0, (int) message.getSize()));
-
- return bufs;
+ try (QpidByteBuffer headerBuf = encoder.getBuffer();
+ QpidByteBuffer content = message.getContent())
+ {
+ return QpidByteBuffer.concatenate(headerBuf, content);
+ }
}
@Override
- public MessageTransferMessage createMessage(final List<QpidByteBuffer> buf,
+ public MessageTransferMessage createMessage(final QpidByteBuffer payload,
final MessageStore store,
final Object connectionReference)
{
try
{
- ServerDecoder serverDecoder = new ServerDecoder(buf);
+ ServerDecoder serverDecoder = new ServerDecoder(payload);
int headerCount = serverDecoder.readInt32();
DeliveryProperties deliveryProperties = null;
MessageProperties messageProperties = null;
@@ -113,20 +112,11 @@ public class MessageFormat_0_10 implements MessageFormat<MessageTransferMessage>
}
}
Header header = new Header(deliveryProperties, messageProperties, nonStandard);
- int bodySize = 0;
- for(QpidByteBuffer content : buf)
- {
- bodySize += content.remaining();
- }
- MessageMetaData_0_10 metaData = new MessageMetaData_0_10(header, bodySize, System.currentTimeMillis());
+ MessageMetaData_0_10 metaData = new MessageMetaData_0_10(header,
+ payload.remaining(),
+ System.currentTimeMillis());
final MessageHandle<MessageMetaData_0_10> handle = store.addMessage(metaData);
- for (QpidByteBuffer content : buf)
- {
- if (content.hasRemaining())
- {
- handle.addContent(content);
- }
- }
+ handle.addContent(payload);
final StoredMessage<MessageMetaData_0_10> storedMessage = handle.allContentAdded();
return new MessageTransferMessage(storedMessage, connectionReference);
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/660c206d/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageMetaDataType_0_10.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageMetaDataType_0_10.java b/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageMetaDataType_0_10.java
index 2dcb5f5..bc3cbed 100644
--- a/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageMetaDataType_0_10.java
+++ b/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageMetaDataType_0_10.java
@@ -20,8 +20,6 @@
*/
package org.apache.qpid.server.protocol.v0_10;
-import java.util.List;
-
import org.apache.qpid.server.bytebuffer.QpidByteBuffer;
import org.apache.qpid.server.message.ServerMessage;
import org.apache.qpid.server.plugin.MessageMetaDataType;
@@ -42,9 +40,9 @@ public class MessageMetaDataType_0_10 implements MessageMetaDataType<MessageMeta
}
@Override
- public MessageMetaData_0_10 createMetaData(List<QpidByteBuffer> bufs)
+ public MessageMetaData_0_10 createMetaData(QpidByteBuffer buf)
{
- return MessageMetaData_0_10.FACTORY.createMetaData(bufs);
+ return MessageMetaData_0_10.FACTORY.createMetaData(buf);
}
@Override
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/660c206d/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageMetaData_0_10.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageMetaData_0_10.java b/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageMetaData_0_10.java
index 2861cc2..d51b158 100755
--- a/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageMetaData_0_10.java
+++ b/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageMetaData_0_10.java
@@ -263,7 +263,7 @@ public class MessageMetaData_0_10 implements StorableMessageMetaData
private static class MetaDataFactory implements MessageMetaDataType.Factory<MessageMetaData_0_10>
{
@Override
- public MessageMetaData_0_10 createMetaData(List<QpidByteBuffer> buf)
+ public MessageMetaData_0_10 createMetaData(QpidByteBuffer buf)
{
ServerDecoder decoder = new ServerDecoder(buf);
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/660c206d/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageTransferMessage.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageTransferMessage.java b/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageTransferMessage.java
index e6b2c8d..ba35c01 100644
--- a/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageTransferMessage.java
+++ b/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageTransferMessage.java
@@ -20,14 +20,12 @@
*/
package org.apache.qpid.server.protocol.v0_10;
-import java.util.Collection;
-
import org.apache.qpid.server.bytebuffer.QpidByteBuffer;
import org.apache.qpid.server.message.AMQMessageHeader;
import org.apache.qpid.server.message.AbstractServerMessageImpl;
+import org.apache.qpid.server.protocol.v0_10.transport.Header;
import org.apache.qpid.server.store.StoredMessage;
import org.apache.qpid.server.store.TransactionLogResource;
-import org.apache.qpid.server.protocol.v0_10.transport.Header;
public class MessageTransferMessage extends AbstractServerMessageImpl<MessageTransferMessage, MessageMetaData_0_10>
@@ -102,8 +100,8 @@ public class MessageTransferMessage extends AbstractServerMessageImpl<MessageTra
return getMetaData().getHeader();
}
- public Collection<QpidByteBuffer> getBody()
+ public QpidByteBuffer getBody()
{
- return getContent(0, (int) getSize());
+ return getContent();
}
}
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/660c206d/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerAssembler.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerAssembler.java b/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerAssembler.java
index f3dde0d..abe0842 100644
--- a/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerAssembler.java
+++ b/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerAssembler.java
@@ -26,9 +26,7 @@ import java.security.AccessControlContext;
import java.security.AccessController;
import java.security.PrivilegedAction;
import java.util.ArrayList;
-import java.util.Collections;
import java.util.HashMap;
-import java.util.Iterator;
import java.util.List;
import java.util.Map;
@@ -39,6 +37,7 @@ import org.slf4j.LoggerFactory;
import org.apache.qpid.server.bytebuffer.QpidByteBuffer;
import org.apache.qpid.server.protocol.v0_10.transport.DeliveryProperties;
+import org.apache.qpid.server.protocol.v0_10.transport.Frame;
import org.apache.qpid.server.protocol.v0_10.transport.Header;
import org.apache.qpid.server.protocol.v0_10.transport.MessageProperties;
import org.apache.qpid.server.protocol.v0_10.transport.Method;
@@ -46,7 +45,6 @@ import org.apache.qpid.server.protocol.v0_10.transport.ProtocolError;
import org.apache.qpid.server.protocol.v0_10.transport.ProtocolEvent;
import org.apache.qpid.server.protocol.v0_10.transport.ProtocolHeader;
import org.apache.qpid.server.protocol.v0_10.transport.Struct;
-import org.apache.qpid.server.protocol.v0_10.transport.Frame;
public class ServerAssembler
{
@@ -77,35 +75,54 @@ public class ServerAssembler
{
PeekingIterator<ServerFrame> itr = Iterators.peekingIterator(frames.iterator());
- while(itr.hasNext())
+ boolean cleanExit = false;
+ try
{
- final ServerFrame frame = itr.next();
- final int frameChannel = frame.getChannel();
-
- ServerSession channel = _connection.getSession(frameChannel);
- if (channel != null)
+ while(itr.hasNext())
{
- final AccessControlContext context = channel.getAccessControllerContext();
- AccessController.doPrivileged((PrivilegedAction<Void>) () ->
+ final ServerFrame frame = itr.next();
+ final int frameChannel = frame.getChannel();
+
+ ServerSession channel = _connection.getSession(frameChannel);
+ if (channel != null)
{
- ServerFrame channelFrame = frame;
- boolean nextIsSameChannel;
- do
+ final AccessControlContext context = channel.getAccessControllerContext();
+ AccessController.doPrivileged((PrivilegedAction<Void>) () ->
{
- received(channelFrame);
- nextIsSameChannel = itr.hasNext() && frameChannel == itr.peek().getChannel();
- if (nextIsSameChannel)
+ ServerFrame channelFrame = frame;
+ boolean nextIsSameChannel;
+ do
{
- channelFrame = itr.next();
+ received(channelFrame);
+ nextIsSameChannel = itr.hasNext() && frameChannel == itr.peek().getChannel();
+ if (nextIsSameChannel)
+ {
+ channelFrame = itr.next();
+ }
}
- }
- while (nextIsSameChannel);
- return null;
- }, context);
+ while (nextIsSameChannel);
+ return null;
+ }, context);
+ }
+ else
+ {
+ received(frame);
+ }
}
- else
+ cleanExit = true;
+ }
+ finally
+ {
+ if (!cleanExit)
{
- received(frame);
+ while (itr.hasNext())
+ {
+ final QpidByteBuffer body = itr.next().getBody();
+ if (body != null)
+ {
+ body.dispose();
+ }
+ }
}
}
}
@@ -183,11 +200,9 @@ public class ServerAssembler
public void frame(ServerFrame frame)
{
- List<QpidByteBuffer> frameBuffers;
if (frame.isFirstFrame() && frame.isLastFrame())
{
- frameBuffers = Collections.singletonList(frame.getBody());
- assemble(frame, frameBuffers);
+ assemble(frame, frame.getBody());
}
else
{
@@ -207,98 +222,103 @@ public class ServerAssembler
if (frame.isLastFrame())
{
clearSegment(frame);
- frameBuffers = new ArrayList<>(frames.size());
+ List<QpidByteBuffer> frameBuffers = new ArrayList<>(frames.size());
for (ServerFrame f : frames)
{
-
frameBuffers.add(f.getBody());
}
- assemble(frame, frameBuffers);
+ QpidByteBuffer combined = QpidByteBuffer.concatenate(frameBuffers);
+ for (QpidByteBuffer buffer : frameBuffers)
+ {
+ buffer.dispose();
+ }
+ assemble(frame, combined);
}
}
}
- private void assemble(ServerFrame frame, List<QpidByteBuffer> frameBuffers)
+ private void assemble(ServerFrame frame, QpidByteBuffer frameBuffer)
{
- ServerDecoder dec = new ServerDecoder(frameBuffers);
+ try
+ {
+ ServerDecoder dec = new ServerDecoder(frameBuffer);
- int channel = frame.getChannel();
- Method command;
+ int channel = frame.getChannel();
+ Method command;
- switch (frame.getType())
- {
- case CONTROL:
- int controlType = dec.readUint16();
- Method control = Method.create(controlType);
- control.read(dec);
- emit(channel, control);
- break;
- case COMMAND:
- int commandType = dec.readUint16();
- // read in the session header, right now we don't use it
- int hdr = dec.readUint16();
- command = Method.create(commandType);
- command.setSync((0x0001 & hdr) != 0);
- command.read(dec);
- if (command.hasPayload() && !frame.isLastSegment())
- {
- setIncompleteCommand(channel, command);
- }
- else
- {
- emit(channel, command);
- }
- break;
- case HEADER:
- command = getIncompleteCommand(channel);
- List<Struct> structs = null;
- DeliveryProperties deliveryProps = null;
- MessageProperties messageProps = null;
-
- while (dec.hasRemaining())
- {
- Struct struct = dec.readStruct32();
- if(struct instanceof DeliveryProperties && deliveryProps == null)
+ switch (frame.getType())
+ {
+ case CONTROL:
+ int controlType = dec.readUint16();
+ Method control = Method.create(controlType);
+ control.read(dec);
+ emit(channel, control);
+ break;
+ case COMMAND:
+ int commandType = dec.readUint16();
+ // read in the session header, right now we don't use it
+ int hdr = dec.readUint16();
+ command = Method.create(commandType);
+ command.setSync((0x0001 & hdr) != 0);
+ command.read(dec);
+ if (command.hasPayload() && !frame.isLastSegment())
{
- deliveryProps = (DeliveryProperties) struct;
+ setIncompleteCommand(channel, command);
}
- else if(struct instanceof MessageProperties && messageProps == null)
+ else
{
- messageProps = (MessageProperties) struct;
+ emit(channel, command);
}
- else
+ break;
+ case HEADER:
+ command = getIncompleteCommand(channel);
+ List<Struct> structs = null;
+ DeliveryProperties deliveryProps = null;
+ MessageProperties messageProps = null;
+
+ while (dec.hasRemaining())
{
- if(structs == null)
+ Struct struct = dec.readStruct32();
+ if (struct instanceof DeliveryProperties && deliveryProps == null)
+ {
+ deliveryProps = (DeliveryProperties) struct;
+ }
+ else if (struct instanceof MessageProperties && messageProps == null)
+ {
+ messageProps = (MessageProperties) struct;
+ }
+ else
{
- structs = new ArrayList<>(2);
+ if (structs == null)
+ {
+ structs = new ArrayList<>(2);
+ }
+ structs.add(struct);
}
- structs.add(struct);
}
+ command.setHeader(new Header(deliveryProps, messageProps, structs));
- }
- command.setHeader(new Header(deliveryProps,messageProps,structs));
-
- if (frame.isLastSegment())
- {
+ if (frame.isLastSegment())
+ {
+ setIncompleteCommand(channel, null);
+ emit(channel, command);
+ }
+ break;
+ case BODY:
+ command = getIncompleteCommand(channel);
+ command.setBody(frameBuffer);
setIncompleteCommand(channel, null);
emit(channel, command);
- }
- break;
- case BODY:
- command = getIncompleteCommand(channel);
- command.setBody(frameBuffers);
- setIncompleteCommand(channel, null);
- emit(channel, command);
-
- break;
- default:
- throw new IllegalStateException("unknown frame type: " + frame.getType());
- }
- for(QpidByteBuffer buf : frameBuffers)
+ break;
+ default:
+ throw new IllegalStateException("unknown frame type: " + frame.getType());
+ }
+ }
+ finally
{
- buf.dispose();
+ frameBuffer.dispose();
}
}
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/660c206d/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnection.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnection.java b/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnection.java
index 11e2ab5..8032b35 100644
--- a/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnection.java
+++ b/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnection.java
@@ -61,13 +61,15 @@ import org.apache.qpid.server.protocol.v0_10.transport.ConnectionCloseOk;
import org.apache.qpid.server.protocol.v0_10.transport.ConnectionException;
import org.apache.qpid.server.protocol.v0_10.transport.ExecutionErrorCode;
import org.apache.qpid.server.protocol.v0_10.transport.ExecutionException;
+import org.apache.qpid.server.protocol.v0_10.transport.MessageTransfer;
import org.apache.qpid.server.protocol.v0_10.transport.Method;
import org.apache.qpid.server.protocol.v0_10.transport.Option;
import org.apache.qpid.server.protocol.v0_10.transport.ProtocolEvent;
import org.apache.qpid.server.protocol.v0_10.transport.SessionDetachCode;
import org.apache.qpid.server.protocol.v0_10.transport.SessionDetached;
import org.apache.qpid.server.session.AMQPSession;
-import org.apache.qpid.server.transport.*;
+import org.apache.qpid.server.transport.AMQPConnection;
+import org.apache.qpid.server.transport.ServerNetworkConnection;
import org.apache.qpid.server.transport.network.NetworkConnection;
import org.apache.qpid.server.util.Action;
import org.apache.qpid.server.util.ServerScopedRuntimeException;
@@ -276,6 +278,13 @@ public class ServerConnection extends ConnectionInvoker
}
event.delegate(this, delegate);
}
+ else
+ {
+ if (event instanceof MessageTransfer)
+ {
+ ((MessageTransfer) event).dispose();
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/660c206d/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerDecoder.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerDecoder.java b/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerDecoder.java
index 04819f0..5d26d59 100644
--- a/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerDecoder.java
+++ b/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerDecoder.java
@@ -20,137 +20,58 @@
*/
package org.apache.qpid.server.protocol.v0_10;
-import java.util.List;
-
import org.apache.qpid.server.bytebuffer.QpidByteBuffer;
import org.apache.qpid.server.protocol.v0_10.transport.AbstractDecoder;
final class ServerDecoder extends AbstractDecoder
{
- private final List<QpidByteBuffer> _underlying;
- private int _bufferIndex;
+ private final QpidByteBuffer _underlying;
- ServerDecoder(List<QpidByteBuffer> in)
+ ServerDecoder(QpidByteBuffer in)
{
_underlying = in;
- _bufferIndex = 0;
- }
-
- private void advanceIfNecessary()
- {
- while(!getCurrentBuffer().hasRemaining() && _bufferIndex != _underlying.size()-1)
- {
- _bufferIndex++;
- }
- }
-
- private QpidByteBuffer getBuffer(int size)
- {
- advanceIfNecessary();
- final QpidByteBuffer currentBuffer = getCurrentBuffer();
- if(currentBuffer.remaining()>= size)
- {
- return currentBuffer;
- }
- else
- {
- return readAsQpidByteBuffer(size);
- }
- }
-
- private QpidByteBuffer readAsQpidByteBuffer(int len)
- {
- QpidByteBuffer currentBuffer = getCurrentBuffer();
- if(currentBuffer.remaining()>=len)
- {
- QpidByteBuffer buf = currentBuffer.slice();
- buf.limit(len);
- currentBuffer.position(currentBuffer.position()+len);
- return buf;
- }
- else
- {
- QpidByteBuffer dest = QpidByteBuffer.allocate(len);
- while(dest.hasRemaining() && available()>0)
- {
- advanceIfNecessary();
- currentBuffer = getCurrentBuffer();
- final int remaining = dest.remaining();
- if(currentBuffer.remaining()>= remaining)
- {
- QpidByteBuffer buf = currentBuffer.slice();
- buf.limit(remaining);
- currentBuffer.position(currentBuffer.position()+remaining);
- dest.put(buf);
- buf.dispose();
- }
- else
- {
- dest.put(currentBuffer);
- }
- }
-
- dest.flip();
- return dest;
- }
- }
-
- private int available()
- {
- int remaining = 0;
- for(int i = _bufferIndex; i < _underlying.size(); i++)
- {
- remaining += _underlying.get(i).remaining();
- }
- return remaining;
- }
-
-
- private QpidByteBuffer getCurrentBuffer()
- {
- return _underlying.get(_bufferIndex);
}
@Override
protected byte doGet()
{
- return getBuffer(1).get();
+ return _underlying.get();
}
@Override
protected void doGet(byte[] bytes)
{
- getBuffer(bytes.length).get(bytes);
+ _underlying.get(bytes);
}
@Override
public boolean hasRemaining()
{
- return available() != 0;
+ return _underlying.remaining() != 0;
}
@Override
public short readUint8()
{
- return (short) (0xFF & getBuffer(1).get());
+ return _underlying.getUnsignedByte();
}
@Override
public int readUint16()
{
- return 0xFFFF & getBuffer(2).getShort();
+ return _underlying.getUnsignedShort();
}
@Override
public long readUint32()
{
- return 0xFFFFFFFFL & getBuffer(4).getInt();
+ return _underlying.getUnsignedInt();
}
@Override
public long readUint64()
{
- return getBuffer(8).getLong();
+ return _underlying.getLong();
}
@Override
@@ -172,44 +93,44 @@ final class ServerDecoder extends AbstractDecoder
@Override
public double readDouble()
{
- return getBuffer(8).getDouble();
+ return _underlying.getDouble();
}
@Override
public float readFloat()
{
- return getBuffer(4).getFloat();
+ return _underlying.getFloat();
}
@Override
public short readInt16()
{
- return getBuffer(2).getShort();
+ return _underlying.getShort();
}
@Override
public int readInt32()
{
- return getBuffer(4).getInt();
+ return _underlying.getInt();
}
@Override
public byte readInt8()
{
- return getBuffer(1).get();
+ return _underlying.get();
}
@Override
public byte[] readRemainingBytes()
{
- byte[] result = new byte[available()];
- get(result);
- return result;
- }
+ byte[] result = new byte[_underlying.remaining()];
+ get(result);
+ return result;
+ }
@Override
public long readInt64()
{
- return getBuffer(8).getLong();
+ return _underlying.getLong();
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org