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/11/10 15:50:49 UTC
[2/4] qpid-jms-amqp-0-x git commit: QPID-7725: [Java Client,
AMQP 0-x] Remove QpidByteBuffer
http://git-wip-us.apache.org/repos/asf/qpid-jms-amqp-0-x/blob/95203e5d/client/src/main/java/org/apache/qpid/framing/ContentHeaderBody.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/qpid/framing/ContentHeaderBody.java b/client/src/main/java/org/apache/qpid/framing/ContentHeaderBody.java
index 0d07817..a263215 100644
--- a/client/src/main/java/org/apache/qpid/framing/ContentHeaderBody.java
+++ b/client/src/main/java/org/apache/qpid/framing/ContentHeaderBody.java
@@ -20,10 +20,12 @@
*/
package org.apache.qpid.framing;
+import java.nio.ByteBuffer;
+
import org.apache.qpid.QpidException;
-import org.apache.qpid.bytebuffer.QpidByteBuffer;
import org.apache.qpid.protocol.AMQVersionAwareProtocolSession;
import org.apache.qpid.transport.ByteBufferSender;
+import org.apache.qpid.util.ByteBufferUtils;
public class ContentHeaderBody implements AMQBody
{
@@ -36,12 +38,12 @@ public class ContentHeaderBody implements AMQBody
/** must never be null */
private final BasicContentHeaderProperties _properties;
- public ContentHeaderBody(QpidByteBuffer buffer, long size) throws AMQFrameDecodingException
+ public ContentHeaderBody(ByteBuffer buffer, long size) throws AMQFrameDecodingException
{
- buffer.getUnsignedShort();
- buffer.getUnsignedShort();
+ ByteBufferUtils.getUnsignedShort(buffer);
+ ByteBufferUtils.getUnsignedShort(buffer);
_bodySize = buffer.getLong();
- int propertyFlags = buffer.getUnsignedShort();
+ int propertyFlags = ByteBufferUtils.getUnsignedShort(buffer);
ContentHeaderPropertiesFactory factory = ContentHeaderPropertiesFactory.getInstance();
_properties = factory.createContentHeaderProperties(CLASS_ID, propertyFlags, buffer, (int)size - 14);
@@ -72,7 +74,7 @@ public class ContentHeaderBody implements AMQBody
* @throws AMQFrameDecodingException if there is a decoding issue
* @throws AMQProtocolVersionException if there is a version issue
*/
- public static ContentHeaderBody createFromBuffer(QpidByteBuffer buffer, long size)
+ public static ContentHeaderBody createFromBuffer(ByteBuffer buffer, long size)
throws AMQFrameDecodingException, AMQProtocolVersionException
{
ContentHeaderBody body = new ContentHeaderBody(buffer, size);
@@ -88,23 +90,22 @@ public class ContentHeaderBody implements AMQBody
@Override
public long writePayload(final ByteBufferSender sender)
{
- QpidByteBuffer data = QpidByteBuffer.allocate(sender.isDirectBufferPreferred(), HEADER_SIZE);
- data.putUnsignedShort(CLASS_ID);
- data.putUnsignedShort(0);
+ ByteBuffer data = ByteBuffer.allocate(HEADER_SIZE);
+ ByteBufferUtils.putUnsignedShort(data, CLASS_ID);
+ ByteBufferUtils.putUnsignedShort(data, 0);
data.putLong(_bodySize);
- data.putUnsignedShort(_properties.getPropertyFlags());
+ ByteBufferUtils.putUnsignedShort(data, _properties.getPropertyFlags());
data.flip();
sender.send(data);
- data.dispose();
return HEADER_SIZE + _properties.writePropertyListPayload(sender);
}
- public long writePayload(final QpidByteBuffer buf)
+ public long writePayload(final ByteBuffer buf)
{
- buf.putUnsignedShort(CLASS_ID);
- buf.putUnsignedShort(0);
+ ByteBufferUtils.putUnsignedShort(buf, CLASS_ID);
+ ByteBufferUtils.putUnsignedShort(buf, 0);
buf.putLong(_bodySize);
- buf.putUnsignedShort(_properties.getPropertyFlags());
+ ByteBufferUtils.putUnsignedShort(buf, _properties.getPropertyFlags());
return HEADER_SIZE + _properties.writePropertyListPayload(buf);
}
@@ -160,15 +161,15 @@ public class ContentHeaderBody implements AMQBody
_bodySize = bodySize;
}
- public static void process(final QpidByteBuffer buffer,
+ public static void process(final ByteBuffer buffer,
final ChannelMethodProcessor methodProcessor, final long size)
throws AMQFrameDecodingException
{
- int classId = buffer.getUnsignedShort();
- buffer.getUnsignedShort();
+ int classId = ByteBufferUtils.getUnsignedShort(buffer);
+ ByteBufferUtils.getUnsignedShort(buffer);
long bodySize = buffer.getLong();
- int propertyFlags = buffer.getUnsignedShort();
+ int propertyFlags = ByteBufferUtils.getUnsignedShort(buffer);
BasicContentHeaderProperties properties;
http://git-wip-us.apache.org/repos/asf/qpid-jms-amqp-0-x/blob/95203e5d/client/src/main/java/org/apache/qpid/framing/ContentHeaderPropertiesFactory.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/qpid/framing/ContentHeaderPropertiesFactory.java b/client/src/main/java/org/apache/qpid/framing/ContentHeaderPropertiesFactory.java
index 29d8849..dd35127 100644
--- a/client/src/main/java/org/apache/qpid/framing/ContentHeaderPropertiesFactory.java
+++ b/client/src/main/java/org/apache/qpid/framing/ContentHeaderPropertiesFactory.java
@@ -20,7 +20,7 @@
*/
package org.apache.qpid.framing;
-import org.apache.qpid.bytebuffer.QpidByteBuffer;
+import java.nio.ByteBuffer;
public class ContentHeaderPropertiesFactory
{
@@ -36,7 +36,7 @@ public class ContentHeaderPropertiesFactory
}
public BasicContentHeaderProperties createContentHeaderProperties(int classId, int propertyFlags,
- QpidByteBuffer buffer, int size)
+ ByteBuffer buffer, int size)
throws AMQFrameDecodingException
{
BasicContentHeaderProperties properties;
http://git-wip-us.apache.org/repos/asf/qpid-jms-amqp-0-x/blob/95203e5d/client/src/main/java/org/apache/qpid/framing/EncodingUtils.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/qpid/framing/EncodingUtils.java b/client/src/main/java/org/apache/qpid/framing/EncodingUtils.java
index 8f156da..0272f9e 100644
--- a/client/src/main/java/org/apache/qpid/framing/EncodingUtils.java
+++ b/client/src/main/java/org/apache/qpid/framing/EncodingUtils.java
@@ -20,12 +20,13 @@
*/
package org.apache.qpid.framing;
+import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.apache.qpid.bytebuffer.QpidByteBuffer;
+import org.apache.qpid.util.ByteBufferUtils;
public class EncodingUtils
{
@@ -119,7 +120,7 @@ public class EncodingUtils
}
}
- public static void writeLongAsShortString(QpidByteBuffer buffer, long l)
+ public static void writeLongAsShortString(ByteBuffer buffer, long l)
{
String s = Long.toString(l);
byte[] encodedString = new byte[1+s.length()];
@@ -134,7 +135,7 @@ public class EncodingUtils
}
- public static void writeShortStringBytes(QpidByteBuffer buffer, AMQShortString s)
+ public static void writeShortStringBytes(ByteBuffer buffer, AMQShortString s)
{
if (s != null)
{
@@ -147,18 +148,18 @@ public class EncodingUtils
}
}
- public static void writeLongStringBytes(QpidByteBuffer buffer, String s)
+ public static void writeLongStringBytes(ByteBuffer buffer, String s)
{
if (s != null)
{
int len = getUTF8Length(s);
- buffer.putUnsignedInt((long) len);
+ ByteBufferUtils.putUnsignedInt(buffer, (long) len);
buffer.put(asUTF8Bytes(s));
}
else
{
- buffer.putUnsignedInt((long) 0);
+ ByteBufferUtils.putUnsignedInt(buffer, (long) 0);
}
}
@@ -167,7 +168,7 @@ public class EncodingUtils
return 4;
}
- public static void writeFieldTableBytes(QpidByteBuffer buffer, FieldTable table)
+ public static void writeFieldTableBytes(ByteBuffer buffer, FieldTable table)
{
if (table != null)
{
@@ -175,26 +176,26 @@ public class EncodingUtils
}
else
{
- buffer.putUnsignedInt((long) 0);
+ ByteBufferUtils.putUnsignedInt(buffer, (long) 0);
}
}
- public static void writeLongstr(QpidByteBuffer buffer, byte[] data)
+ public static void writeLongstr(ByteBuffer buffer, byte[] data)
{
if (data != null)
{
- buffer.putUnsignedInt((long) data.length);
+ ByteBufferUtils.putUnsignedInt(buffer, (long) data.length);
buffer.put(data);
}
else
{
- buffer.putUnsignedInt((long) 0);
+ ByteBufferUtils.putUnsignedInt(buffer, (long) 0);
}
}
- public static FieldTable readFieldTable(QpidByteBuffer input) throws AMQFrameDecodingException
+ public static FieldTable readFieldTable(ByteBuffer input) throws AMQFrameDecodingException
{
- long length = input.getUnsignedInt();
+ long length = ByteBufferUtils.getUnsignedInt(input);
if (length == 0)
{
return null;
@@ -206,7 +207,7 @@ public class EncodingUtils
}
- public static String readLongString(QpidByteBuffer buffer)
+ public static String readLongString(ByteBuffer buffer)
{
long length = ((long)(buffer.getInt())) & 0xFFFFFFFFL;
if (length == 0)
@@ -222,7 +223,7 @@ public class EncodingUtils
}
}
- public static byte[] readLongstr(QpidByteBuffer buffer)
+ public static byte[] readLongstr(ByteBuffer buffer)
{
long length = ((long)(buffer.getInt())) & 0xFFFFFFFFL;
if (length == 0)
@@ -242,7 +243,7 @@ public class EncodingUtils
// AMQP_BOOLEAN_PROPERTY_PREFIX
- public static void writeBoolean(QpidByteBuffer buffer, boolean aBoolean)
+ public static void writeBoolean(ByteBuffer buffer, boolean aBoolean)
{
buffer.put(aBoolean ? (byte)1 : (byte)0);
}
@@ -282,9 +283,9 @@ public class EncodingUtils
return 8;
}
- public static byte[] readBytes(QpidByteBuffer buffer)
+ public static byte[] readBytes(ByteBuffer buffer)
{
- long length = buffer.getUnsignedInt();
+ long length = ByteBufferUtils.getUnsignedInt(buffer);
if (length == 0)
{
return null;
@@ -298,17 +299,17 @@ public class EncodingUtils
}
}
- public static void writeBytes(QpidByteBuffer buffer, byte[] data)
+ public static void writeBytes(ByteBuffer buffer, byte[] data)
{
if (data != null)
{
// TODO: check length fits in an unsigned byte
- buffer.putUnsignedInt((long)data.length);
+ ByteBufferUtils.putUnsignedInt(buffer, data.length);
buffer.put(data);
}
else
{
- buffer.putUnsignedInt(0L);
+ ByteBufferUtils.putUnsignedInt(buffer, 0L);
}
}
@@ -318,9 +319,9 @@ public class EncodingUtils
return encodedByteLength();
}
- public static long readLongAsShortString(QpidByteBuffer buffer) throws AMQFrameDecodingException
+ public static long readLongAsShortString(ByteBuffer buffer) throws AMQFrameDecodingException
{
- short length = buffer.getUnsignedByte();
+ short length = ByteBufferUtils.getUnsignedByte(buffer);
short pos = 0;
if (length == 0)
{
http://git-wip-us.apache.org/repos/asf/qpid-jms-amqp-0-x/blob/95203e5d/client/src/main/java/org/apache/qpid/framing/ExchangeBoundBody.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/qpid/framing/ExchangeBoundBody.java b/client/src/main/java/org/apache/qpid/framing/ExchangeBoundBody.java
index 8d5dcfd..880f2ed 100644
--- a/client/src/main/java/org/apache/qpid/framing/ExchangeBoundBody.java
+++ b/client/src/main/java/org/apache/qpid/framing/ExchangeBoundBody.java
@@ -27,8 +27,9 @@
package org.apache.qpid.framing;
+import java.nio.ByteBuffer;
+
import org.apache.qpid.QpidException;
-import org.apache.qpid.bytebuffer.QpidByteBuffer;
public class ExchangeBoundBody extends AMQMethodBodyImpl implements EncodableAMQDataBlock, AMQMethodBody
{
@@ -84,7 +85,7 @@ public class ExchangeBoundBody extends AMQMethodBodyImpl implements EncodableAMQ
return size;
}
- public void writeMethodPayload(QpidByteBuffer buffer)
+ public void writeMethodPayload(ByteBuffer buffer)
{
writeAMQShortString( buffer, _exchange );
writeAMQShortString( buffer, _routingKey );
@@ -111,7 +112,7 @@ public class ExchangeBoundBody extends AMQMethodBodyImpl implements EncodableAMQ
return buf.toString();
}
- public static void process(final QpidByteBuffer buffer,
+ public static void process(final ByteBuffer buffer,
final ServerChannelMethodProcessor dispatcher)
{
http://git-wip-us.apache.org/repos/asf/qpid-jms-amqp-0-x/blob/95203e5d/client/src/main/java/org/apache/qpid/framing/ExchangeBoundOkBody.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/qpid/framing/ExchangeBoundOkBody.java b/client/src/main/java/org/apache/qpid/framing/ExchangeBoundOkBody.java
index 6b3379f..296e7d8 100644
--- a/client/src/main/java/org/apache/qpid/framing/ExchangeBoundOkBody.java
+++ b/client/src/main/java/org/apache/qpid/framing/ExchangeBoundOkBody.java
@@ -27,8 +27,10 @@
package org.apache.qpid.framing;
+import java.nio.ByteBuffer;
+
import org.apache.qpid.QpidException;
-import org.apache.qpid.bytebuffer.QpidByteBuffer;
+import org.apache.qpid.util.ByteBufferUtils;
public class ExchangeBoundOkBody extends AMQMethodBodyImpl implements EncodableAMQDataBlock, AMQMethodBody
{
@@ -82,7 +84,7 @@ public class ExchangeBoundOkBody extends AMQMethodBodyImpl implements EncodableA
return size;
}
- public void writeMethodPayload(QpidByteBuffer buffer)
+ public void writeMethodPayload(ByteBuffer buffer)
{
writeUnsignedShort( buffer, _replyCode );
writeAMQShortString( buffer, _replyText );
@@ -105,11 +107,10 @@ public class ExchangeBoundOkBody extends AMQMethodBodyImpl implements EncodableA
return buf.toString();
}
- public static void process(final QpidByteBuffer buffer,
+ public static void process(final ByteBuffer buffer,
final ClientChannelMethodProcessor dispatcher)
{
-
- int replyCode = buffer.getUnsignedShort();
+ int replyCode = ByteBufferUtils.getUnsignedShort(buffer);
AMQShortString replyText = AMQShortString.readAMQShortString(buffer);
if(!dispatcher.ignoreAllButCloseOk())
{
http://git-wip-us.apache.org/repos/asf/qpid-jms-amqp-0-x/blob/95203e5d/client/src/main/java/org/apache/qpid/framing/ExchangeDeclareBody.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/qpid/framing/ExchangeDeclareBody.java b/client/src/main/java/org/apache/qpid/framing/ExchangeDeclareBody.java
index 5bded0a..48f583a 100644
--- a/client/src/main/java/org/apache/qpid/framing/ExchangeDeclareBody.java
+++ b/client/src/main/java/org/apache/qpid/framing/ExchangeDeclareBody.java
@@ -27,8 +27,10 @@
package org.apache.qpid.framing;
+import java.nio.ByteBuffer;
+
import org.apache.qpid.QpidException;
-import org.apache.qpid.bytebuffer.QpidByteBuffer;
+import org.apache.qpid.util.ByteBufferUtils;
public class ExchangeDeclareBody extends AMQMethodBodyImpl implements EncodableAMQDataBlock, AMQMethodBody
{
@@ -144,7 +146,7 @@ public class ExchangeDeclareBody extends AMQMethodBodyImpl implements EncodableA
return size;
}
- public void writeMethodPayload(QpidByteBuffer buffer)
+ public void writeMethodPayload(ByteBuffer buffer)
{
writeUnsignedShort( buffer, _ticket );
writeAMQShortString( buffer, _exchange );
@@ -191,11 +193,11 @@ public class ExchangeDeclareBody extends AMQMethodBodyImpl implements EncodableA
return buf.toString();
}
- public static void process(final QpidByteBuffer buffer,
+ public static void process(final ByteBuffer buffer,
final ServerChannelMethodProcessor dispatcher) throws AMQFrameDecodingException
{
- int ticket = buffer.getUnsignedShort();
+ int ticket = ByteBufferUtils.getUnsignedShort(buffer);
AMQShortString exchange = AMQShortString.readAMQShortString(buffer);
AMQShortString type = AMQShortString.readAMQShortString(buffer);
byte bitfield = buffer.get();
http://git-wip-us.apache.org/repos/asf/qpid-jms-amqp-0-x/blob/95203e5d/client/src/main/java/org/apache/qpid/framing/ExchangeDeclareOkBody.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/qpid/framing/ExchangeDeclareOkBody.java b/client/src/main/java/org/apache/qpid/framing/ExchangeDeclareOkBody.java
index 288c559..9f377d0 100644
--- a/client/src/main/java/org/apache/qpid/framing/ExchangeDeclareOkBody.java
+++ b/client/src/main/java/org/apache/qpid/framing/ExchangeDeclareOkBody.java
@@ -27,8 +27,9 @@
package org.apache.qpid.framing;
+import java.nio.ByteBuffer;
+
import org.apache.qpid.QpidException;
-import org.apache.qpid.bytebuffer.QpidByteBuffer;
public class ExchangeDeclareOkBody extends AMQMethodBodyImpl implements EncodableAMQDataBlock, AMQMethodBody
{
@@ -60,7 +61,7 @@ public class ExchangeDeclareOkBody extends AMQMethodBodyImpl implements Encodabl
return size;
}
- public void writeMethodPayload(QpidByteBuffer buffer)
+ public void writeMethodPayload(ByteBuffer buffer)
{
}
http://git-wip-us.apache.org/repos/asf/qpid-jms-amqp-0-x/blob/95203e5d/client/src/main/java/org/apache/qpid/framing/ExchangeDeleteBody.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/qpid/framing/ExchangeDeleteBody.java b/client/src/main/java/org/apache/qpid/framing/ExchangeDeleteBody.java
index 4f69bcf..05dd6c5 100644
--- a/client/src/main/java/org/apache/qpid/framing/ExchangeDeleteBody.java
+++ b/client/src/main/java/org/apache/qpid/framing/ExchangeDeleteBody.java
@@ -27,8 +27,10 @@
package org.apache.qpid.framing;
+import java.nio.ByteBuffer;
+
import org.apache.qpid.QpidException;
-import org.apache.qpid.bytebuffer.QpidByteBuffer;
+import org.apache.qpid.util.ByteBufferUtils;
public class ExchangeDeleteBody extends AMQMethodBodyImpl implements EncodableAMQDataBlock, AMQMethodBody
{
@@ -97,7 +99,7 @@ public class ExchangeDeleteBody extends AMQMethodBodyImpl implements EncodableAM
return size;
}
- public void writeMethodPayload(QpidByteBuffer buffer)
+ public void writeMethodPayload(ByteBuffer buffer)
{
writeUnsignedShort( buffer, _ticket );
writeAMQShortString( buffer, _exchange );
@@ -127,11 +129,11 @@ public class ExchangeDeleteBody extends AMQMethodBodyImpl implements EncodableAM
return buf.toString();
}
- public static void process(final QpidByteBuffer buffer,
+ public static void process(final ByteBuffer buffer,
final ServerChannelMethodProcessor dispatcher)
{
- int ticket = buffer.getUnsignedShort();
+ int ticket = ByteBufferUtils.getUnsignedShort(buffer);
AMQShortString exchange = AMQShortString.readAMQShortString(buffer);
byte bitfield = buffer.get();
boolean ifUnused = (bitfield & 0x01) == 0x01;
http://git-wip-us.apache.org/repos/asf/qpid-jms-amqp-0-x/blob/95203e5d/client/src/main/java/org/apache/qpid/framing/ExchangeDeleteOkBody.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/qpid/framing/ExchangeDeleteOkBody.java b/client/src/main/java/org/apache/qpid/framing/ExchangeDeleteOkBody.java
index 7fca59e..d6bad7a 100644
--- a/client/src/main/java/org/apache/qpid/framing/ExchangeDeleteOkBody.java
+++ b/client/src/main/java/org/apache/qpid/framing/ExchangeDeleteOkBody.java
@@ -27,8 +27,9 @@
package org.apache.qpid.framing;
+import java.nio.ByteBuffer;
+
import org.apache.qpid.QpidException;
-import org.apache.qpid.bytebuffer.QpidByteBuffer;
public class ExchangeDeleteOkBody extends AMQMethodBodyImpl implements EncodableAMQDataBlock, AMQMethodBody
{
@@ -60,7 +61,7 @@ public class ExchangeDeleteOkBody extends AMQMethodBodyImpl implements Encodable
return size;
}
- public void writeMethodPayload(QpidByteBuffer buffer)
+ public void writeMethodPayload(ByteBuffer buffer)
{
}
http://git-wip-us.apache.org/repos/asf/qpid-jms-amqp-0-x/blob/95203e5d/client/src/main/java/org/apache/qpid/framing/FieldArray.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/qpid/framing/FieldArray.java b/client/src/main/java/org/apache/qpid/framing/FieldArray.java
index 7b7c2ec..33a32f9 100644
--- a/client/src/main/java/org/apache/qpid/framing/FieldArray.java
+++ b/client/src/main/java/org/apache/qpid/framing/FieldArray.java
@@ -21,6 +21,7 @@
package org.apache.qpid.framing;
import java.math.BigDecimal;
+import java.nio.ByteBuffer;
import java.util.AbstractCollection;
import java.util.ArrayList;
import java.util.Arrays;
@@ -31,7 +32,7 @@ import java.util.Iterator;
import java.util.Map;
import java.util.Set;
-import org.apache.qpid.bytebuffer.QpidByteBuffer;
+import org.apache.qpid.util.ByteBufferUtils;
public class FieldArray<T> extends AbstractCollection<T>
{
@@ -104,7 +105,7 @@ public class FieldArray<T> extends AbstractCollection<T>
}
}
- public void writeToBuffer(final QpidByteBuffer buffer)
+ public void writeToBuffer(final ByteBuffer buffer)
{
buffer.putInt(getEncodingSize());
for( T obj : this)
@@ -113,17 +114,16 @@ public class FieldArray<T> extends AbstractCollection<T>
}
}
- public static FieldArray<?> readFromBuffer(final QpidByteBuffer buffer)
+ public static FieldArray<?> readFromBuffer(final ByteBuffer buffer)
{
ArrayList<Object> result = new ArrayList<>();
int size = buffer.getInt();
- QpidByteBuffer slicedBuffer = buffer.view(0,size);
+ ByteBuffer slicedBuffer = ByteBufferUtils.view(buffer, 0, size);
buffer.position(buffer.position()+size);
while(slicedBuffer.hasRemaining())
{
result.add(AMQTypedValue.readFromBuffer(slicedBuffer).getValue());
}
- slicedBuffer.dispose();
return new FieldArray<>(result);
}
}
http://git-wip-us.apache.org/repos/asf/qpid-jms-amqp-0-x/blob/95203e5d/client/src/main/java/org/apache/qpid/framing/FieldTable.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/qpid/framing/FieldTable.java b/client/src/main/java/org/apache/qpid/framing/FieldTable.java
index ad3c6c6..d6d90e3 100644
--- a/client/src/main/java/org/apache/qpid/framing/FieldTable.java
+++ b/client/src/main/java/org/apache/qpid/framing/FieldTable.java
@@ -21,6 +21,7 @@
package org.apache.qpid.framing;
import java.math.BigDecimal;
+import java.nio.ByteBuffer;
import java.util.Collection;
import java.util.Collections;
import java.util.Date;
@@ -36,7 +37,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.qpid.AMQPInvalidClassException;
-import org.apache.qpid.bytebuffer.QpidByteBuffer;
+import org.apache.qpid.util.ByteBufferUtils;
public class FieldTable
{
@@ -44,7 +45,7 @@ public class FieldTable
private static final String STRICT_AMQP_NAME = "STRICT_AMQP";
private static final boolean STRICT_AMQP = Boolean.valueOf(System.getProperty(STRICT_AMQP_NAME, "false"));
- private QpidByteBuffer _encodedForm;
+ private ByteBuffer _encodedForm;
private Map<AMQShortString, AMQTypedValue> _properties = null;
private long _encodedSize;
private static final int INITIAL_HASHMAP_CAPACITY = 16;
@@ -73,15 +74,15 @@ public class FieldTable
}
}
- public FieldTable(QpidByteBuffer input, int len)
+ public FieldTable(ByteBuffer input, int len)
{
this();
- _encodedForm = input.view(0,len);
+ _encodedForm = ByteBufferUtils.view(input, 0, len);
input.position(input.position()+len);
_encodedSize = len;
}
- public FieldTable(QpidByteBuffer buffer)
+ public FieldTable(ByteBuffer buffer)
{
this();
_encodedForm = buffer.duplicate();
@@ -813,7 +814,7 @@ public class FieldTable
// ************************* Byte Buffer Processing
- public void writeToBuffer(QpidByteBuffer buffer)
+ public void writeToBuffer(ByteBuffer buffer)
{
final boolean trace = _logger.isDebugEnabled();
@@ -826,7 +827,7 @@ public class FieldTable
}
}
- buffer.putUnsignedInt(getEncodedSize());
+ ByteBufferUtils.putUnsignedInt(buffer, getEncodedSize());
putDataInBuffer(buffer);
}
@@ -837,14 +838,14 @@ public class FieldTable
if(_encodedForm == null)
{
byte[] data = new byte[(int) getEncodedSize()];
- QpidByteBuffer buf = QpidByteBuffer.wrap(data);
+ ByteBuffer buf = ByteBuffer.wrap(data);
putDataInBuffer(buf);
return data;
}
else
{
byte[] encodedCopy = new byte[_encodedForm.remaining()];
- _encodedForm.copyTo(encodedCopy);
+ ByteBufferUtils.copyTo(_encodedForm, encodedCopy);
return encodedCopy;
}
@@ -933,7 +934,6 @@ public class FieldTable
if (_encodedForm != null)
{
- _encodedForm.dispose();
_encodedForm = null;
}
}
@@ -953,7 +953,6 @@ public class FieldTable
if (_encodedForm != null)
{
- _encodedForm.dispose();
_encodedForm = null;
}
}
@@ -1084,7 +1083,6 @@ public class FieldTable
initMapIfNecessary();
if (_encodedForm != null)
{
- _encodedForm.dispose();
_encodedForm = null;
}
_properties.clear();
@@ -1100,14 +1098,11 @@ public class FieldTable
return _properties.keySet();
}
- private void putDataInBuffer(QpidByteBuffer buffer)
+ private void putDataInBuffer(ByteBuffer buffer)
{
if (_encodedForm != null)
{
- byte[] encodedCopy = new byte[_encodedForm.remaining()];
- _encodedForm.copyTo(encodedCopy);
-
- buffer.put(encodedCopy);
+ ByteBufferUtils.copyTo(_encodedForm, buffer);
}
else if (_properties != null)
{
@@ -1137,7 +1132,7 @@ public class FieldTable
private void setFromBuffer() throws AMQFrameDecodingException
{
- final QpidByteBuffer slice = _encodedForm.slice();
+ final ByteBuffer slice = _encodedForm.slice();
if (_encodedSize > 0)
{
@@ -1156,7 +1151,6 @@ public class FieldTable
while (slice.hasRemaining());
}
- slice.dispose();
}
public int hashCode()
http://git-wip-us.apache.org/repos/asf/qpid-jms-amqp-0-x/blob/95203e5d/client/src/main/java/org/apache/qpid/framing/FrameCreatingMethodProcessor.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/qpid/framing/FrameCreatingMethodProcessor.java b/client/src/main/java/org/apache/qpid/framing/FrameCreatingMethodProcessor.java
index 9875cc9..c40529c 100644
--- a/client/src/main/java/org/apache/qpid/framing/FrameCreatingMethodProcessor.java
+++ b/client/src/main/java/org/apache/qpid/framing/FrameCreatingMethodProcessor.java
@@ -20,11 +20,10 @@
*/
package org.apache.qpid.framing;
+import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
-import org.apache.qpid.bytebuffer.QpidByteBuffer;
-
public class FrameCreatingMethodProcessor implements MethodProcessor<FrameCreatingMethodProcessor.ClientAndServerChannelMethodProcessor>,
ClientMethodProcessor<FrameCreatingMethodProcessor.ClientAndServerChannelMethodProcessor>,
ServerMethodProcessor<FrameCreatingMethodProcessor.ClientAndServerChannelMethodProcessor>
@@ -605,7 +604,7 @@ public class FrameCreatingMethodProcessor implements MethodProcessor<FrameCreati
}
@Override
- public void receiveMessageContent(QpidByteBuffer data)
+ public void receiveMessageContent(ByteBuffer data)
{
_processedMethods.add(new AMQFrame(_channelId, new ContentBody(data)));
}
http://git-wip-us.apache.org/repos/asf/qpid-jms-amqp-0-x/blob/95203e5d/client/src/main/java/org/apache/qpid/framing/HeartbeatBody.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/qpid/framing/HeartbeatBody.java b/client/src/main/java/org/apache/qpid/framing/HeartbeatBody.java
index cb985f5..a7461ee 100644
--- a/client/src/main/java/org/apache/qpid/framing/HeartbeatBody.java
+++ b/client/src/main/java/org/apache/qpid/framing/HeartbeatBody.java
@@ -22,9 +22,9 @@ package org.apache.qpid.framing;
import java.io.DataInputStream;
import java.io.IOException;
+import java.nio.ByteBuffer;
import org.apache.qpid.QpidException;
-import org.apache.qpid.bytebuffer.QpidByteBuffer;
import org.apache.qpid.protocol.AMQVersionAwareProtocolSession;
import org.apache.qpid.transport.ByteBufferSender;
@@ -84,7 +84,7 @@ public class HeartbeatBody implements AMQBody
}
public static void process(final int channel,
- final QpidByteBuffer in,
+ final ByteBuffer in,
final MethodProcessor processor,
final long bodySize)
{
http://git-wip-us.apache.org/repos/asf/qpid-jms-amqp-0-x/blob/95203e5d/client/src/main/java/org/apache/qpid/framing/ProtocolInitiation.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/qpid/framing/ProtocolInitiation.java b/client/src/main/java/org/apache/qpid/framing/ProtocolInitiation.java
index 757ae51..818ef41 100644
--- a/client/src/main/java/org/apache/qpid/framing/ProtocolInitiation.java
+++ b/client/src/main/java/org/apache/qpid/framing/ProtocolInitiation.java
@@ -21,11 +21,11 @@
package org.apache.qpid.framing;
import java.io.UnsupportedEncodingException;
+import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import org.apache.qpid.QpidException;
-import org.apache.qpid.bytebuffer.QpidByteBuffer;
import org.apache.qpid.transport.ByteBufferSender;
public class ProtocolInitiation extends AMQDataBlock implements EncodableAMQDataBlock
@@ -61,7 +61,7 @@ public class ProtocolInitiation extends AMQDataBlock implements EncodableAMQData
pv.equals(ProtocolVersion.v0_91) ? 1 : pv.getMinorVersion());
}
- public ProtocolInitiation(QpidByteBuffer in)
+ public ProtocolInitiation(ByteBuffer in)
{
_protocolHeader = new byte[4];
in.get(_protocolHeader);
@@ -90,7 +90,7 @@ public class ProtocolInitiation extends AMQDataBlock implements EncodableAMQData
data[5] = _protocolInstance;
data[6] = _protocolMajor;
data[7] = _protocolMinor;
- sender.send(QpidByteBuffer.wrap(data));
+ sender.send(ByteBuffer.wrap(data));
return 8l;
}
@@ -149,7 +149,7 @@ public class ProtocolInitiation extends AMQDataBlock implements EncodableAMQData
* @param in input buffer
* @return number of extra octets of data required data to decode the PI frame fully
*/
- public int decodable(QpidByteBuffer in)
+ public int decodable(ByteBuffer in)
{
return (in.remaining() >= 8) ? 0 : 8 - in.remaining();
}
http://git-wip-us.apache.org/repos/asf/qpid-jms-amqp-0-x/blob/95203e5d/client/src/main/java/org/apache/qpid/framing/QueueBindBody.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/qpid/framing/QueueBindBody.java b/client/src/main/java/org/apache/qpid/framing/QueueBindBody.java
index e7feebd..e776d96 100644
--- a/client/src/main/java/org/apache/qpid/framing/QueueBindBody.java
+++ b/client/src/main/java/org/apache/qpid/framing/QueueBindBody.java
@@ -27,8 +27,10 @@
package org.apache.qpid.framing;
+import java.nio.ByteBuffer;
+
import org.apache.qpid.QpidException;
-import org.apache.qpid.bytebuffer.QpidByteBuffer;
+import org.apache.qpid.util.ByteBufferUtils;
public class QueueBindBody extends AMQMethodBodyImpl implements EncodableAMQDataBlock, AMQMethodBody
{
@@ -112,7 +114,7 @@ public class QueueBindBody extends AMQMethodBodyImpl implements EncodableAMQData
return size;
}
- public void writeMethodPayload(QpidByteBuffer buffer)
+ public void writeMethodPayload(ByteBuffer buffer)
{
writeUnsignedShort( buffer, _ticket );
writeAMQShortString( buffer, _queue );
@@ -151,11 +153,11 @@ public class QueueBindBody extends AMQMethodBodyImpl implements EncodableAMQData
return buf.toString();
}
- public static void process(final QpidByteBuffer buffer,
+ public static void process(final ByteBuffer buffer,
final ServerChannelMethodProcessor dispatcher) throws AMQFrameDecodingException
{
- int ticket = buffer.getUnsignedShort();
+ int ticket = ByteBufferUtils.getUnsignedShort(buffer);
AMQShortString queue = AMQShortString.readAMQShortString(buffer);
AMQShortString exchange = AMQShortString.readAMQShortString(buffer);
AMQShortString bindingKey = AMQShortString.readAMQShortString(buffer);
http://git-wip-us.apache.org/repos/asf/qpid-jms-amqp-0-x/blob/95203e5d/client/src/main/java/org/apache/qpid/framing/QueueBindOkBody.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/qpid/framing/QueueBindOkBody.java b/client/src/main/java/org/apache/qpid/framing/QueueBindOkBody.java
index 13dd32f..f3e5045 100644
--- a/client/src/main/java/org/apache/qpid/framing/QueueBindOkBody.java
+++ b/client/src/main/java/org/apache/qpid/framing/QueueBindOkBody.java
@@ -27,8 +27,9 @@
package org.apache.qpid.framing;
+import java.nio.ByteBuffer;
+
import org.apache.qpid.QpidException;
-import org.apache.qpid.bytebuffer.QpidByteBuffer;
public class QueueBindOkBody extends AMQMethodBodyImpl implements EncodableAMQDataBlock, AMQMethodBody
{
@@ -60,7 +61,7 @@ public class QueueBindOkBody extends AMQMethodBodyImpl implements EncodableAMQDa
return size;
}
- public void writeMethodPayload(QpidByteBuffer buffer)
+ public void writeMethodPayload(ByteBuffer buffer)
{
}
http://git-wip-us.apache.org/repos/asf/qpid-jms-amqp-0-x/blob/95203e5d/client/src/main/java/org/apache/qpid/framing/QueueDeclareBody.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/qpid/framing/QueueDeclareBody.java b/client/src/main/java/org/apache/qpid/framing/QueueDeclareBody.java
index dbedd5f..76d2377 100644
--- a/client/src/main/java/org/apache/qpid/framing/QueueDeclareBody.java
+++ b/client/src/main/java/org/apache/qpid/framing/QueueDeclareBody.java
@@ -27,8 +27,10 @@
package org.apache.qpid.framing;
+import java.nio.ByteBuffer;
+
import org.apache.qpid.QpidException;
-import org.apache.qpid.bytebuffer.QpidByteBuffer;
+import org.apache.qpid.util.ByteBufferUtils;
public class QueueDeclareBody extends AMQMethodBodyImpl implements EncodableAMQDataBlock, AMQMethodBody
{
@@ -136,7 +138,7 @@ public class QueueDeclareBody extends AMQMethodBodyImpl implements EncodableAMQD
return size;
}
- public void writeMethodPayload(QpidByteBuffer buffer)
+ public void writeMethodPayload(ByteBuffer buffer)
{
writeUnsignedShort( buffer, _ticket );
writeAMQShortString( buffer, _queue );
@@ -179,11 +181,11 @@ public class QueueDeclareBody extends AMQMethodBodyImpl implements EncodableAMQD
return buf.toString();
}
- public static void process(final QpidByteBuffer buffer,
+ public static void process(final ByteBuffer buffer,
final ServerChannelMethodProcessor dispatcher) throws AMQFrameDecodingException
{
- int ticket = buffer.getUnsignedShort();
+ int ticket = ByteBufferUtils.getUnsignedShort(buffer);
AMQShortString queue = AMQShortString.readAMQShortString(buffer);
byte bitfield = buffer.get();
http://git-wip-us.apache.org/repos/asf/qpid-jms-amqp-0-x/blob/95203e5d/client/src/main/java/org/apache/qpid/framing/QueueDeclareOkBody.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/qpid/framing/QueueDeclareOkBody.java b/client/src/main/java/org/apache/qpid/framing/QueueDeclareOkBody.java
index bbdd41e..b79921b 100644
--- a/client/src/main/java/org/apache/qpid/framing/QueueDeclareOkBody.java
+++ b/client/src/main/java/org/apache/qpid/framing/QueueDeclareOkBody.java
@@ -27,8 +27,10 @@
package org.apache.qpid.framing;
+import java.nio.ByteBuffer;
+
import org.apache.qpid.QpidException;
-import org.apache.qpid.bytebuffer.QpidByteBuffer;
+import org.apache.qpid.util.ByteBufferUtils;
public class QueueDeclareOkBody extends AMQMethodBodyImpl implements EncodableAMQDataBlock, AMQMethodBody
{
@@ -82,7 +84,7 @@ public class QueueDeclareOkBody extends AMQMethodBodyImpl implements EncodableAM
return size;
}
- public void writeMethodPayload(QpidByteBuffer buffer)
+ public void writeMethodPayload(ByteBuffer buffer)
{
writeAMQShortString( buffer, _queue );
writeUnsignedInteger( buffer, _messageCount );
@@ -109,12 +111,12 @@ public class QueueDeclareOkBody extends AMQMethodBodyImpl implements EncodableAM
return buf.toString();
}
- public static void process(final QpidByteBuffer buffer,
+ public static void process(final ByteBuffer buffer,
final ClientChannelMethodProcessor dispatcher)
{
AMQShortString queue = AMQShortString.readAMQShortString(buffer);
- long messageCount = buffer.getUnsignedInt();
- long consumerCount = buffer.getUnsignedInt();
+ long messageCount = ByteBufferUtils.getUnsignedInt(buffer);
+ long consumerCount = ByteBufferUtils.getUnsignedInt(buffer);
if(!dispatcher.ignoreAllButCloseOk())
{
dispatcher.receiveQueueDeclareOk(queue, messageCount, consumerCount);
http://git-wip-us.apache.org/repos/asf/qpid-jms-amqp-0-x/blob/95203e5d/client/src/main/java/org/apache/qpid/framing/QueueDeleteBody.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/qpid/framing/QueueDeleteBody.java b/client/src/main/java/org/apache/qpid/framing/QueueDeleteBody.java
index 007f309..ed085eb 100644
--- a/client/src/main/java/org/apache/qpid/framing/QueueDeleteBody.java
+++ b/client/src/main/java/org/apache/qpid/framing/QueueDeleteBody.java
@@ -27,8 +27,10 @@
package org.apache.qpid.framing;
+import java.nio.ByteBuffer;
+
import org.apache.qpid.QpidException;
-import org.apache.qpid.bytebuffer.QpidByteBuffer;
+import org.apache.qpid.util.ByteBufferUtils;
public class QueueDeleteBody extends AMQMethodBodyImpl implements EncodableAMQDataBlock, AMQMethodBody
{
@@ -107,7 +109,7 @@ public class QueueDeleteBody extends AMQMethodBodyImpl implements EncodableAMQDa
return size;
}
- public void writeMethodPayload(QpidByteBuffer buffer)
+ public void writeMethodPayload(ByteBuffer buffer)
{
writeUnsignedShort( buffer, _ticket );
writeAMQShortString( buffer, _queue );
@@ -140,11 +142,11 @@ public class QueueDeleteBody extends AMQMethodBodyImpl implements EncodableAMQDa
return buf.toString();
}
- public static void process(final QpidByteBuffer buffer,
+ public static void process(final ByteBuffer buffer,
final ServerChannelMethodProcessor dispatcher)
{
- int ticket = buffer.getUnsignedShort();
+ int ticket = ByteBufferUtils.getUnsignedShort(buffer);
AMQShortString queue = AMQShortString.readAMQShortString(buffer);
byte bitfield = buffer.get();
http://git-wip-us.apache.org/repos/asf/qpid-jms-amqp-0-x/blob/95203e5d/client/src/main/java/org/apache/qpid/framing/QueueDeleteOkBody.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/qpid/framing/QueueDeleteOkBody.java b/client/src/main/java/org/apache/qpid/framing/QueueDeleteOkBody.java
index 3fe0f40..8b9b303 100644
--- a/client/src/main/java/org/apache/qpid/framing/QueueDeleteOkBody.java
+++ b/client/src/main/java/org/apache/qpid/framing/QueueDeleteOkBody.java
@@ -27,8 +27,10 @@
package org.apache.qpid.framing;
+import java.nio.ByteBuffer;
+
import org.apache.qpid.QpidException;
-import org.apache.qpid.bytebuffer.QpidByteBuffer;
+import org.apache.qpid.util.ByteBufferUtils;
public class QueueDeleteOkBody extends AMQMethodBodyImpl implements EncodableAMQDataBlock, AMQMethodBody
{
@@ -67,7 +69,7 @@ public class QueueDeleteOkBody extends AMQMethodBodyImpl implements EncodableAMQ
return size;
}
- public void writeMethodPayload(QpidByteBuffer buffer)
+ public void writeMethodPayload(ByteBuffer buffer)
{
writeUnsignedInteger( buffer, _messageCount );
}
@@ -86,10 +88,10 @@ public class QueueDeleteOkBody extends AMQMethodBodyImpl implements EncodableAMQ
return buf.toString();
}
- public static void process(QpidByteBuffer buffer,
+ public static void process(ByteBuffer buffer,
final ClientChannelMethodProcessor dispatcher)
{
- long messageCount = buffer.getUnsignedInt();
+ long messageCount = ByteBufferUtils.getUnsignedInt(buffer);
if(!dispatcher.ignoreAllButCloseOk())
{
dispatcher.receiveQueueDeleteOk(messageCount);
http://git-wip-us.apache.org/repos/asf/qpid-jms-amqp-0-x/blob/95203e5d/client/src/main/java/org/apache/qpid/framing/QueuePurgeBody.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/qpid/framing/QueuePurgeBody.java b/client/src/main/java/org/apache/qpid/framing/QueuePurgeBody.java
index dac9822..d0b91a4 100644
--- a/client/src/main/java/org/apache/qpid/framing/QueuePurgeBody.java
+++ b/client/src/main/java/org/apache/qpid/framing/QueuePurgeBody.java
@@ -27,8 +27,10 @@
package org.apache.qpid.framing;
+import java.nio.ByteBuffer;
+
import org.apache.qpid.QpidException;
-import org.apache.qpid.bytebuffer.QpidByteBuffer;
+import org.apache.qpid.util.ByteBufferUtils;
public class QueuePurgeBody extends AMQMethodBodyImpl implements EncodableAMQDataBlock, AMQMethodBody
{
@@ -87,7 +89,7 @@ public class QueuePurgeBody extends AMQMethodBodyImpl implements EncodableAMQDat
return size;
}
- public void writeMethodPayload(QpidByteBuffer buffer)
+ public void writeMethodPayload(ByteBuffer buffer)
{
writeUnsignedShort( buffer, _ticket );
writeAMQShortString( buffer, _queue );
@@ -114,11 +116,11 @@ public class QueuePurgeBody extends AMQMethodBodyImpl implements EncodableAMQDat
return buf.toString();
}
- public static void process(final QpidByteBuffer buffer,
+ public static void process(final ByteBuffer buffer,
final ServerChannelMethodProcessor dispatcher)
{
- int ticket = buffer.getUnsignedShort();
+ int ticket = ByteBufferUtils.getUnsignedShort(buffer);
AMQShortString queue = AMQShortString.readAMQShortString(buffer);
boolean nowait = (buffer.get() & 0x01) == 0x01;
if(!dispatcher.ignoreAllButCloseOk())
http://git-wip-us.apache.org/repos/asf/qpid-jms-amqp-0-x/blob/95203e5d/client/src/main/java/org/apache/qpid/framing/QueuePurgeOkBody.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/qpid/framing/QueuePurgeOkBody.java b/client/src/main/java/org/apache/qpid/framing/QueuePurgeOkBody.java
index b26a6d8..451f030 100644
--- a/client/src/main/java/org/apache/qpid/framing/QueuePurgeOkBody.java
+++ b/client/src/main/java/org/apache/qpid/framing/QueuePurgeOkBody.java
@@ -27,8 +27,10 @@
package org.apache.qpid.framing;
+import java.nio.ByteBuffer;
+
import org.apache.qpid.QpidException;
-import org.apache.qpid.bytebuffer.QpidByteBuffer;
+import org.apache.qpid.util.ByteBufferUtils;
public class QueuePurgeOkBody extends AMQMethodBodyImpl implements EncodableAMQDataBlock, AMQMethodBody
{
@@ -67,7 +69,7 @@ public class QueuePurgeOkBody extends AMQMethodBodyImpl implements EncodableAMQD
return size;
}
- public void writeMethodPayload(QpidByteBuffer buffer)
+ public void writeMethodPayload(ByteBuffer buffer)
{
writeUnsignedInteger( buffer, _messageCount );
}
@@ -86,10 +88,10 @@ public class QueuePurgeOkBody extends AMQMethodBodyImpl implements EncodableAMQD
return buf.toString();
}
- public static void process(final QpidByteBuffer buffer,
+ public static void process(final ByteBuffer buffer,
final ClientChannelMethodProcessor dispatcher)
{
- long messageCount = buffer.getUnsignedInt();
+ long messageCount = ByteBufferUtils.getUnsignedInt(buffer);
if(!dispatcher.ignoreAllButCloseOk())
{
dispatcher.receiveQueuePurgeOk(messageCount);
http://git-wip-us.apache.org/repos/asf/qpid-jms-amqp-0-x/blob/95203e5d/client/src/main/java/org/apache/qpid/framing/QueueUnbindBody.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/qpid/framing/QueueUnbindBody.java b/client/src/main/java/org/apache/qpid/framing/QueueUnbindBody.java
index 1c2fbb7..12bafd1 100644
--- a/client/src/main/java/org/apache/qpid/framing/QueueUnbindBody.java
+++ b/client/src/main/java/org/apache/qpid/framing/QueueUnbindBody.java
@@ -27,8 +27,10 @@
package org.apache.qpid.framing;
+import java.nio.ByteBuffer;
+
import org.apache.qpid.QpidException;
-import org.apache.qpid.bytebuffer.QpidByteBuffer;
+import org.apache.qpid.util.ByteBufferUtils;
public class QueueUnbindBody extends AMQMethodBodyImpl implements EncodableAMQDataBlock, AMQMethodBody
{
@@ -99,7 +101,7 @@ public class QueueUnbindBody extends AMQMethodBodyImpl implements EncodableAMQDa
return size;
}
- public void writeMethodPayload(QpidByteBuffer buffer)
+ public void writeMethodPayload(ByteBuffer buffer)
{
writeUnsignedShort( buffer, _ticket );
writeAMQShortString( buffer, _queue );
@@ -134,11 +136,11 @@ public class QueueUnbindBody extends AMQMethodBodyImpl implements EncodableAMQDa
return buf.toString();
}
- public static void process(final QpidByteBuffer buffer,
+ public static void process(final ByteBuffer buffer,
final ServerChannelMethodProcessor dispatcher) throws AMQFrameDecodingException
{
- int ticket = buffer.getUnsignedShort();
+ int ticket = ByteBufferUtils.getUnsignedShort(buffer);
AMQShortString queue = AMQShortString.readAMQShortString(buffer);
AMQShortString exchange = AMQShortString.readAMQShortString(buffer);
AMQShortString routingKey = AMQShortString.readAMQShortString(buffer);
http://git-wip-us.apache.org/repos/asf/qpid-jms-amqp-0-x/blob/95203e5d/client/src/main/java/org/apache/qpid/framing/QueueUnbindOkBody.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/qpid/framing/QueueUnbindOkBody.java b/client/src/main/java/org/apache/qpid/framing/QueueUnbindOkBody.java
index 5e0e68e..4a9dd33 100644
--- a/client/src/main/java/org/apache/qpid/framing/QueueUnbindOkBody.java
+++ b/client/src/main/java/org/apache/qpid/framing/QueueUnbindOkBody.java
@@ -27,8 +27,9 @@
package org.apache.qpid.framing;
+import java.nio.ByteBuffer;
+
import org.apache.qpid.QpidException;
-import org.apache.qpid.bytebuffer.QpidByteBuffer;
public class QueueUnbindOkBody extends AMQMethodBodyImpl implements EncodableAMQDataBlock, AMQMethodBody
{
@@ -60,7 +61,7 @@ public class QueueUnbindOkBody extends AMQMethodBodyImpl implements EncodableAMQ
return size;
}
- public void writeMethodPayload(QpidByteBuffer buffer)
+ public void writeMethodPayload(ByteBuffer buffer)
{
}
http://git-wip-us.apache.org/repos/asf/qpid-jms-amqp-0-x/blob/95203e5d/client/src/main/java/org/apache/qpid/framing/TxCommitBody.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/qpid/framing/TxCommitBody.java b/client/src/main/java/org/apache/qpid/framing/TxCommitBody.java
index 6594a3e..8f8a864 100644
--- a/client/src/main/java/org/apache/qpid/framing/TxCommitBody.java
+++ b/client/src/main/java/org/apache/qpid/framing/TxCommitBody.java
@@ -27,8 +27,9 @@
package org.apache.qpid.framing;
+import java.nio.ByteBuffer;
+
import org.apache.qpid.QpidException;
-import org.apache.qpid.bytebuffer.QpidByteBuffer;
public class TxCommitBody extends AMQMethodBodyImpl implements EncodableAMQDataBlock, AMQMethodBody
{
@@ -61,7 +62,7 @@ public class TxCommitBody extends AMQMethodBodyImpl implements EncodableAMQDataB
return 0;
}
- public void writeMethodPayload(QpidByteBuffer buffer)
+ public void writeMethodPayload(ByteBuffer buffer)
{
}
http://git-wip-us.apache.org/repos/asf/qpid-jms-amqp-0-x/blob/95203e5d/client/src/main/java/org/apache/qpid/framing/TxCommitOkBody.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/qpid/framing/TxCommitOkBody.java b/client/src/main/java/org/apache/qpid/framing/TxCommitOkBody.java
index ebdfa99..c5f9731 100644
--- a/client/src/main/java/org/apache/qpid/framing/TxCommitOkBody.java
+++ b/client/src/main/java/org/apache/qpid/framing/TxCommitOkBody.java
@@ -27,8 +27,9 @@
package org.apache.qpid.framing;
+import java.nio.ByteBuffer;
+
import org.apache.qpid.QpidException;
-import org.apache.qpid.bytebuffer.QpidByteBuffer;
public class TxCommitOkBody extends AMQMethodBodyImpl implements EncodableAMQDataBlock, AMQMethodBody
{
@@ -59,7 +60,7 @@ public class TxCommitOkBody extends AMQMethodBodyImpl implements EncodableAMQDat
return 0;
}
- public void writeMethodPayload(QpidByteBuffer buffer)
+ public void writeMethodPayload(ByteBuffer buffer)
{
}
http://git-wip-us.apache.org/repos/asf/qpid-jms-amqp-0-x/blob/95203e5d/client/src/main/java/org/apache/qpid/framing/TxRollbackBody.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/qpid/framing/TxRollbackBody.java b/client/src/main/java/org/apache/qpid/framing/TxRollbackBody.java
index 2576a38..881191c 100644
--- a/client/src/main/java/org/apache/qpid/framing/TxRollbackBody.java
+++ b/client/src/main/java/org/apache/qpid/framing/TxRollbackBody.java
@@ -27,8 +27,9 @@
package org.apache.qpid.framing;
+import java.nio.ByteBuffer;
+
import org.apache.qpid.QpidException;
-import org.apache.qpid.bytebuffer.QpidByteBuffer;
public class TxRollbackBody extends AMQMethodBodyImpl implements EncodableAMQDataBlock, AMQMethodBody
{
@@ -59,7 +60,7 @@ public class TxRollbackBody extends AMQMethodBodyImpl implements EncodableAMQDat
return 0;
}
- public void writeMethodPayload(QpidByteBuffer buffer)
+ public void writeMethodPayload(ByteBuffer buffer)
{
}
http://git-wip-us.apache.org/repos/asf/qpid-jms-amqp-0-x/blob/95203e5d/client/src/main/java/org/apache/qpid/framing/TxRollbackOkBody.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/qpid/framing/TxRollbackOkBody.java b/client/src/main/java/org/apache/qpid/framing/TxRollbackOkBody.java
index 38189fd..816269d 100644
--- a/client/src/main/java/org/apache/qpid/framing/TxRollbackOkBody.java
+++ b/client/src/main/java/org/apache/qpid/framing/TxRollbackOkBody.java
@@ -27,8 +27,9 @@
package org.apache.qpid.framing;
+import java.nio.ByteBuffer;
+
import org.apache.qpid.QpidException;
-import org.apache.qpid.bytebuffer.QpidByteBuffer;
public class TxRollbackOkBody extends AMQMethodBodyImpl implements EncodableAMQDataBlock, AMQMethodBody
{
@@ -60,7 +61,7 @@ public class TxRollbackOkBody extends AMQMethodBodyImpl implements EncodableAMQD
return 0;
}
- public void writeMethodPayload(QpidByteBuffer buffer)
+ public void writeMethodPayload(ByteBuffer buffer)
{
}
http://git-wip-us.apache.org/repos/asf/qpid-jms-amqp-0-x/blob/95203e5d/client/src/main/java/org/apache/qpid/framing/TxSelectBody.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/qpid/framing/TxSelectBody.java b/client/src/main/java/org/apache/qpid/framing/TxSelectBody.java
index 184a692..d5752f9 100644
--- a/client/src/main/java/org/apache/qpid/framing/TxSelectBody.java
+++ b/client/src/main/java/org/apache/qpid/framing/TxSelectBody.java
@@ -27,8 +27,9 @@
package org.apache.qpid.framing;
+import java.nio.ByteBuffer;
+
import org.apache.qpid.QpidException;
-import org.apache.qpid.bytebuffer.QpidByteBuffer;
public class TxSelectBody extends AMQMethodBodyImpl implements EncodableAMQDataBlock, AMQMethodBody
{
@@ -62,7 +63,7 @@ public class TxSelectBody extends AMQMethodBodyImpl implements EncodableAMQDataB
return 0;
}
- public void writeMethodPayload(QpidByteBuffer buffer)
+ public void writeMethodPayload(ByteBuffer buffer)
{
}
http://git-wip-us.apache.org/repos/asf/qpid-jms-amqp-0-x/blob/95203e5d/client/src/main/java/org/apache/qpid/framing/TxSelectOkBody.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/qpid/framing/TxSelectOkBody.java b/client/src/main/java/org/apache/qpid/framing/TxSelectOkBody.java
index d8e8c8c..82d14c2 100644
--- a/client/src/main/java/org/apache/qpid/framing/TxSelectOkBody.java
+++ b/client/src/main/java/org/apache/qpid/framing/TxSelectOkBody.java
@@ -27,8 +27,9 @@
package org.apache.qpid.framing;
+import java.nio.ByteBuffer;
+
import org.apache.qpid.QpidException;
-import org.apache.qpid.bytebuffer.QpidByteBuffer;
public class TxSelectOkBody extends AMQMethodBodyImpl implements EncodableAMQDataBlock, AMQMethodBody
{
@@ -58,7 +59,7 @@ public class TxSelectOkBody extends AMQMethodBodyImpl implements EncodableAMQDat
return 0;
}
- public void writeMethodPayload(QpidByteBuffer buffer)
+ public void writeMethodPayload(ByteBuffer buffer)
{
}
http://git-wip-us.apache.org/repos/asf/qpid-jms-amqp-0-x/blob/95203e5d/client/src/main/java/org/apache/qpid/transport/ByteBufferSender.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/qpid/transport/ByteBufferSender.java b/client/src/main/java/org/apache/qpid/transport/ByteBufferSender.java
index aac4ebd..7dcaf61 100644
--- a/client/src/main/java/org/apache/qpid/transport/ByteBufferSender.java
+++ b/client/src/main/java/org/apache/qpid/transport/ByteBufferSender.java
@@ -20,13 +20,11 @@
*/
package org.apache.qpid.transport;
-import org.apache.qpid.bytebuffer.QpidByteBuffer;
+import java.nio.ByteBuffer;
public interface ByteBufferSender
{
- boolean isDirectBufferPreferred();
-
- void send(QpidByteBuffer msg);
+ void send(ByteBuffer msg);
void flush();
http://git-wip-us.apache.org/repos/asf/qpid-jms-amqp-0-x/blob/95203e5d/client/src/main/java/org/apache/qpid/transport/MessageTransfer.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/qpid/transport/MessageTransfer.java b/client/src/main/java/org/apache/qpid/transport/MessageTransfer.java
index cd0d255..53d9940 100644
--- a/client/src/main/java/org/apache/qpid/transport/MessageTransfer.java
+++ b/client/src/main/java/org/apache/qpid/transport/MessageTransfer.java
@@ -21,19 +21,14 @@ package org.apache.qpid.transport;
*/
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
+import java.nio.ByteBuffer;
import java.util.LinkedHashMap;
-import java.util.List;
import java.util.Map;
-import org.apache.qpid.bytebuffer.QpidByteBuffer;
import org.apache.qpid.transport.codec.Decoder;
import org.apache.qpid.transport.codec.Encoder;
-
-import org.apache.qpid.util.Strings;
import org.apache.qpid.transport.network.Frame;
+import org.apache.qpid.util.Strings;
public final class MessageTransfer extends Method {
@@ -71,22 +66,13 @@ public final class MessageTransfer extends Method {
private MessageAcceptMode acceptMode;
private MessageAcquireMode acquireMode;
private Header header;
- private Collection<QpidByteBuffer> _body;
+ private ByteBuffer _body;
public MessageTransfer() {}
public MessageTransfer(String destination, MessageAcceptMode acceptMode, MessageAcquireMode acquireMode, Header header, java.nio.ByteBuffer body, Option ... options)
{
- this(destination,
- acceptMode,
- acquireMode,
- header,
- Collections.singletonList(QpidByteBuffer.wrap(body)),
- options);
- }
-
- public MessageTransfer(String destination, MessageAcceptMode acceptMode, MessageAcquireMode acquireMode, Header header, Collection<QpidByteBuffer> body, Option ... _options) {
if(destination != null) {
setDestination(destination);
}
@@ -99,13 +85,13 @@ public final class MessageTransfer extends Method {
setHeader(header);
setBody(body);
- for (int i=0; i < _options.length; i++) {
- switch (_options[i]) {
+ for (int i=0; i < options.length; i++) {
+ switch (options[i]) {
case SYNC: this.setSync(true); break;
case BATCH: this.setBatch(true); break;
case UNRELIABLE: this.setUnreliable(true); break;
case NONE: break;
- default: throw new IllegalArgumentException("invalid option: " + _options[i]);
+ default: throw new IllegalArgumentException("invalid option: " + options[i]);
}
}
@@ -209,19 +195,13 @@ public final class MessageTransfer extends Method {
}
@Override
- public final Collection<QpidByteBuffer> getBody() {
- if (this._body == null)
- {
- return null;
- }
- else
- {
- return Collections.unmodifiableCollection(_body);
- }
+ public final ByteBuffer getBody()
+ {
+ return _body;
}
@Override
- public final void setBody(Collection<QpidByteBuffer> body)
+ public final void setBody(ByteBuffer body)
{
if (body == null)
{
@@ -230,14 +210,8 @@ public final class MessageTransfer extends Method {
}
else
{
- _body = new ArrayList<>(body.size());
- int size = 0;
- for (QpidByteBuffer buf : body)
- {
- size += buf.remaining();
- _body.add(buf.duplicate());
- }
- _bodySize = size;
+ _body = body.duplicate();
+ _bodySize = _body.remaining();
}
}
@@ -247,25 +221,16 @@ public final class MessageTransfer extends Method {
return _bodySize;
}
- public final MessageTransfer body(List<QpidByteBuffer> body)
+ public final byte[] getBodyBytes()
{
- setBody(body);
- return this;
- }
-
- public final byte[] getBodyBytes() {
- Collection<QpidByteBuffer> body = getBody();
byte[] bytes = new byte[getBodySize()];
- for(QpidByteBuffer buf : body)
- {
- buf.duplicate().get(bytes);
- }
+ _body.duplicate().get(bytes);
return bytes;
}
public final void setBody(byte[] body)
{
- setBody(Collections.singletonList(QpidByteBuffer.wrap(body)));
+ setBody(ByteBuffer.wrap(body));
}
public final String getBodyString() {
@@ -338,10 +303,7 @@ public final class MessageTransfer extends Method {
{
if (_body != null)
{
- for (QpidByteBuffer buf : _body)
- {
- buf.dispose();
- }
+ _body = null;
}
}
}
http://git-wip-us.apache.org/repos/asf/qpid-jms-amqp-0-x/blob/95203e5d/client/src/main/java/org/apache/qpid/transport/Method.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/qpid/transport/Method.java b/client/src/main/java/org/apache/qpid/transport/Method.java
index d23ce72..744556a 100644
--- a/client/src/main/java/org/apache/qpid/transport/Method.java
+++ b/client/src/main/java/org/apache/qpid/transport/Method.java
@@ -20,12 +20,11 @@
*/
package org.apache.qpid.transport;
-import org.apache.qpid.bytebuffer.QpidByteBuffer;
-import org.apache.qpid.transport.network.Frame;
-
import static org.apache.qpid.transport.util.Functions.str;
-import java.util.Collection;
+import java.nio.ByteBuffer;
+
+import org.apache.qpid.transport.network.Frame;
/**
* Method
@@ -126,12 +125,12 @@ public abstract class Method extends Struct implements ProtocolEvent
throw new UnsupportedOperationException();
}
- public Collection<QpidByteBuffer> getBody()
+ public ByteBuffer getBody()
{
return null;
}
- public void setBody(Collection<QpidByteBuffer> body)
+ public void setBody(ByteBuffer body)
{
throw new UnsupportedOperationException();
}
@@ -217,7 +216,7 @@ public abstract class Method extends Struct implements ProtocolEvent
str.append(st);
}
}
- Collection<QpidByteBuffer> body = getBody();
+ ByteBuffer body = getBody();
if (body != null)
{
str.append("\n body=");
http://git-wip-us.apache.org/repos/asf/qpid-jms-amqp-0-x/blob/95203e5d/client/src/main/java/org/apache/qpid/transport/ProtocolHeader.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/qpid/transport/ProtocolHeader.java b/client/src/main/java/org/apache/qpid/transport/ProtocolHeader.java
index f0d4887..a9481ac 100644
--- a/client/src/main/java/org/apache/qpid/transport/ProtocolHeader.java
+++ b/client/src/main/java/org/apache/qpid/transport/ProtocolHeader.java
@@ -20,7 +20,8 @@
*/
package org.apache.qpid.transport;
-import org.apache.qpid.bytebuffer.QpidByteBuffer;
+import java.nio.ByteBuffer;
+
import org.apache.qpid.transport.network.Frame;
import org.apache.qpid.transport.network.NetworkDelegate;
import org.apache.qpid.transport.network.NetworkEvent;
@@ -92,9 +93,9 @@ public final class ProtocolHeader implements NetworkEvent, ProtocolEvent
return false;
}
- public QpidByteBuffer toByteBuffer()
+ public ByteBuffer toByteBuffer()
{
- QpidByteBuffer buf = QpidByteBuffer.allocate(8);
+ ByteBuffer buf = ByteBuffer.allocate(8);
buf.put(AMQP);
buf.put(protoClass);
buf.put(instance);
http://git-wip-us.apache.org/repos/asf/qpid-jms-amqp-0-x/blob/95203e5d/client/src/main/java/org/apache/qpid/transport/network/Assembler.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/qpid/transport/network/Assembler.java b/client/src/main/java/org/apache/qpid/transport/network/Assembler.java
index 931b629..0f383b9 100644
--- a/client/src/main/java/org/apache/qpid/transport/network/Assembler.java
+++ b/client/src/main/java/org/apache/qpid/transport/network/Assembler.java
@@ -22,12 +22,10 @@ package org.apache.qpid.transport.network;
import java.nio.ByteBuffer;
import java.util.ArrayList;
-import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import org.apache.qpid.bytebuffer.QpidByteBuffer;
import org.apache.qpid.transport.DeliveryProperties;
import org.apache.qpid.transport.Header;
import org.apache.qpid.transport.MessageProperties;
@@ -243,7 +241,7 @@ public class Assembler implements NetworkEventReceiver, NetworkDelegate
break;
case BODY:
command = getIncompleteCommand(channel);
- command.setBody(Collections.singletonList(QpidByteBuffer.wrap(segment)));
+ command.setBody(segment);
setIncompleteCommand(channel, null);
emit(channel, command);
break;
http://git-wip-us.apache.org/repos/asf/qpid-jms-amqp-0-x/blob/95203e5d/client/src/main/java/org/apache/qpid/transport/network/Disassembler.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/qpid/transport/network/Disassembler.java b/client/src/main/java/org/apache/qpid/transport/network/Disassembler.java
index 47f9d7a..785e252 100644
--- a/client/src/main/java/org/apache/qpid/transport/network/Disassembler.java
+++ b/client/src/main/java/org/apache/qpid/transport/network/Disassembler.java
@@ -28,14 +28,10 @@ import static org.apache.qpid.transport.network.Frame.LAST_FRAME;
import static org.apache.qpid.transport.network.Frame.LAST_SEG;
import java.nio.ByteBuffer;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.apache.qpid.bytebuffer.QpidByteBuffer;
import org.apache.qpid.transport.ByteBufferSender;
import org.apache.qpid.transport.FrameSizeObserver;
import org.apache.qpid.transport.Header;
@@ -48,6 +44,7 @@ import org.apache.qpid.transport.ProtocolHeader;
import org.apache.qpid.transport.SegmentType;
import org.apache.qpid.transport.Struct;
import org.apache.qpid.transport.codec.BBEncoder;
+import org.apache.qpid.util.ByteBufferUtils;
/**
* Disassembler
@@ -172,47 +169,31 @@ public final class Disassembler implements ProtocolEventSender, ProtocolDelegate
{
ByteBuffer buf = enc.underlyingBuffer();
buf.flip();
- QpidByteBuffer copy = QpidByteBuffer.allocate(_sender.isDirectBufferPreferred(), buf.remaining());
- copy.putCopyOf(QpidByteBuffer.wrap(buf));
+ ByteBuffer copy = ByteBuffer.allocate(buf.remaining());
+ ByteBufferUtils.copyTo(buf, copy);
copy.flip();
- final QpidByteBuffer methodBuf = copy.view(0, methodLimit);
- fragment(flags, type, method, Collections.singletonList(methodBuf));
- methodBuf.dispose();
+ final ByteBuffer methodBuf = ByteBufferUtils.view(copy,0, methodLimit);
+ fragment(flags, type, method, methodBuf);
if (payload)
{
- Collection<QpidByteBuffer> bodies = method.getBody();
- QpidByteBuffer headerBuf = copy.view(methodLimit, headerLimit);
- fragment(bodies == null ? LAST_SEG : 0x0, SegmentType.HEADER, method, Collections.singletonList(headerBuf));
- headerBuf.dispose();
- if (bodies != null)
+ ByteBuffer body = method.getBody();
+ ByteBuffer headerBuf = ByteBufferUtils.view(copy, methodLimit, headerLimit);
+ fragment(body == null ? LAST_SEG : 0x0, SegmentType.HEADER, method, headerBuf);
+ if (body != null)
{
- Collection<QpidByteBuffer> dup = new ArrayList<>(bodies.size());
- for(QpidByteBuffer b : bodies)
- {
- dup.add(b.duplicate());
- }
- fragment(LAST_SEG, SegmentType.BODY, method, dup);
- for(QpidByteBuffer b : dup)
- {
- b.dispose();
- }
+ fragment(LAST_SEG, SegmentType.BODY, method, body.duplicate());
}
}
- copy.dispose();
}
}
- private void fragment(byte flags, SegmentType type, ProtocolEvent event, Collection<QpidByteBuffer> buffers)
+ private void fragment(byte flags, SegmentType type, ProtocolEvent event, ByteBuffer buffer)
{
byte typeb = (byte) type.getValue();
byte track = event.getEncodedTrack() == Frame.L4 ? (byte) 1 : (byte) 0;
- int remaining = 0;
- for(QpidByteBuffer b : buffers)
- {
- remaining += b.remaining();
- }
+ int remaining = buffer.remaining();
boolean first = true;
while (true)
{
@@ -230,7 +211,7 @@ public final class Disassembler implements ProtocolEventSender, ProtocolDelegate
newflags |= LAST_FRAME;
}
- frame(newflags, typeb, track, event.getChannel(), size, buffers);
+ frame(newflags, typeb, track, event.getChannel(), size, buffer);
if (remaining == 0)
{
@@ -239,9 +220,9 @@ public final class Disassembler implements ProtocolEventSender, ProtocolDelegate
}
}
- private void frame(byte flags, byte type, byte track, int channel, int size, Collection<QpidByteBuffer> buffers)
+ private void frame(byte flags, byte type, byte track, int channel, int size, ByteBuffer buffer)
{
- QpidByteBuffer data = QpidByteBuffer.allocate(_sender.isDirectBufferPreferred(), HEADER_SIZE);
+ ByteBuffer data = ByteBuffer.allocate(HEADER_SIZE);
data.put(0, flags);
data.put(1, type);
@@ -252,31 +233,12 @@ public final class Disassembler implements ProtocolEventSender, ProtocolDelegate
_sender.send(data);
- data.dispose();
if(size > 0)
{
- int residual = size;
- for(QpidByteBuffer b : buffers)
- {
- final int remaining = b.remaining();
- if(remaining > 0 )
- {
- if(remaining >= residual)
- {
- final QpidByteBuffer buffer = b.view(0, residual);
- _sender.send(buffer);
- buffer.dispose();
- b.position(b.position() + residual);
- break;
- }
- else
- {
- _sender.send(b);
- residual-=remaining;
- }
- }
- }
+ final ByteBuffer view = ByteBufferUtils.view(buffer, 0, size);
+ _sender.send(view);
+ buffer.position(buffer.position() + size);
}
}
http://git-wip-us.apache.org/repos/asf/qpid-jms-amqp-0-x/blob/95203e5d/client/src/main/java/org/apache/qpid/transport/network/io/IoSender.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/qpid/transport/network/io/IoSender.java b/client/src/main/java/org/apache/qpid/transport/network/io/IoSender.java
index 3ab0601..b17e3de 100644
--- a/client/src/main/java/org/apache/qpid/transport/network/io/IoSender.java
+++ b/client/src/main/java/org/apache/qpid/transport/network/io/IoSender.java
@@ -23,6 +23,7 @@ import static org.apache.qpid.transport.util.Functions.mod;
import java.io.IOException;
import java.io.OutputStream;
import java.net.Socket;
+import java.nio.ByteBuffer;
import java.util.concurrent.atomic.AtomicBoolean;
import javax.net.ssl.SSLSocket;
@@ -30,7 +31,6 @@ import javax.net.ssl.SSLSocket;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.apache.qpid.bytebuffer.QpidByteBuffer;
import org.apache.qpid.thread.Threading;
import org.apache.qpid.transport.ByteBufferSender;
import org.apache.qpid.transport.SenderClosedException;
@@ -117,13 +117,7 @@ public final class IoSender implements Runnable, ByteBufferSender
return result;
}
- @Override
- public boolean isDirectBufferPreferred()
- {
- return false;
- }
-
- public void send(QpidByteBuffer buf)
+ public void send(ByteBuffer buf)
{
checkNotAlreadyClosed();
http://git-wip-us.apache.org/repos/asf/qpid-jms-amqp-0-x/blob/95203e5d/client/src/main/java/org/apache/qpid/transport/network/security/sasl/SASLSender.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/qpid/transport/network/security/sasl/SASLSender.java b/client/src/main/java/org/apache/qpid/transport/network/security/sasl/SASLSender.java
index 16d5a00..b7802f7 100644
--- a/client/src/main/java/org/apache/qpid/transport/network/security/sasl/SASLSender.java
+++ b/client/src/main/java/org/apache/qpid/transport/network/security/sasl/SASLSender.java
@@ -21,6 +21,7 @@
package org.apache.qpid.transport.network.security.sasl;
+import java.nio.ByteBuffer;
import java.util.concurrent.atomic.AtomicBoolean;
import javax.security.sasl.SaslException;
@@ -28,7 +29,6 @@ import javax.security.sasl.SaslException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.apache.qpid.bytebuffer.QpidByteBuffer;
import org.apache.qpid.transport.ByteBufferSender;
import org.apache.qpid.transport.SenderException;
@@ -71,13 +71,7 @@ public class SASLSender extends SASLEncryptor implements ByteBufferSender
delegate.flush();
}
- @Override
- public boolean isDirectBufferPreferred()
- {
- return false;
- }
-
- public void send(QpidByteBuffer buf)
+ public void send(ByteBuffer buf)
{
if (closed.get())
{
@@ -98,7 +92,7 @@ public class SASLSender extends SASLEncryptor implements ByteBufferSender
byte[] out = getSaslClient().wrap(appData, 0, length);
LOGGER.debug("out.length {}", out.length);
- delegate.send(QpidByteBuffer.wrap(out));
+ delegate.send(ByteBuffer.wrap(out));
}
catch (SaslException e)
{
@@ -111,7 +105,6 @@ public class SASLSender extends SASLEncryptor implements ByteBufferSender
{
delegate.send(buf);
}
- buf.dispose();
}
http://git-wip-us.apache.org/repos/asf/qpid-jms-amqp-0-x/blob/95203e5d/client/src/main/java/org/apache/qpid/transport/network/security/ssl/SSLSender.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/qpid/transport/network/security/ssl/SSLSender.java b/client/src/main/java/org/apache/qpid/transport/network/security/ssl/SSLSender.java
index 236b68d..a8c8eb5 100644
--- a/client/src/main/java/org/apache/qpid/transport/network/security/ssl/SSLSender.java
+++ b/client/src/main/java/org/apache/qpid/transport/network/security/ssl/SSLSender.java
@@ -19,6 +19,7 @@
*/
package org.apache.qpid.transport.network.security.ssl;
+import java.nio.ByteBuffer;
import java.util.Collections;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -32,10 +33,10 @@ import javax.net.ssl.SSLException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.apache.qpid.bytebuffer.QpidByteBuffer;
import org.apache.qpid.transport.ByteBufferSender;
import org.apache.qpid.transport.SenderException;
import org.apache.qpid.transport.network.security.SSLStatus;
+import org.apache.qpid.util.ByteBufferUtils;
public class SSLSender implements ByteBufferSender
{
@@ -44,14 +45,14 @@ public class SSLSender implements ByteBufferSender
private final ByteBufferSender delegate;
private final SSLEngine engine;
private final int sslBufSize;
- private final QpidByteBuffer netData;
+ private final ByteBuffer netData;
private final long timeout;
private final SSLStatus _sslStatus;
private String _hostname;
private final AtomicBoolean closed = new AtomicBoolean(false);
- private final ConcurrentLinkedQueue<QpidByteBuffer> _pending = new ConcurrentLinkedQueue<>();
+ private final ConcurrentLinkedQueue<ByteBuffer> _pending = new ConcurrentLinkedQueue<>();
public SSLSender(SSLEngine engine, ByteBufferSender delegate, SSLStatus sslStatus)
@@ -59,7 +60,7 @@ public class SSLSender implements ByteBufferSender
this.engine = engine;
this.delegate = delegate;
sslBufSize = engine.getSession().getPacketBufferSize();
- netData = QpidByteBuffer.allocate(sslBufSize);
+ netData = ByteBuffer.allocate(sslBufSize);
timeout = Long.getLong("qpid.ssl_timeout", 60000);
_sslStatus = sslStatus;
}
@@ -111,9 +112,9 @@ public class SSLSender implements ByteBufferSender
private void tearDownSSLConnection() throws Exception
{
- SSLEngineResult result = QpidByteBuffer.encryptSSL(engine,
- Collections.singletonList(QpidByteBuffer.allocate(0)),
- netData);
+ SSLEngineResult result = ByteBufferUtils.encryptSSL(engine,
+ Collections.singletonList(ByteBuffer.allocate(0)),
+ netData);
Status status = result.getStatus();
int read = result.bytesProduced();
while (status != Status.CLOSED)
@@ -128,7 +129,7 @@ public class SSLSender implements ByteBufferSender
netData.limit(netData.position());
netData.position(netData.position() - read);
- QpidByteBuffer data = netData.slice();
+ ByteBuffer data = netData.slice();
netData.limit(limit);
netData.position(netData.position() + read);
@@ -136,9 +137,9 @@ public class SSLSender implements ByteBufferSender
delegate.send(data);
flush();
}
- result = QpidByteBuffer.encryptSSL(engine,
- Collections.singletonList(QpidByteBuffer.allocate(0)),
- netData);
+ result = ByteBufferUtils.encryptSSL(engine,
+ Collections.singletonList(ByteBuffer.allocate(0)),
+ netData);
status = result.getStatus();
read = result.bytesProduced();
}
@@ -151,13 +152,7 @@ public class SSLSender implements ByteBufferSender
}
- @Override
- public boolean isDirectBufferPreferred()
- {
- return false;
- }
-
- public void send(QpidByteBuffer appData)
+ public void send(ByteBuffer appData)
{
_pending.add(appData.duplicate());
@@ -178,16 +173,15 @@ public class SSLSender implements ByteBufferSender
int read = 0;
try
{
- SSLEngineResult result = QpidByteBuffer.encryptSSL(engine, _pending, netData);
+ SSLEngineResult result = ByteBufferUtils.encryptSSL(engine, _pending, netData);
while(!_pending.isEmpty())
{
- QpidByteBuffer buf = _pending.peek();
+ ByteBuffer buf = _pending.peek();
if (buf.hasRemaining())
{
break;
}
- buf.dispose();
_pending.poll();
}
@@ -207,7 +201,7 @@ public class SSLSender implements ByteBufferSender
netData.limit(netData.position());
netData.position(netData.position() - read);
- QpidByteBuffer data = netData.slice();
+ ByteBuffer data = netData.slice();
netData.limit(limit);
netData.position(netData.position() + read);
http://git-wip-us.apache.org/repos/asf/qpid-jms-amqp-0-x/blob/95203e5d/client/src/main/java/org/apache/qpid/transport/util/Functions.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/qpid/transport/util/Functions.java b/client/src/main/java/org/apache/qpid/transport/util/Functions.java
index 68523f7..3c60f1b 100644
--- a/client/src/main/java/org/apache/qpid/transport/util/Functions.java
+++ b/client/src/main/java/org/apache/qpid/transport/util/Functions.java
@@ -23,10 +23,6 @@ package org.apache.qpid.transport.util;
import static java.lang.Math.min;
import java.nio.ByteBuffer;
-import java.util.Collection;
-
-import org.apache.qpid.bytebuffer.QpidByteBuffer;
-import org.apache.qpid.util.ByteBufferUtils;
/**
@@ -72,32 +68,6 @@ public final class Functions
public static final String str(ByteBuffer buf, int limit,int start)
{
- return str(QpidByteBuffer.wrap(buf), limit, start);
- }
-
- public static final String str(QpidByteBuffer buf)
- {
- return str(buf, buf.remaining());
- }
-
- public static final String str(QpidByteBuffer buf, int limit)
- {
- return str(buf, limit, buf.position());
- }
-
- public static final String str(Collection<QpidByteBuffer> buf, int limit, int start)
- {
- return str(ByteBufferUtils.combine(buf),limit,start);
- }
-
-
- public static final String str(Collection<QpidByteBuffer> buf, int limit)
- {
- return str(buf, limit, 0);
- }
-
- public static final String str(QpidByteBuffer buf, int limit, int start)
- {
StringBuilder str = new StringBuilder();
str.append('"');
@@ -139,10 +109,6 @@ public final class Functions
{
return hex(bytes, limit, "");
}
- public static String hex(QpidByteBuffer bytes, int limit)
- {
- return hex(bytes, limit, "");
- }
public static String hex(byte[] bytes, int limit, CharSequence separator)
{
@@ -163,25 +129,4 @@ public final class Functions
}
return sb.toString();
}
-
- public static String hex(QpidByteBuffer bytes, int limit, CharSequence separator)
- {
- limit = Math.min(limit, bytes == null ? 0 : bytes.remaining());
- StringBuilder sb = new StringBuilder(3 + limit*2);
- for(int i = 0; i < limit; i++)
- {
- sb.append(HEX_CHARACTERS[(((int)(bytes.get(bytes.position()+i))) & 0xf0)>>4]);
- sb.append(HEX_CHARACTERS[(((int)bytes.get(bytes.position()+i)) & 0x0f)]);
- if(i != bytes.remaining() - 1)
- {
- sb.append(separator);
- }
- }
- if(bytes != null && bytes.remaining()>limit)
- {
- sb.append("...");
- }
- return sb.toString();
- }
-
}
http://git-wip-us.apache.org/repos/asf/qpid-jms-amqp-0-x/blob/95203e5d/client/src/main/java/org/apache/qpid/util/ByteBufferUtils.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/qpid/util/ByteBufferUtils.java b/client/src/main/java/org/apache/qpid/util/ByteBufferUtils.java
index ce1ae1c..0f31c93 100644
--- a/client/src/main/java/org/apache/qpid/util/ByteBufferUtils.java
+++ b/client/src/main/java/org/apache/qpid/util/ByteBufferUtils.java
@@ -23,13 +23,16 @@ package org.apache.qpid.util;
import java.nio.ByteBuffer;
import java.util.Collection;
-import org.apache.qpid.bytebuffer.QpidByteBuffer;
+import javax.net.ssl.SSLEngine;
+import javax.net.ssl.SSLEngineResult;
+import javax.net.ssl.SSLException;
public class ByteBufferUtils
{
private static final ByteBuffer EMPTY_BYTE_BUFFER = ByteBuffer.allocate(0);
+ private static final ByteBuffer[] EMPTY_BYTE_BUFFER_ARRAY = new ByteBuffer[0];
- public static ByteBuffer combine(Collection<QpidByteBuffer> bufs)
+ public static ByteBuffer combine(Collection<ByteBuffer> bufs)
{
if(bufs == null || bufs.isEmpty())
{
@@ -39,33 +42,88 @@ public class ByteBufferUtils
{
int size = 0;
boolean isDirect = false;
- for(QpidByteBuffer buf : bufs)
+ for(ByteBuffer buf : bufs)
{
size += buf.remaining();
isDirect = isDirect || buf.isDirect();
}
ByteBuffer combined = isDirect ? ByteBuffer.allocateDirect(size) : ByteBuffer.allocate(size);
- for(QpidByteBuffer buf : bufs)
+ for(ByteBuffer buf : bufs)
{
- buf.copyTo(combined);
+ copyTo(buf, combined);
}
combined.flip();
return combined;
}
}
- public static int remaining(Collection<QpidByteBuffer> bufs)
+
+ public static long getUnsignedInt(ByteBuffer buffer)
{
- int size = 0;
- if (bufs != null && !bufs.isEmpty())
- {
- for (QpidByteBuffer buf : bufs)
- {
- size += buf.remaining();
- }
+ return ((long) buffer.getInt()) & 0xffffffffL;
+ }
+
+ public static void putUnsignedInt(ByteBuffer buffer, long value)
+ {
+ buffer.putInt((int) value);
+ }
+
+ public static int getUnsignedShort(ByteBuffer buffer)
+ {
+ return ((int) buffer.getShort()) & 0xffff;
+ }
+
+ public static void putUnsignedShort(ByteBuffer buffer, int value)
+ {
+ buffer.putShort((short) value);
+ }
+
+ public static short getUnsignedByte(ByteBuffer buffer)
+ {
+ return (short) (((short) buffer.get()) & 0xff);
+ }
+ public static void putUnsignedByte(ByteBuffer buffer, short value)
+ {
+ buffer.put((byte) value);
+ }
+
+ public static ByteBuffer view(ByteBuffer buffer, int offset, int length)
+ {
+ ByteBuffer view = buffer.slice();
+ view.position(offset);
+ int newLimit = Math.min(view.position() + length, view.capacity());
+ view.limit(newLimit);
+ return view.slice();
+ }
+
+ public static void copyTo(ByteBuffer src, byte[] dst)
+ {
+ ByteBuffer copy = src.duplicate();
+ copy.get(dst);
+ }
+
+ public static void copyTo(ByteBuffer src, ByteBuffer dst)
+ {
+ ByteBuffer copy = src.duplicate();
+ dst.put(copy);
+ }
+
+ public static SSLEngineResult encryptSSL(SSLEngine engine,
+ final Collection<ByteBuffer> buffers,
+ ByteBuffer dest) throws SSLException
+ {
+ final ByteBuffer[] src;
+ // QPID-7447: prevent unnecessary allocations
+ if (buffers.isEmpty())
+ {
+ src = EMPTY_BYTE_BUFFER_ARRAY;
+ }
+ else
+ {
+ src = buffers.toArray(new ByteBuffer[buffers.size()]);
}
- return size;
+ return engine.wrap(src, dest);
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org