You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by sh...@apache.org on 2011/10/21 03:20:13 UTC
svn commit: r1187150 [34/43] - in /qpid/branches/QPID-2519: ./ bin/ cpp/
cpp/bindings/ cpp/bindings/qmf/python/ cpp/bindings/qmf/ruby/
cpp/bindings/qmf/tests/ cpp/bindings/qmf2/ cpp/bindings/qmf2/examples/cpp/
cpp/bindings/qmf2/python/ cpp/bindings/qmf...
Modified: qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/BasicContentHeaderProperties.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/BasicContentHeaderProperties.java?rev=1187150&r1=1187149&r2=1187150&view=diff
==============================================================================
--- qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/BasicContentHeaderProperties.java (original)
+++ qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/BasicContentHeaderProperties.java Fri Oct 21 01:19:00 2011
@@ -20,9 +20,7 @@
*/
package org.apache.qpid.framing;
-import java.io.DataInputStream;
-import java.io.DataOutputStream;
-import java.io.IOException;
+import org.apache.mina.common.ByteBuffer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -37,6 +35,27 @@ public class BasicContentHeaderPropertie
private static final AMQShortString ZERO_STRING = null;
+ /**
+ * We store the encoded form when we decode the content header so that if we need to write it out without modifying
+ * it we can do so without incurring the expense of reencoding it
+ */
+ private byte[] _encodedForm;
+
+ /** Flag indicating whether the entire content header has been decoded yet */
+ private boolean _decoded = true;
+
+ /**
+ * We have some optimisations for partial decoding for maximum performance. The headers are used in the broker for
+ * routing in some cases so we can decode that separately.
+ */
+ private boolean _decodedHeaders = true;
+
+ /**
+ * We have some optimisations for partial decoding for maximum performance. The content type is used by all clients
+ * to determine the message type
+ */
+ private boolean _decodedContentType = true;
+
private AMQShortString _contentType;
private AMQShortString _encoding;
@@ -67,10 +86,10 @@ public class BasicContentHeaderPropertie
private int _propertyFlags = 0;
private static final int CONTENT_TYPE_MASK = 1 << 15;
- private static final int ENCODING_MASK = 1 << 14;
+ private static final int ENCONDING_MASK = 1 << 14;
private static final int HEADERS_MASK = 1 << 13;
private static final int DELIVERY_MODE_MASK = 1 << 12;
- private static final int PRIORITY_MASK = 1 << 11;
+ private static final int PROPRITY_MASK = 1 << 11;
private static final int CORRELATION_ID_MASK = 1 << 10;
private static final int REPLY_TO_MASK = 1 << 9;
private static final int EXPIRATION_MASK = 1 << 8;
@@ -82,11 +101,34 @@ public class BasicContentHeaderPropertie
private static final int CLUSTER_ID_MASK = 1 << 2;
+ /**
+ * This is 0_10 specific. We use this property to check if some message properties have been changed.
+ */
+ private boolean _hasBeenUpdated = false;
+
+ public boolean reset()
+ {
+ boolean result = _hasBeenUpdated;
+ _hasBeenUpdated = false;
+ return result;
+ }
+
+ public void updated()
+ {
+ _hasBeenUpdated = true;
+ }
+
public BasicContentHeaderProperties()
{ }
public int getPropertyListSize()
{
+ if (_encodedForm != null)
+ {
+ return _encodedForm.length;
+ }
+ else
+ {
int size = 0;
if ((_propertyFlags & (CONTENT_TYPE_MASK)) > 0)
@@ -94,7 +136,7 @@ public class BasicContentHeaderPropertie
size += EncodingUtils.encodedShortStringLength(_contentType);
}
- if ((_propertyFlags & ENCODING_MASK) > 0)
+ if ((_propertyFlags & ENCONDING_MASK) > 0)
{
size += EncodingUtils.encodedShortStringLength(_encoding);
}
@@ -109,7 +151,7 @@ public class BasicContentHeaderPropertie
size += 1;
}
- if ((_propertyFlags & PRIORITY_MASK) > 0)
+ if ((_propertyFlags & PROPRITY_MASK) > 0)
{
size += 1;
}
@@ -167,10 +209,23 @@ public class BasicContentHeaderPropertie
}
return size;
+ }
+ }
+
+ private void clearEncodedForm()
+ {
+ if (!_decoded && (_encodedForm != null))
+ {
+ // decode();
+ }
+
+ _encodedForm = null;
}
public void setPropertyFlags(int propertyFlags)
{
+ _hasBeenUpdated = true;
+ clearEncodedForm();
_propertyFlags = propertyFlags;
}
@@ -179,87 +234,94 @@ public class BasicContentHeaderPropertie
return _propertyFlags;
}
- public void writePropertyListPayload(DataOutputStream buffer) throws IOException
+ public void writePropertyListPayload(ByteBuffer buffer)
{
- if ((_propertyFlags & (CONTENT_TYPE_MASK)) != 0)
+ if (_encodedForm != null)
{
- EncodingUtils.writeShortStringBytes(buffer, _contentType);
+ buffer.put(_encodedForm);
}
-
- if ((_propertyFlags & ENCODING_MASK) != 0)
+ else
{
- EncodingUtils.writeShortStringBytes(buffer, _encoding);
- }
+ if ((_propertyFlags & (CONTENT_TYPE_MASK)) != 0)
+ {
+ EncodingUtils.writeShortStringBytes(buffer, _contentType);
+ }
- if ((_propertyFlags & HEADERS_MASK) != 0)
- {
- EncodingUtils.writeFieldTableBytes(buffer, _headers);
- }
+ if ((_propertyFlags & ENCONDING_MASK) != 0)
+ {
+ EncodingUtils.writeShortStringBytes(buffer, _encoding);
+ }
- if ((_propertyFlags & DELIVERY_MODE_MASK) != 0)
- {
- buffer.writeByte(_deliveryMode);
- }
+ if ((_propertyFlags & HEADERS_MASK) != 0)
+ {
+ EncodingUtils.writeFieldTableBytes(buffer, _headers);
+ }
- if ((_propertyFlags & PRIORITY_MASK) != 0)
- {
- buffer.writeByte(_priority);
- }
+ if ((_propertyFlags & DELIVERY_MODE_MASK) != 0)
+ {
+ buffer.put(_deliveryMode);
+ }
- if ((_propertyFlags & CORRELATION_ID_MASK) != 0)
- {
- EncodingUtils.writeShortStringBytes(buffer, _correlationId);
- }
+ if ((_propertyFlags & PROPRITY_MASK) != 0)
+ {
+ buffer.put(_priority);
+ }
- if ((_propertyFlags & REPLY_TO_MASK) != 0)
- {
- EncodingUtils.writeShortStringBytes(buffer, _replyTo);
- }
+ if ((_propertyFlags & CORRELATION_ID_MASK) != 0)
+ {
+ EncodingUtils.writeShortStringBytes(buffer, _correlationId);
+ }
- if ((_propertyFlags & EXPIRATION_MASK) != 0)
- {
- if (_expiration == 0L)
+ if ((_propertyFlags & REPLY_TO_MASK) != 0)
{
- EncodingUtils.writeShortStringBytes(buffer, ZERO_STRING);
+ EncodingUtils.writeShortStringBytes(buffer, _replyTo);
}
- else
+
+ if ((_propertyFlags & EXPIRATION_MASK) != 0)
{
- EncodingUtils.writeShortStringBytes(buffer, String.valueOf(_expiration));
+ if (_expiration == 0L)
+ {
+ EncodingUtils.writeShortStringBytes(buffer, ZERO_STRING);
+ }
+ else
+ {
+ EncodingUtils.writeShortStringBytes(buffer, String.valueOf(_expiration));
+ }
}
- }
- if ((_propertyFlags & MESSAGE_ID_MASK) != 0)
- {
- EncodingUtils.writeShortStringBytes(buffer, _messageId);
- }
+ if ((_propertyFlags & MESSAGE_ID_MASK) != 0)
+ {
+ EncodingUtils.writeShortStringBytes(buffer, _messageId);
+ }
- if ((_propertyFlags & TIMESTAMP_MASK) != 0)
- {
- EncodingUtils.writeTimestamp(buffer, _timestamp);
- }
+ if ((_propertyFlags & TIMESTAMP_MASK) != 0)
+ {
+ EncodingUtils.writeTimestamp(buffer, _timestamp);
+ }
- if ((_propertyFlags & TYPE_MASK) != 0)
- {
- EncodingUtils.writeShortStringBytes(buffer, _type);
- }
+ if ((_propertyFlags & TYPE_MASK) != 0)
+ {
+ EncodingUtils.writeShortStringBytes(buffer, _type);
+ }
- if ((_propertyFlags & USER_ID_MASK) != 0)
- {
- EncodingUtils.writeShortStringBytes(buffer, _userId);
- }
+ if ((_propertyFlags & USER_ID_MASK) != 0)
+ {
+ EncodingUtils.writeShortStringBytes(buffer, _userId);
+ }
- if ((_propertyFlags & APPLICATION_ID_MASK) != 0)
- {
- EncodingUtils.writeShortStringBytes(buffer, _appId);
- }
+ if ((_propertyFlags & APPLICATION_ID_MASK) != 0)
+ {
+ EncodingUtils.writeShortStringBytes(buffer, _appId);
+ }
- if ((_propertyFlags & CLUSTER_ID_MASK) != 0)
- {
- EncodingUtils.writeShortStringBytes(buffer, _clusterId);
+ if ((_propertyFlags & CLUSTER_ID_MASK) != 0)
+ {
+ EncodingUtils.writeShortStringBytes(buffer, _clusterId);
+ }
}
}
- public void populatePropertiesFromBuffer(DataInputStream buffer, int propertyFlags, int size) throws AMQFrameDecodingException, IOException
+ public void populatePropertiesFromBuffer(ByteBuffer buffer, int propertyFlags, int size) throws AMQFrameDecodingException
{
_propertyFlags = propertyFlags;
@@ -269,18 +331,25 @@ public class BasicContentHeaderPropertie
}
decode(buffer);
+ /*_encodedForm = new byte[size];
+ buffer.get(_encodedForm, 0, size);
+ _decoded = false;
+ _decodedHeaders = false;
+ _decodedContentType = false;*/
}
- private void decode(DataInputStream buffer) throws IOException, AMQFrameDecodingException
+ private void decode(ByteBuffer buffer)
{
// ByteBuffer buffer = ByteBuffer.wrap(_encodedForm);
-
+ int pos = buffer.position();
+ try
+ {
if ((_propertyFlags & (CONTENT_TYPE_MASK)) != 0)
{
_contentType = EncodingUtils.readAMQShortString(buffer);
}
- if ((_propertyFlags & ENCODING_MASK) != 0)
+ if ((_propertyFlags & ENCONDING_MASK) != 0)
{
_encoding = EncodingUtils.readAMQShortString(buffer);
}
@@ -292,12 +361,12 @@ public class BasicContentHeaderPropertie
if ((_propertyFlags & DELIVERY_MODE_MASK) != 0)
{
- _deliveryMode = buffer.readByte();
+ _deliveryMode = buffer.get();
}
- if ((_propertyFlags & PRIORITY_MASK) != 0)
+ if ((_propertyFlags & PROPRITY_MASK) != 0)
{
- _priority = buffer.readByte();
+ _priority = buffer.get();
}
if ((_propertyFlags & CORRELATION_ID_MASK) != 0)
@@ -344,29 +413,116 @@ public class BasicContentHeaderPropertie
{
_clusterId = EncodingUtils.readAMQShortString(buffer);
}
+ }
+ catch (AMQFrameDecodingException e)
+ {
+ throw new RuntimeException("Error in content header data: " + e, e);
+ }
+
+ final int endPos = buffer.position();
+ buffer.position(pos);
+ final int len = endPos - pos;
+ _encodedForm = new byte[len];
+ final int limit = buffer.limit();
+ buffer.limit(endPos);
+ buffer.get(_encodedForm, 0, len);
+ buffer.limit(limit);
+ buffer.position(endPos);
+ _decoded = true;
+ }
+
+ private void decodeUpToHeaders()
+ {
+ ByteBuffer buffer = ByteBuffer.wrap(_encodedForm);
+ try
+ {
+ if ((_propertyFlags & (CONTENT_TYPE_MASK)) != 0)
+ {
+ byte length = buffer.get();
+ buffer.skip(length);
+ }
+ if ((_propertyFlags & ENCONDING_MASK) != 0)
+ {
+ byte length = buffer.get();
+ buffer.skip(length);
+ }
+ if ((_propertyFlags & HEADERS_MASK) != 0)
+ {
+ _headers = EncodingUtils.readFieldTable(buffer);
+
+ }
+
+ _decodedHeaders = true;
+ }
+ catch (AMQFrameDecodingException e)
+ {
+ throw new RuntimeException("Error in content header data: " + e, e);
+ }
}
+ private void decodeUpToContentType()
+ {
+ ByteBuffer buffer = ByteBuffer.wrap(_encodedForm);
+
+ if ((_propertyFlags & (CONTENT_TYPE_MASK)) != 0)
+ {
+ _contentType = EncodingUtils.readAMQShortString(buffer);
+ }
+
+ _decodedContentType = true;
+ }
+
+ private void decodeIfNecessary()
+ {
+ if (!_decoded)
+ {
+ // decode();
+ }
+ }
+
+ private void decodeHeadersIfNecessary()
+ {
+ if (!_decoded && !_decodedHeaders)
+ {
+ decodeUpToHeaders();
+ }
+ }
+
+ private void decodeContentTypeIfNecessary()
+ {
+ if (!_decoded && !_decodedContentType)
+ {
+ decodeUpToContentType();
+ }
+ }
public AMQShortString getContentType()
{
+ decodeContentTypeIfNecessary();
+
return _contentType;
}
public String getContentTypeAsString()
{
+ decodeContentTypeIfNecessary();
+
return (_contentType == null) ? null : _contentType.toString();
}
public void setContentType(AMQShortString contentType)
{
+ _hasBeenUpdated = true;
+ clearEncodedForm();
_propertyFlags |= (CONTENT_TYPE_MASK);
_contentType = contentType;
}
public void setContentType(String contentType)
{
+ _hasBeenUpdated = true;
setContentType((contentType == null) ? null : new AMQShortString(contentType));
}
@@ -378,23 +534,31 @@ public class BasicContentHeaderPropertie
public AMQShortString getEncoding()
{
+ decodeIfNecessary();
+
return _encoding;
}
public void setEncoding(String encoding)
{
- _propertyFlags |= ENCODING_MASK;
+ _hasBeenUpdated = true;
+ clearEncodedForm();
+ _propertyFlags |= ENCONDING_MASK;
_encoding = (encoding == null) ? null : new AMQShortString(encoding);
}
public void setEncoding(AMQShortString encoding)
{
- _propertyFlags |= ENCODING_MASK;
+ _hasBeenUpdated = true;
+ clearEncodedForm();
+ _propertyFlags |= ENCONDING_MASK;
_encoding = encoding;
}
public FieldTable getHeaders()
{
+ decodeHeadersIfNecessary();
+
if (_headers == null)
{
setHeaders(FieldTableFactory.newFieldTable());
@@ -405,146 +569,191 @@ public class BasicContentHeaderPropertie
public void setHeaders(FieldTable headers)
{
+ _hasBeenUpdated = true;
+ clearEncodedForm();
_propertyFlags |= HEADERS_MASK;
_headers = headers;
}
public byte getDeliveryMode()
{
+ decodeIfNecessary();
+
return _deliveryMode;
}
public void setDeliveryMode(byte deliveryMode)
{
+ clearEncodedForm();
_propertyFlags |= DELIVERY_MODE_MASK;
_deliveryMode = deliveryMode;
}
public byte getPriority()
{
+ decodeIfNecessary();
+
return _priority;
}
public void setPriority(byte priority)
{
- _propertyFlags |= PRIORITY_MASK;
+ clearEncodedForm();
+ _propertyFlags |= PROPRITY_MASK;
_priority = priority;
}
public AMQShortString getCorrelationId()
{
+ decodeIfNecessary();
+
return _correlationId;
}
public String getCorrelationIdAsString()
{
+ decodeIfNecessary();
+
return (_correlationId == null) ? null : _correlationId.toString();
}
public void setCorrelationId(String correlationId)
{
+ _hasBeenUpdated = true;
setCorrelationId((correlationId == null) ? null : new AMQShortString(correlationId));
}
public void setCorrelationId(AMQShortString correlationId)
{
+ _hasBeenUpdated = true;
+ clearEncodedForm();
_propertyFlags |= CORRELATION_ID_MASK;
_correlationId = correlationId;
}
public String getReplyToAsString()
{
+ decodeIfNecessary();
+
return (_replyTo == null) ? null : _replyTo.toString();
}
public AMQShortString getReplyTo()
{
+ decodeIfNecessary();
+
return _replyTo;
}
public void setReplyTo(String replyTo)
{
+ _hasBeenUpdated = true;
setReplyTo((replyTo == null) ? null : new AMQShortString(replyTo));
}
public void setReplyTo(AMQShortString replyTo)
{
+ _hasBeenUpdated = true;
+ clearEncodedForm();
_propertyFlags |= REPLY_TO_MASK;
_replyTo = replyTo;
}
public long getExpiration()
{
+ decodeIfNecessary();
return _expiration;
}
public void setExpiration(long expiration)
{
+ clearEncodedForm();
_propertyFlags |= EXPIRATION_MASK;
_expiration = expiration;
}
public AMQShortString getMessageId()
{
+ decodeIfNecessary();
+
return _messageId;
}
public String getMessageIdAsString()
{
+ decodeIfNecessary();
+
return (_messageId == null) ? null : _messageId.toString();
}
public void setMessageId(String messageId)
{
+ _hasBeenUpdated = true;
+ clearEncodedForm();
_propertyFlags |= MESSAGE_ID_MASK;
_messageId = (messageId == null) ? null : new AMQShortString(messageId);
}
public void setMessageId(AMQShortString messageId)
{
+ _hasBeenUpdated = true;
+ clearEncodedForm();
_propertyFlags |= MESSAGE_ID_MASK;
_messageId = messageId;
}
public long getTimestamp()
{
+ decodeIfNecessary();
return _timestamp;
}
public void setTimestamp(long timestamp)
{
+ clearEncodedForm();
_propertyFlags |= TIMESTAMP_MASK;
_timestamp = timestamp;
}
public String getTypeAsString()
{
+ decodeIfNecessary();
+
return (_type == null) ? null : _type.toString();
}
public AMQShortString getType()
{
+ decodeIfNecessary();
+
return _type;
}
public void setType(String type)
{
+ _hasBeenUpdated = true;
setType((type == null) ? null : new AMQShortString(type));
}
public void setType(AMQShortString type)
{
+ _hasBeenUpdated = true;
+ clearEncodedForm();
_propertyFlags |= TYPE_MASK;
_type = type;
}
public String getUserIdAsString()
{
+ decodeIfNecessary();
+
return (_userId == null) ? null : _userId.toString();
}
public AMQShortString getUserId()
{
+ decodeIfNecessary();
+
return _userId;
}
@@ -555,48 +764,65 @@ public class BasicContentHeaderPropertie
public void setUserId(AMQShortString userId)
{
+ _hasBeenUpdated = true;
+ clearEncodedForm();
_propertyFlags |= USER_ID_MASK;
_userId = userId;
}
public String getAppIdAsString()
{
+ decodeIfNecessary();
+
return (_appId == null) ? null : _appId.toString();
}
public AMQShortString getAppId()
{
+ decodeIfNecessary();
+
return _appId;
}
public void setAppId(String appId)
{
+ _hasBeenUpdated = true;
setAppId((appId == null) ? null : new AMQShortString(appId));
}
public void setAppId(AMQShortString appId)
{
+ _hasBeenUpdated = true;
+ clearEncodedForm();
_propertyFlags |= APPLICATION_ID_MASK;
_appId = appId;
+ _hasBeenUpdated = true;
}
public String getClusterIdAsString()
{
+ _hasBeenUpdated = true;
+ decodeIfNecessary();
return (_clusterId == null) ? null : _clusterId.toString();
}
public AMQShortString getClusterId()
{
+ _hasBeenUpdated = true;
+ decodeIfNecessary();
return _clusterId;
}
public void setClusterId(String clusterId)
{
+ _hasBeenUpdated = true;
setClusterId((clusterId == null) ? null : new AMQShortString(clusterId));
}
public void setClusterId(AMQShortString clusterId)
{
+ _hasBeenUpdated = true;
+ clearEncodedForm();
_propertyFlags |= CLUSTER_ID_MASK;
_clusterId = clusterId;
}
Modified: qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/BodyFactory.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/BodyFactory.java?rev=1187150&r1=1187149&r2=1187150&view=diff
==============================================================================
--- qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/BodyFactory.java (original)
+++ qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/BodyFactory.java Fri Oct 21 01:19:00 2011
@@ -20,13 +20,12 @@
*/
package org.apache.qpid.framing;
-import java.io.DataInputStream;
-import java.io.IOException;
+import org.apache.mina.common.ByteBuffer;
/**
* Any class that is capable of turning a stream of bytes into an AMQ structure must implement this interface.
*/
public interface BodyFactory
{
- AMQBody createBody(DataInputStream in, long bodySize) throws AMQFrameDecodingException, IOException;
+ AMQBody createBody(ByteBuffer in, long bodySize) throws AMQFrameDecodingException;
}
Modified: qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/CompositeAMQDataBlock.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/CompositeAMQDataBlock.java?rev=1187150&r1=1187149&r2=1187150&view=diff
==============================================================================
--- qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/CompositeAMQDataBlock.java (original)
+++ qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/CompositeAMQDataBlock.java Fri Oct 21 01:19:00 2011
@@ -20,8 +20,7 @@
*/
package org.apache.qpid.framing;
-import java.io.DataOutputStream;
-import java.io.IOException;
+import org.apache.mina.common.ByteBuffer;
public class CompositeAMQDataBlock extends AMQDataBlock implements EncodableAMQDataBlock
{
@@ -50,7 +49,7 @@ public class CompositeAMQDataBlock exten
return frameSize;
}
- public void writePayload(DataOutputStream buffer) throws IOException
+ public void writePayload(ByteBuffer buffer)
{
for (int i = 0; i < _blocks.length; i++)
{
Modified: qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/ContentBody.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/ContentBody.java?rev=1187150&r1=1187149&r2=1187150&view=diff
==============================================================================
--- qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/ContentBody.java (original)
+++ qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/ContentBody.java Fri Oct 21 01:19:00 2011
@@ -20,10 +20,7 @@
*/
package org.apache.qpid.framing;
-import java.io.DataInputStream;
-import java.io.DataOutputStream;
-import java.io.IOException;
-
+import org.apache.mina.common.ByteBuffer;
import org.apache.qpid.protocol.AMQVersionAwareProtocolSession;
import org.apache.qpid.AMQException;
@@ -31,22 +28,27 @@ public class ContentBody implements AMQB
{
public static final byte TYPE = 3;
- public byte[] _payload;
+ public ByteBuffer payload;
public ContentBody()
{
}
- public ContentBody(DataInputStream buffer, long size) throws AMQFrameDecodingException, IOException
+ public ContentBody(ByteBuffer buffer, long size) throws AMQFrameDecodingException
{
- _payload = new byte[(int)size];
- buffer.read(_payload);
+ if (size > 0)
+ {
+ payload = buffer.slice();
+ payload.limit((int) size);
+ buffer.skip((int) size);
+ }
+
}
- public ContentBody(byte[] payload)
+ public ContentBody(ByteBuffer payload)
{
- _payload = payload;
+ this.payload = payload;
}
public byte getFrameType()
@@ -56,12 +58,23 @@ public class ContentBody implements AMQB
public int getSize()
{
- return _payload == null ? 0 : _payload.length;
+ return (payload == null ? 0 : payload.limit());
}
- public void writePayload(DataOutputStream buffer) throws IOException
+ public void writePayload(ByteBuffer buffer)
{
- buffer.write(_payload);
+ if (payload != null)
+ {
+ if(payload.isDirect() || payload.isReadOnly())
+ {
+ ByteBuffer copy = payload.duplicate();
+ buffer.put(copy.rewind());
+ }
+ else
+ {
+ buffer.put(payload.array(),payload.arrayOffset(),payload.limit());
+ }
+ }
}
public void handle(final int channelId, final AMQVersionAwareProtocolSession session)
@@ -70,18 +83,32 @@ public class ContentBody implements AMQB
session.contentBodyReceived(channelId, this);
}
- protected void populateFromBuffer(DataInputStream buffer, long size) throws AMQFrameDecodingException, IOException
+ protected void populateFromBuffer(ByteBuffer buffer, long size) throws AMQFrameDecodingException
{
if (size > 0)
{
- _payload = new byte[(int)size];
- buffer.read(_payload);
+ payload = buffer.slice();
+ payload.limit((int) size);
+ buffer.skip((int) size);
}
}
public void reduceBufferToFit()
{
+ if (payload != null && (payload.remaining() < payload.capacity() / 2))
+ {
+ int size = payload.limit();
+ ByteBuffer newPayload = ByteBuffer.allocate(size);
+
+ newPayload.put(payload);
+ newPayload.flip();
+
+ //reduce reference count on payload
+ payload.release();
+
+ payload = newPayload;
+ }
}
Modified: qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/ContentBodyFactory.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/ContentBodyFactory.java?rev=1187150&r1=1187149&r2=1187150&view=diff
==============================================================================
--- qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/ContentBodyFactory.java (original)
+++ qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/ContentBodyFactory.java Fri Oct 21 01:19:00 2011
@@ -20,8 +20,7 @@
*/
package org.apache.qpid.framing;
-import java.io.DataInputStream;
-import java.io.IOException;
+import org.apache.mina.common.ByteBuffer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -42,7 +41,7 @@ public class ContentBodyFactory implemen
_log.debug("Creating content body factory");
}
- public AMQBody createBody(DataInputStream in, long bodySize) throws AMQFrameDecodingException, IOException
+ public AMQBody createBody(ByteBuffer in, long bodySize) throws AMQFrameDecodingException
{
return new ContentBody(in, bodySize);
}
Modified: qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/ContentHeaderBody.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/ContentHeaderBody.java?rev=1187150&r1=1187149&r2=1187150&view=diff
==============================================================================
--- qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/ContentHeaderBody.java (original)
+++ qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/ContentHeaderBody.java Fri Oct 21 01:19:00 2011
@@ -20,10 +20,7 @@
*/
package org.apache.qpid.framing;
-import java.io.DataInputStream;
-import java.io.DataOutputStream;
-import java.io.IOException;
-
+import org.apache.mina.common.ByteBuffer;
import org.apache.qpid.protocol.AMQVersionAwareProtocolSession;
import org.apache.qpid.AMQException;
@@ -39,18 +36,18 @@ public class ContentHeaderBody implement
public long bodySize;
/** must never be null */
- private ContentHeaderProperties properties;
+ public ContentHeaderProperties properties;
public ContentHeaderBody()
{
}
- public ContentHeaderBody(DataInputStream buffer, long size) throws AMQFrameDecodingException, IOException
+ public ContentHeaderBody(ByteBuffer buffer, long size) throws AMQFrameDecodingException
{
- classId = buffer.readUnsignedShort();
- weight = buffer.readUnsignedShort();
- bodySize = buffer.readLong();
- int propertyFlags = buffer.readUnsignedShort();
+ classId = buffer.getUnsignedShort();
+ weight = buffer.getUnsignedShort();
+ bodySize = buffer.getLong();
+ int propertyFlags = buffer.getUnsignedShort();
ContentHeaderPropertiesFactory factory = ContentHeaderPropertiesFactory.getInstance();
properties = factory.createContentHeaderProperties(classId, propertyFlags, buffer, (int)size - 14);
@@ -75,13 +72,13 @@ public class ContentHeaderBody implement
return TYPE;
}
- protected void populateFromBuffer(DataInputStream buffer, long size)
- throws AMQFrameDecodingException, AMQProtocolVersionException, IOException
+ protected void populateFromBuffer(ByteBuffer buffer, long size)
+ throws AMQFrameDecodingException, AMQProtocolVersionException
{
- classId = buffer.readUnsignedShort();
- weight = buffer.readUnsignedShort();
- bodySize = buffer.readLong();
- int propertyFlags = buffer.readUnsignedShort();
+ classId = buffer.getUnsignedShort();
+ weight = buffer.getUnsignedShort();
+ bodySize = buffer.getLong();
+ int propertyFlags = buffer.getUnsignedShort();
ContentHeaderPropertiesFactory factory = ContentHeaderPropertiesFactory.getInstance();
properties = factory.createContentHeaderProperties(classId, propertyFlags, buffer, (int)size - 14);
}
@@ -93,8 +90,8 @@ public class ContentHeaderBody implement
* @return
* @throws AMQFrameDecodingException
*/
- public static ContentHeaderBody createFromBuffer(DataInputStream buffer, long size)
- throws AMQFrameDecodingException, AMQProtocolVersionException, IOException
+ public static ContentHeaderBody createFromBuffer(ByteBuffer buffer, long size)
+ throws AMQFrameDecodingException, AMQProtocolVersionException
{
ContentHeaderBody body = new ContentHeaderBody(buffer, size);
@@ -106,11 +103,11 @@ public class ContentHeaderBody implement
return 2 + 2 + 8 + 2 + properties.getPropertyListSize();
}
- public void writePayload(DataOutputStream buffer) throws IOException
+ public void writePayload(ByteBuffer buffer)
{
EncodingUtils.writeUnsignedShort(buffer, classId);
EncodingUtils.writeUnsignedShort(buffer, weight);
- buffer.writeLong(bodySize);
+ buffer.putLong(bodySize);
EncodingUtils.writeUnsignedShort(buffer, properties.getPropertyFlags());
properties.writePropertyListPayload(buffer);
}
@@ -131,25 +128,4 @@ public class ContentHeaderBody implement
{
return new AMQFrame(channelId, body);
}
-
- public ContentHeaderProperties getProperties()
- {
- return properties;
- }
-
- public void setProperties(ContentHeaderProperties props)
- {
- properties = props;
- }
-
- @Override
- public String toString()
- {
- return "ContentHeaderBody{" +
- "classId=" + classId +
- ", weight=" + weight +
- ", bodySize=" + bodySize +
- ", properties=" + properties +
- '}';
- }
}
Modified: qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/ContentHeaderBodyFactory.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/ContentHeaderBodyFactory.java?rev=1187150&r1=1187149&r2=1187150&view=diff
==============================================================================
--- qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/ContentHeaderBodyFactory.java (original)
+++ qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/ContentHeaderBodyFactory.java Fri Oct 21 01:19:00 2011
@@ -20,8 +20,7 @@
*/
package org.apache.qpid.framing;
-import java.io.DataInputStream;
-import java.io.IOException;
+import org.apache.mina.common.ByteBuffer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -42,7 +41,7 @@ public class ContentHeaderBodyFactory im
_log.debug("Creating content header body factory");
}
- public AMQBody createBody(DataInputStream in, long bodySize) throws AMQFrameDecodingException, IOException
+ public AMQBody createBody(ByteBuffer in, long bodySize) throws AMQFrameDecodingException
{
// all content headers are the same - it is only the properties that differ.
// the content header body further delegates construction of properties
Modified: qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/ContentHeaderProperties.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/ContentHeaderProperties.java?rev=1187150&r1=1187149&r2=1187150&view=diff
==============================================================================
--- qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/ContentHeaderProperties.java (original)
+++ qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/ContentHeaderProperties.java Fri Oct 21 01:19:00 2011
@@ -20,10 +20,7 @@
*/
package org.apache.qpid.framing;
-import java.io.DataInputStream;
-import java.io.DataOutputStream;
-import java.io.IOException;
-
+import org.apache.mina.common.ByteBuffer;
/**
* There will be an implementation of this interface for each content type. All content types have associated
@@ -35,7 +32,7 @@ public interface ContentHeaderProperties
* Writes the property list to the buffer, in a suitably encoded form.
* @param buffer The buffer to write to
*/
- void writePropertyListPayload(DataOutputStream buffer) throws IOException;
+ void writePropertyListPayload(ByteBuffer buffer);
/**
* Populates the properties from buffer.
@@ -43,8 +40,8 @@ public interface ContentHeaderProperties
* @param propertyFlags he property flags.
* @throws AMQFrameDecodingException when the buffer does not contain valid data
*/
- void populatePropertiesFromBuffer(DataInputStream buffer, int propertyFlags, int size)
- throws AMQFrameDecodingException, IOException;
+ void populatePropertiesFromBuffer(ByteBuffer buffer, int propertyFlags, int size)
+ throws AMQFrameDecodingException;
/**
* @return the size of the encoded property list in bytes.
@@ -59,4 +56,5 @@ public interface ContentHeaderProperties
*/
int getPropertyFlags();
+ void updated();
}
Modified: qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/ContentHeaderPropertiesFactory.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/ContentHeaderPropertiesFactory.java?rev=1187150&r1=1187149&r2=1187150&view=diff
==============================================================================
--- qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/ContentHeaderPropertiesFactory.java (original)
+++ qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/ContentHeaderPropertiesFactory.java Fri Oct 21 01:19:00 2011
@@ -20,8 +20,7 @@
*/
package org.apache.qpid.framing;
-import java.io.DataInputStream;
-import java.io.IOException;
+import org.apache.mina.common.ByteBuffer;
import org.apache.qpid.framing.amqp_8_0.BasicConsumeBodyImpl;
@@ -39,8 +38,8 @@ public class ContentHeaderPropertiesFact
}
public ContentHeaderProperties createContentHeaderProperties(int classId, int propertyFlags,
- DataInputStream buffer, int size)
- throws AMQFrameDecodingException, IOException
+ ByteBuffer buffer, int size)
+ throws AMQFrameDecodingException
{
ContentHeaderProperties properties;
// AMQP version change: "Hardwired" version to major=8, minor=0
Modified: qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/EncodingUtils.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/EncodingUtils.java?rev=1187150&r1=1187149&r2=1187150&view=diff
==============================================================================
--- qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/EncodingUtils.java (original)
+++ qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/EncodingUtils.java Fri Oct 21 01:19:00 2011
@@ -20,12 +20,11 @@
*/
package org.apache.qpid.framing;
-import java.io.*;
+import org.apache.mina.common.ByteBuffer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.nio.ByteBuffer;
import java.nio.charset.Charset;
public class EncodingUtils
@@ -219,7 +218,7 @@ public class EncodingUtils
return 0;
}
- public static void writeShortStringBytes(DataOutputStream buffer, String s) throws IOException
+ public static void writeShortStringBytes(ByteBuffer buffer, String s)
{
if (s != null)
{
@@ -232,18 +231,18 @@ public class EncodingUtils
// TODO: check length fits in an unsigned byte
writeUnsignedByte(buffer, (short)encodedString.length);
- buffer.write(encodedString);
+ buffer.put(encodedString);
}
else
{
// really writing out unsigned byte
- buffer.write((byte) 0);
+ buffer.put((byte) 0);
}
}
- public static void writeShortStringBytes(DataOutputStream buffer, AMQShortString s) throws IOException
+ public static void writeShortStringBytes(ByteBuffer buffer, AMQShortString s)
{
if (s != null)
{
@@ -253,11 +252,11 @@ public class EncodingUtils
else
{
// really writing out unsigned byte
- buffer.write((byte) 0);
+ buffer.put((byte) 0);
}
}
- public static void writeLongStringBytes(DataOutputStream buffer, String s) throws IOException
+ public static void writeLongStringBytes(ByteBuffer buffer, String s)
{
assert (s == null) || (s.length() <= 0xFFFE);
if (s != null)
@@ -271,7 +270,7 @@ public class EncodingUtils
encodedString[i] = (byte) cha[i];
}
- buffer.write(encodedString);
+ buffer.put(encodedString);
}
else
{
@@ -279,7 +278,7 @@ public class EncodingUtils
}
}
- public static void writeLongStringBytes(DataOutputStream buffer, char[] s) throws IOException
+ public static void writeLongStringBytes(ByteBuffer buffer, char[] s)
{
assert (s == null) || (s.length <= 0xFFFE);
if (s != null)
@@ -292,7 +291,7 @@ public class EncodingUtils
encodedString[i] = (byte) s[i];
}
- buffer.write(encodedString);
+ buffer.put(encodedString);
}
else
{
@@ -300,13 +299,13 @@ public class EncodingUtils
}
}
- public static void writeLongStringBytes(DataOutputStream buffer, byte[] bytes) throws IOException
+ public static void writeLongStringBytes(ByteBuffer buffer, byte[] bytes)
{
assert (bytes == null) || (bytes.length <= 0xFFFE);
if (bytes != null)
{
writeUnsignedInteger(buffer, bytes.length);
- buffer.write(bytes);
+ buffer.put(bytes);
}
else
{
@@ -314,24 +313,24 @@ public class EncodingUtils
}
}
- public static void writeUnsignedByte(DataOutputStream buffer, short b) throws IOException
+ public static void writeUnsignedByte(ByteBuffer buffer, short b)
{
byte bv = (byte) b;
- buffer.write(bv);
+ buffer.put(bv);
}
- public static void writeUnsignedShort(DataOutputStream buffer, int s) throws IOException
+ public static void writeUnsignedShort(ByteBuffer buffer, int s)
{
// TODO: Is this comparison safe? Do I need to cast RHS to long?
if (s < Short.MAX_VALUE)
{
- buffer.writeShort(s);
+ buffer.putShort((short) s);
}
else
{
short sv = (short) s;
- buffer.write((byte) (0xFF & (sv >> 8)));
- buffer.write((byte) (0xFF & sv));
+ buffer.put((byte) (0xFF & (sv >> 8)));
+ buffer.put((byte) (0xFF & sv));
}
}
@@ -340,12 +339,12 @@ public class EncodingUtils
return 4;
}
- public static void writeUnsignedInteger(DataOutputStream buffer, long l) throws IOException
+ public static void writeUnsignedInteger(ByteBuffer buffer, long l)
{
// TODO: Is this comparison safe? Do I need to cast RHS to long?
if (l < Integer.MAX_VALUE)
{
- buffer.writeInt((int) l);
+ buffer.putInt((int) l);
}
else
{
@@ -353,14 +352,14 @@ public class EncodingUtils
// FIXME: This *may* go faster if we build this into a local 4-byte array and then
// put the array in a single call.
- buffer.write((byte) (0xFF & (iv >> 24)));
- buffer.write((byte) (0xFF & (iv >> 16)));
- buffer.write((byte) (0xFF & (iv >> 8)));
- buffer.write((byte) (0xFF & iv));
+ buffer.put((byte) (0xFF & (iv >> 24)));
+ buffer.put((byte) (0xFF & (iv >> 16)));
+ buffer.put((byte) (0xFF & (iv >> 8)));
+ buffer.put((byte) (0xFF & iv));
}
}
- public static void writeFieldTableBytes(DataOutputStream buffer, FieldTable table) throws IOException
+ public static void writeFieldTableBytes(ByteBuffer buffer, FieldTable table)
{
if (table != null)
{
@@ -372,12 +371,12 @@ public class EncodingUtils
}
}
- public static void writeContentBytes(DataOutputStream buffer, Content content)
+ public static void writeContentBytes(ByteBuffer buffer, Content content)
{
// TODO: New Content class required for AMQP 0-9.
}
- public static void writeBooleans(DataOutputStream buffer, boolean[] values) throws IOException
+ public static void writeBooleans(ByteBuffer buffer, boolean[] values)
{
byte packedValue = 0;
for (int i = 0; i < values.length; i++)
@@ -388,16 +387,16 @@ public class EncodingUtils
}
}
- buffer.write(packedValue);
+ buffer.put(packedValue);
}
- public static void writeBooleans(DataOutputStream buffer, boolean value) throws IOException
+ public static void writeBooleans(ByteBuffer buffer, boolean value)
{
- buffer.write(value ? (byte) 1 : (byte) 0);
+ buffer.put(value ? (byte) 1 : (byte) 0);
}
- public static void writeBooleans(DataOutputStream buffer, boolean value0, boolean value1) throws IOException
+ public static void writeBooleans(ByteBuffer buffer, boolean value0, boolean value1)
{
byte packedValue = value0 ? (byte) 1 : (byte) 0;
@@ -406,10 +405,10 @@ public class EncodingUtils
packedValue = (byte) (packedValue | (byte) (1 << 1));
}
- buffer.write(packedValue);
+ buffer.put(packedValue);
}
- public static void writeBooleans(DataOutputStream buffer, boolean value0, boolean value1, boolean value2) throws IOException
+ public static void writeBooleans(ByteBuffer buffer, boolean value0, boolean value1, boolean value2)
{
byte packedValue = value0 ? (byte) 1 : (byte) 0;
@@ -423,10 +422,10 @@ public class EncodingUtils
packedValue = (byte) (packedValue | (byte) (1 << 2));
}
- buffer.write(packedValue);
+ buffer.put(packedValue);
}
- public static void writeBooleans(DataOutputStream buffer, boolean value0, boolean value1, boolean value2, boolean value3) throws IOException
+ public static void writeBooleans(ByteBuffer buffer, boolean value0, boolean value1, boolean value2, boolean value3)
{
byte packedValue = value0 ? (byte) 1 : (byte) 0;
@@ -445,11 +444,11 @@ public class EncodingUtils
packedValue = (byte) (packedValue | (byte) (1 << 3));
}
- buffer.write(packedValue);
+ buffer.put(packedValue);
}
- public static void writeBooleans(DataOutputStream buffer, boolean value0, boolean value1, boolean value2, boolean value3,
- boolean value4) throws IOException
+ public static void writeBooleans(ByteBuffer buffer, boolean value0, boolean value1, boolean value2, boolean value3,
+ boolean value4)
{
byte packedValue = value0 ? (byte) 1 : (byte) 0;
@@ -473,11 +472,11 @@ public class EncodingUtils
packedValue = (byte) (packedValue | (byte) (1 << 4));
}
- buffer.write(packedValue);
+ buffer.put(packedValue);
}
- public static void writeBooleans(DataOutputStream buffer, boolean value0, boolean value1, boolean value2, boolean value3,
- boolean value4, boolean value5) throws IOException
+ public static void writeBooleans(ByteBuffer buffer, boolean value0, boolean value1, boolean value2, boolean value3,
+ boolean value4, boolean value5)
{
byte packedValue = value0 ? (byte) 1 : (byte) 0;
@@ -506,11 +505,11 @@ public class EncodingUtils
packedValue = (byte) (packedValue | (byte) (1 << 5));
}
- buffer.write(packedValue);
+ buffer.put(packedValue);
}
- public static void writeBooleans(DataOutputStream buffer, boolean value0, boolean value1, boolean value2, boolean value3,
- boolean value4, boolean value5, boolean value6) throws IOException
+ public static void writeBooleans(ByteBuffer buffer, boolean value0, boolean value1, boolean value2, boolean value3,
+ boolean value4, boolean value5, boolean value6)
{
byte packedValue = value0 ? (byte) 1 : (byte) 0;
@@ -544,11 +543,11 @@ public class EncodingUtils
packedValue = (byte) (packedValue | (byte) (1 << 6));
}
- buffer.write(packedValue);
+ buffer.put(packedValue);
}
- public static void writeBooleans(DataOutputStream buffer, boolean value0, boolean value1, boolean value2, boolean value3,
- boolean value4, boolean value5, boolean value6, boolean value7) throws IOException
+ public static void writeBooleans(ByteBuffer buffer, boolean value0, boolean value1, boolean value2, boolean value3,
+ boolean value4, boolean value5, boolean value6, boolean value7)
{
byte packedValue = value0 ? (byte) 1 : (byte) 0;
@@ -587,7 +586,7 @@ public class EncodingUtils
packedValue = (byte) (packedValue | (byte) (1 << 7));
}
- buffer.write(packedValue);
+ buffer.put(packedValue);
}
/**
@@ -596,12 +595,12 @@ public class EncodingUtils
* @param buffer
* @param data
*/
- public static void writeLongstr(DataOutputStream buffer, byte[] data) throws IOException
+ public static void writeLongstr(ByteBuffer buffer, byte[] data)
{
if (data != null)
{
writeUnsignedInteger(buffer, data.length);
- buffer.write(data);
+ buffer.put(data);
}
else
{
@@ -609,14 +608,14 @@ public class EncodingUtils
}
}
- public static void writeTimestamp(DataOutputStream buffer, long timestamp) throws IOException
+ public static void writeTimestamp(ByteBuffer buffer, long timestamp)
{
writeLong(buffer, timestamp);
}
- public static boolean[] readBooleans(DataInputStream buffer) throws IOException
+ public static boolean[] readBooleans(ByteBuffer buffer)
{
- final byte packedValue = buffer.readByte();
+ final byte packedValue = buffer.get();
if (packedValue == 0)
{
return ALL_FALSE_ARRAY;
@@ -641,9 +640,9 @@ public class EncodingUtils
return result;
}
- public static FieldTable readFieldTable(DataInputStream buffer) throws AMQFrameDecodingException, IOException
+ public static FieldTable readFieldTable(ByteBuffer buffer) throws AMQFrameDecodingException
{
- long length = ((long)(buffer.readInt())) & 0xFFFFFFFFL;
+ long length = buffer.getUnsignedInt();
if (length == 0)
{
return null;
@@ -654,21 +653,21 @@ public class EncodingUtils
}
}
- public static Content readContent(DataInputStream buffer) throws AMQFrameDecodingException
+ public static Content readContent(ByteBuffer buffer) throws AMQFrameDecodingException
{
// TODO: New Content class required for AMQP 0-9.
return null;
}
- public static AMQShortString readAMQShortString(DataInputStream buffer) throws IOException
+ public static AMQShortString readAMQShortString(ByteBuffer buffer)
{
return AMQShortString.readFromBuffer(buffer);
}
- public static String readShortString(DataInputStream buffer) throws IOException
+ public static String readShortString(ByteBuffer buffer)
{
- short length = (short) (((short)buffer.readByte()) & 0xFF);
+ short length = buffer.getUnsigned();
if (length == 0)
{
return null;
@@ -681,7 +680,7 @@ public class EncodingUtils
// this approach here is valid since we know that all the chars are
// ASCII (0-127)
byte[] stringBytes = new byte[length];
- buffer.read(stringBytes, 0, length);
+ buffer.get(stringBytes, 0, length);
char[] stringChars = new char[length];
for (int i = 0; i < stringChars.length; i++)
{
@@ -692,9 +691,9 @@ public class EncodingUtils
}
}
- public static String readLongString(DataInputStream buffer) throws IOException
+ public static String readLongString(ByteBuffer buffer)
{
- long length = ((long)(buffer.readInt())) & 0xFFFFFFFFL;
+ long length = buffer.getUnsignedInt();
if (length == 0)
{
return "";
@@ -707,7 +706,7 @@ public class EncodingUtils
// this approach here is valid since we know that all the chars are
// ASCII (0-127)
byte[] stringBytes = new byte[(int) length];
- buffer.read(stringBytes, 0, (int) length);
+ buffer.get(stringBytes, 0, (int) length);
char[] stringChars = new char[(int) length];
for (int i = 0; i < stringChars.length; i++)
{
@@ -718,9 +717,9 @@ public class EncodingUtils
}
}
- public static byte[] readLongstr(DataInputStream buffer) throws IOException
+ public static byte[] readLongstr(ByteBuffer buffer)
{
- long length = ((long)(buffer.readInt())) & 0xFFFFFFFFL;
+ long length = buffer.getUnsignedInt();
if (length == 0)
{
return null;
@@ -728,17 +727,17 @@ public class EncodingUtils
else
{
byte[] result = new byte[(int) length];
- buffer.read(result);
+ buffer.get(result);
return result;
}
}
- public static long readTimestamp(DataInputStream buffer) throws IOException
+ public static long readTimestamp(ByteBuffer buffer)
{
// Discard msb from AMQ timestamp
// buffer.getUnsignedInt();
- return buffer.readLong();
+ return buffer.getLong();
}
static byte[] hexToByteArray(String id)
@@ -818,14 +817,14 @@ public class EncodingUtils
// AMQP_BOOLEAN_PROPERTY_PREFIX
- public static void writeBoolean(DataOutputStream buffer, Boolean aBoolean) throws IOException
+ public static void writeBoolean(ByteBuffer buffer, Boolean aBoolean)
{
- buffer.write(aBoolean ? 1 : 0);
+ buffer.put((byte) (aBoolean ? 1 : 0));
}
- public static boolean readBoolean(DataInputStream buffer) throws IOException
+ public static boolean readBoolean(ByteBuffer buffer)
{
- byte packedValue = buffer.readByte();
+ byte packedValue = buffer.get();
return (packedValue == 1);
}
@@ -836,14 +835,14 @@ public class EncodingUtils
}
// AMQP_BYTE_PROPERTY_PREFIX
- public static void writeByte(DataOutputStream buffer, Byte aByte) throws IOException
+ public static void writeByte(ByteBuffer buffer, Byte aByte)
{
- buffer.writeByte(aByte);
+ buffer.put(aByte);
}
- public static byte readByte(DataInputStream buffer) throws IOException
+ public static byte readByte(ByteBuffer buffer)
{
- return buffer.readByte();
+ return buffer.get();
}
public static int encodedByteLength()
@@ -852,14 +851,14 @@ public class EncodingUtils
}
// AMQP_SHORT_PROPERTY_PREFIX
- public static void writeShort(DataOutputStream buffer, Short aShort) throws IOException
+ public static void writeShort(ByteBuffer buffer, Short aShort)
{
- buffer.writeShort(aShort);
+ buffer.putShort(aShort);
}
- public static short readShort(DataInputStream buffer) throws IOException
+ public static short readShort(ByteBuffer buffer)
{
- return buffer.readShort();
+ return buffer.getShort();
}
public static int encodedShortLength()
@@ -868,14 +867,14 @@ public class EncodingUtils
}
// INTEGER_PROPERTY_PREFIX
- public static void writeInteger(DataOutputStream buffer, Integer aInteger) throws IOException
+ public static void writeInteger(ByteBuffer buffer, Integer aInteger)
{
- buffer.writeInt(aInteger);
+ buffer.putInt(aInteger);
}
- public static int readInteger(DataInputStream buffer) throws IOException
+ public static int readInteger(ByteBuffer buffer)
{
- return buffer.readInt();
+ return buffer.getInt();
}
public static int encodedIntegerLength()
@@ -884,14 +883,14 @@ public class EncodingUtils
}
// AMQP_LONG_PROPERTY_PREFIX
- public static void writeLong(DataOutputStream buffer, Long aLong) throws IOException
+ public static void writeLong(ByteBuffer buffer, Long aLong)
{
- buffer.writeLong(aLong);
+ buffer.putLong(aLong);
}
- public static long readLong(DataInputStream buffer) throws IOException
+ public static long readLong(ByteBuffer buffer)
{
- return buffer.readLong();
+ return buffer.getLong();
}
public static int encodedLongLength()
@@ -900,14 +899,14 @@ public class EncodingUtils
}
// Float_PROPERTY_PREFIX
- public static void writeFloat(DataOutputStream buffer, Float aFloat) throws IOException
+ public static void writeFloat(ByteBuffer buffer, Float aFloat)
{
- buffer.writeFloat(aFloat);
+ buffer.putFloat(aFloat);
}
- public static float readFloat(DataInputStream buffer) throws IOException
+ public static float readFloat(ByteBuffer buffer)
{
- return buffer.readFloat();
+ return buffer.getFloat();
}
public static int encodedFloatLength()
@@ -916,14 +915,14 @@ public class EncodingUtils
}
// Double_PROPERTY_PREFIX
- public static void writeDouble(DataOutputStream buffer, Double aDouble) throws IOException
+ public static void writeDouble(ByteBuffer buffer, Double aDouble)
{
- buffer.writeDouble(aDouble);
+ buffer.putDouble(aDouble);
}
- public static double readDouble(DataInputStream buffer) throws IOException
+ public static double readDouble(ByteBuffer buffer)
{
- return buffer.readDouble();
+ return buffer.getDouble();
}
public static int encodedDoubleLength()
@@ -931,9 +930,9 @@ public class EncodingUtils
return 8;
}
- public static byte[] readBytes(DataInputStream buffer) throws IOException
+ public static byte[] readBytes(ByteBuffer buffer)
{
- long length = ((long)(buffer.readInt())) & 0xFFFFFFFFL;
+ long length = buffer.getUnsignedInt();
if (length == 0)
{
return null;
@@ -941,19 +940,19 @@ public class EncodingUtils
else
{
byte[] dataBytes = new byte[(int)length];
- buffer.read(dataBytes, 0, (int) length);
+ buffer.get(dataBytes, 0, (int)length);
return dataBytes;
}
}
- public static void writeBytes(DataOutputStream buffer, byte[] data) throws IOException
+ public static void writeBytes(ByteBuffer buffer, byte[] data)
{
if (data != null)
{
// TODO: check length fits in an unsigned byte
writeUnsignedInteger(buffer, (long)data.length);
- buffer.write(data);
+ buffer.put(data);
}
else
{
@@ -969,35 +968,35 @@ public class EncodingUtils
return encodedByteLength();
}
- public static char readChar(DataInputStream buffer) throws IOException
+ public static char readChar(ByteBuffer buffer)
{
// This is valid as we know that the Character is ASCII 0..127
- return (char) buffer.read();
+ return (char) buffer.get();
}
- public static void writeChar(DataOutputStream buffer, char character) throws IOException
+ public static void writeChar(ByteBuffer buffer, char character)
{
// This is valid as we know that the Character is ASCII 0..127
writeByte(buffer, (byte) character);
}
- public static long readLongAsShortString(DataInputStream buffer) throws IOException
+ public static long readLongAsShortString(ByteBuffer buffer)
{
- short length = (short) buffer.readUnsignedByte();
+ short length = buffer.getUnsigned();
short pos = 0;
if (length == 0)
{
return 0L;
}
- byte digit = buffer.readByte();
+ byte digit = buffer.get();
boolean isNegative;
long result = 0;
if (digit == (byte) '-')
{
isNegative = true;
pos++;
- digit = buffer.readByte();
+ digit = buffer.get();
}
else
{
@@ -1010,7 +1009,7 @@ public class EncodingUtils
while (pos < length)
{
pos++;
- digit = buffer.readByte();
+ digit = buffer.get();
result = (result << 3) + (result << 1);
result += digit - (byte) '0';
}
@@ -1018,15 +1017,15 @@ public class EncodingUtils
return result;
}
- public static long readUnsignedInteger(DataInputStream buffer) throws IOException
+ public static long readUnsignedInteger(ByteBuffer buffer)
{
- long l = 0xFF & buffer.readByte();
+ long l = 0xFF & buffer.get();
l <<= 8;
- l = l | (0xFF & buffer.readByte());
+ l = l | (0xFF & buffer.get());
l <<= 8;
- l = l | (0xFF & buffer.readByte());
+ l = l | (0xFF & buffer.get());
l <<= 8;
- l = l | (0xFF & buffer.readByte());
+ l = l | (0xFF & buffer.get());
return l;
}
Modified: qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/FieldTable.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/FieldTable.java?rev=1187150&r1=1187149&r2=1187150&view=diff
==============================================================================
--- qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/FieldTable.java (original)
+++ qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/FieldTable.java Fri Oct 21 01:19:00 2011
@@ -20,16 +20,12 @@
*/
package org.apache.qpid.framing;
+import org.apache.mina.common.ByteBuffer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.qpid.AMQPInvalidClassException;
-import java.io.ByteArrayInputStream;
-import java.io.ByteArrayOutputStream;
-import java.io.DataInputStream;
-import java.io.DataOutputStream;
-import java.io.IOException;
import java.math.BigDecimal;
import java.util.Collections;
import java.util.Enumeration;
@@ -47,8 +43,8 @@ public class FieldTable
private static final String STRICT_AMQP = "STRICT_AMQP";
private final boolean _strictAMQP = Boolean.valueOf(System.getProperty(STRICT_AMQP, "false"));
- private byte[] _encodedForm;
- private LinkedHashMap<AMQShortString, AMQTypedValue> _properties = null;
+ private ByteBuffer _encodedForm;
+ private LinkedHashMap<AMQShortString, AMQTypedValue> _properties;
private long _encodedSize;
private static final int INITIAL_HASHMAP_CAPACITY = 16;
private static final int INITIAL_ENCODED_FORM_SIZE = 256;
@@ -56,6 +52,9 @@ public class FieldTable
public FieldTable()
{
super();
+ // _encodedForm = ByteBuffer.allocate(INITIAL_ENCODED_FORM_SIZE);
+ // _encodedForm.setAutoExpand(true);
+ // _encodedForm.limit(0);
}
/**
@@ -64,12 +63,16 @@ public class FieldTable
* @param buffer the buffer from which to read data. The length byte must be read already
* @param length the length of the field table. Must be > 0.
*/
- public FieldTable(DataInputStream buffer, long length) throws IOException
+ public FieldTable(ByteBuffer buffer, long length)
{
this();
- _encodedForm = new byte[(int) length];
- buffer.read(_encodedForm);
+ ByteBuffer encodedForm = buffer.slice();
+ encodedForm.limit((int) length);
+ _encodedForm = ByteBuffer.allocate((int)length);
+ _encodedForm.put(encodedForm);
+ _encodedForm.flip();
_encodedSize = length;
+ buffer.skip((int) length);
}
public AMQTypedValue getProperty(AMQShortString string)
@@ -105,19 +108,13 @@ public class FieldTable
{
try
{
- setFromBuffer();
+ setFromBuffer(_encodedForm, _encodedSize);
}
catch (AMQFrameDecodingException e)
{
_logger.error("Error decoding FieldTable in deferred decoding mode ", e);
throw new IllegalArgumentException(e);
}
- catch (IOException e)
- {
- _logger.error("Unexpected IO exception decoding field table");
- throw new IllegalArgumentException(e);
-
- }
}
private AMQTypedValue setProperty(AMQShortString key, AMQTypedValue val)
@@ -769,7 +766,7 @@ public class FieldTable
// ************************* Byte Buffer Processing
- public void writeToBuffer(DataOutputStream buffer) throws IOException
+ public void writeToBuffer(ByteBuffer buffer)
{
final boolean trace = _logger.isDebugEnabled();
@@ -789,21 +786,17 @@ public class FieldTable
public byte[] getDataAsBytes()
{
- if(_encodedForm == null)
- {
- ByteArrayOutputStream baos = new ByteArrayOutputStream();
- try
- {
- putDataInBuffer(new DataOutputStream(baos));
- return baos.toByteArray();
- }
- catch (IOException e)
- {
- throw new IllegalArgumentException("IO Exception should never be thrown here");
- }
+ final int encodedSize = (int) getEncodedSize();
+ final ByteBuffer buffer = ByteBuffer.allocate(encodedSize); // FIXME XXX: Is cast a problem?
- }
- return _encodedForm.clone();
+ putDataInBuffer(buffer);
+
+ final byte[] result = new byte[encodedSize];
+ buffer.flip();
+ buffer.get(result);
+ buffer.release();
+
+ return result;
}
public long getEncodedSize()
@@ -933,8 +926,15 @@ public class FieldTable
public Iterator<Map.Entry<AMQShortString, AMQTypedValue>> iterator()
{
- initMapIfNecessary();
- return _properties.entrySet().iterator();
+ if(_encodedForm != null)
+ {
+ return new FieldTableIterator(_encodedForm.duplicate().rewind(),(int)_encodedSize);
+ }
+ else
+ {
+ initMapIfNecessary();
+ return _properties.entrySet().iterator();
+ }
}
public Object get(String key)
@@ -1002,12 +1002,26 @@ public class FieldTable
return _properties.keySet();
}
- private void putDataInBuffer(DataOutputStream buffer) throws IOException
+ private void putDataInBuffer(ByteBuffer buffer)
{
if (_encodedForm != null)
{
- buffer.write(_encodedForm);
+ if(buffer.isDirect() || buffer.isReadOnly())
+ {
+ ByteBuffer encodedForm = _encodedForm.duplicate();
+
+ if (encodedForm.position() != 0)
+ {
+ encodedForm.flip();
+ }
+
+ buffer.put(encodedForm);
+ }
+ else
+ {
+ buffer.put(_encodedForm.array(),_encodedForm.arrayOffset(),(int)_encodedSize);
+ }
}
else if (_properties != null)
{
@@ -1021,27 +1035,41 @@ public class FieldTable
final Map.Entry<AMQShortString, AMQTypedValue> me = it.next();
try
{
+ if (_logger.isDebugEnabled())
+ {
+ _logger.debug("Writing Property:" + me.getKey() + " Type:" + me.getValue().getType() + " Value:"
+ + me.getValue().getValue());
+ _logger.debug("Buffer Position:" + buffer.position() + " Remaining:" + buffer.remaining());
+ }
+
// Write the actual parameter name
EncodingUtils.writeShortStringBytes(buffer, me.getKey());
me.getValue().writeToBuffer(buffer);
}
catch (Exception e)
{
+ if (_logger.isDebugEnabled())
+ {
+ _logger.debug("Exception thrown:" + e);
+ _logger.debug("Writing Property:" + me.getKey() + " Type:" + me.getValue().getType() + " Value:"
+ + me.getValue().getValue());
+ _logger.debug("Buffer Position:" + buffer.position() + " Remaining:" + buffer.remaining());
+ }
+
throw new RuntimeException(e);
}
}
}
}
- private void setFromBuffer() throws AMQFrameDecodingException, IOException
+ private void setFromBuffer(ByteBuffer buffer, long length) throws AMQFrameDecodingException
{
- final ByteArrayInputStream in = new ByteArrayInputStream(_encodedForm);
- DataInputStream buffer = new DataInputStream(in);
final boolean trace = _logger.isDebugEnabled();
- if (_encodedSize > 0)
+ if (length > 0)
{
+ final int expectedRemaining = buffer.remaining() - (int) length;
_properties = new LinkedHashMap<AMQShortString, AMQTypedValue>(INITIAL_HASHMAP_CAPACITY);
@@ -1049,16 +1077,121 @@ public class FieldTable
{
final AMQShortString key = EncodingUtils.readAMQShortString(buffer);
+
+ _logger.debug("FieldTable::PropFieldTable(buffer," + length + "): Read key '" + key);
+
AMQTypedValue value = AMQTypedValue.readFromBuffer(buffer);
+
+ if (trace)
+ {
+ _logger.debug("FieldTable::PropFieldTable(buffer," + length + "): Read type '" + value.getType()
+ + "', key '" + key + "', value '" + value.getValue() + "'");
+ }
+
_properties.put(key, value);
}
- while (in.available() > 0);
+ while (buffer.remaining() > expectedRemaining);
+
+ }
+
+ _encodedSize = length;
+
+ if (trace)
+ {
+ _logger.debug("FieldTable::FieldTable(buffer," + length + "): Done.");
+ }
+ }
+
+ private static final class FieldTableEntry implements Map.Entry<AMQShortString, AMQTypedValue>
+ {
+ private final AMQTypedValue _value;
+ private final AMQShortString _key;
+
+ public FieldTableEntry(final AMQShortString key, final AMQTypedValue value)
+ {
+ _key = key;
+ _value = value;
+ }
+
+ public AMQShortString getKey()
+ {
+ return _key;
+ }
+
+ public AMQTypedValue getValue()
+ {
+ return _value;
+ }
+
+ public AMQTypedValue setValue(final AMQTypedValue value)
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ public boolean equals(Object o)
+ {
+ if(o instanceof FieldTableEntry)
+ {
+ FieldTableEntry other = (FieldTableEntry) o;
+ return (_key == null ? other._key == null : _key.equals(other._key))
+ && (_value == null ? other._value == null : _value.equals(other._value));
+ }
+ else
+ {
+ return false;
+ }
+ }
+
+ public int hashCode()
+ {
+ return (getKey()==null ? 0 : getKey().hashCode())
+ ^ (getValue()==null ? 0 : getValue().hashCode());
+ }
+
+ }
+
+
+ private static final class FieldTableIterator implements Iterator<Map.Entry<AMQShortString, AMQTypedValue>>
+ {
+ private final ByteBuffer _buffer;
+ private int _expectedRemaining;
+
+ public FieldTableIterator(ByteBuffer buffer, int length)
+ {
+ _buffer = buffer;
+ _expectedRemaining = buffer.remaining() - length;
+ }
+
+ public boolean hasNext()
+ {
+ return (_buffer.remaining() > _expectedRemaining);
}
+ public Map.Entry<AMQShortString, AMQTypedValue> next()
+ {
+ if(hasNext())
+ {
+ final AMQShortString key = EncodingUtils.readAMQShortString(_buffer);
+ AMQTypedValue value = AMQTypedValue.readFromBuffer(_buffer);
+ return new FieldTableEntry(key, value);
+ }
+ else
+ {
+ return null;
+ }
+ }
+
+ public void remove()
+ {
+ throw new UnsupportedOperationException();
+ }
}
+
+
+
public int hashCode()
{
initMapIfNecessary();
Modified: qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/FieldTableFactory.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/FieldTableFactory.java?rev=1187150&r1=1187149&r2=1187150&view=diff
==============================================================================
--- qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/FieldTableFactory.java (original)
+++ qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/FieldTableFactory.java Fri Oct 21 01:19:00 2011
@@ -20,8 +20,7 @@
*/
package org.apache.qpid.framing;
-import java.io.DataInputStream;
-import java.io.IOException;
+import org.apache.mina.common.ByteBuffer;
public class FieldTableFactory
{
@@ -30,7 +29,7 @@ public class FieldTableFactory
return new FieldTable();
}
- public static FieldTable newFieldTable(DataInputStream byteBuffer, long length) throws AMQFrameDecodingException, IOException
+ public static FieldTable newFieldTable(ByteBuffer byteBuffer, long length) throws AMQFrameDecodingException
{
return new FieldTable(byteBuffer, length);
}
Modified: qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/HeartbeatBody.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/HeartbeatBody.java?rev=1187150&r1=1187149&r2=1187150&view=diff
==============================================================================
--- qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/HeartbeatBody.java (original)
+++ qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/HeartbeatBody.java Fri Oct 21 01:19:00 2011
@@ -20,10 +20,7 @@
*/
package org.apache.qpid.framing;
-import java.io.DataInputStream;
-import java.io.DataOutputStream;
-import java.io.IOException;
-
+import org.apache.mina.common.ByteBuffer;
import org.apache.qpid.protocol.AMQVersionAwareProtocolSession;
import org.apache.qpid.AMQException;
@@ -37,12 +34,12 @@ public class HeartbeatBody implements AM
}
- public HeartbeatBody(DataInputStream buffer, long size) throws IOException
+ public HeartbeatBody(ByteBuffer buffer, long size)
{
if(size > 0)
{
//allow other implementations to have a payload, but ignore it:
- buffer.skip(size);
+ buffer.skip((int) size);
}
}
@@ -56,7 +53,7 @@ public class HeartbeatBody implements AM
return 0;//heartbeats we generate have no payload
}
- public void writePayload(DataOutputStream buffer)
+ public void writePayload(ByteBuffer buffer)
{
}
@@ -66,12 +63,12 @@ public class HeartbeatBody implements AM
session.heartbeatBodyReceived(channelId, this);
}
- protected void populateFromBuffer(DataInputStream buffer, long size) throws AMQFrameDecodingException, IOException
+ protected void populateFromBuffer(ByteBuffer buffer, long size) throws AMQFrameDecodingException
{
if(size > 0)
{
//allow other implementations to have a payload, but ignore it:
- buffer.skip(size);
+ buffer.skip((int) size);
}
}
Modified: qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/HeartbeatBodyFactory.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/HeartbeatBodyFactory.java?rev=1187150&r1=1187149&r2=1187150&view=diff
==============================================================================
--- qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/HeartbeatBodyFactory.java (original)
+++ qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/HeartbeatBodyFactory.java Fri Oct 21 01:19:00 2011
@@ -20,11 +20,11 @@
*/
package org.apache.qpid.framing;
-import java.io.DataInputStream;
+import org.apache.mina.common.ByteBuffer;
public class HeartbeatBodyFactory implements BodyFactory
{
- public AMQBody createBody(DataInputStream in, long bodySize) throws AMQFrameDecodingException
+ public AMQBody createBody(ByteBuffer in, long bodySize) throws AMQFrameDecodingException
{
return new HeartbeatBody();
}
Modified: qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/ProtocolInitiation.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/ProtocolInitiation.java?rev=1187150&r1=1187149&r2=1187150&view=diff
==============================================================================
--- qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/ProtocolInitiation.java (original)
+++ qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/ProtocolInitiation.java Fri Oct 21 01:19:00 2011
@@ -22,10 +22,6 @@ package org.apache.qpid.framing;
import org.apache.qpid.AMQException;
-import java.io.DataInputStream;
-import java.io.DataOutputStream;
-import java.io.IOException;
-
import java.io.UnsupportedEncodingException;
import java.nio.ByteBuffer;
import java.util.Arrays;
@@ -66,30 +62,35 @@ public class ProtocolInitiation extends
pv.equals(ProtocolVersion.v0_91) ? 1 : pv.getMinorVersion());
}
- public ProtocolInitiation(DataInputStream in) throws IOException
+ public ProtocolInitiation(ByteBuffer in)
{
_protocolHeader = new byte[4];
- in.read(_protocolHeader);
+ in.get(_protocolHeader);
- _protocolClass = in.readByte();
- _protocolInstance = in.readByte();
- _protocolMajor = in.readByte();
- _protocolMinor = in.readByte();
+ _protocolClass = in.get();
+ _protocolInstance = in.get();
+ _protocolMajor = in.get();
+ _protocolMinor = in.get();
}
+ public void writePayload(org.apache.mina.common.ByteBuffer buffer)
+ {
+ writePayload(buffer.buf());
+ }
+
public long getSize()
{
return 4 + 1 + 1 + 1 + 1;
}
- public void writePayload(DataOutputStream buffer) throws IOException
+ public void writePayload(ByteBuffer buffer)
{
- buffer.write(_protocolHeader);
- buffer.write(_protocolClass);
- buffer.write(_protocolInstance);
- buffer.write(_protocolMajor);
- buffer.write(_protocolMinor);
+ buffer.put(_protocolHeader);
+ buffer.put(_protocolClass);
+ buffer.put(_protocolInstance);
+ buffer.put(_protocolMajor);
+ buffer.put(_protocolMinor);
}
public boolean equals(Object o)
@@ -143,9 +144,9 @@ public class ProtocolInitiation extends
* @return true if we have enough data to decode the PI frame fully, false if more
* data is required
*/
- public boolean decodable(DataInputStream in) throws IOException
+ public boolean decodable(ByteBuffer in)
{
- return (in.available() >= 8);
+ return (in.remaining() >= 8);
}
}
Modified: qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/SmallCompositeAMQDataBlock.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/SmallCompositeAMQDataBlock.java?rev=1187150&r1=1187149&r2=1187150&view=diff
==============================================================================
--- qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/SmallCompositeAMQDataBlock.java (original)
+++ qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/SmallCompositeAMQDataBlock.java Fri Oct 21 01:19:00 2011
@@ -21,8 +21,7 @@
package org.apache.qpid.framing;
-import java.io.DataOutputStream;
-import java.io.IOException;
+import org.apache.mina.common.ByteBuffer;
public class SmallCompositeAMQDataBlock extends AMQDataBlock implements EncodableAMQDataBlock
{
@@ -69,7 +68,7 @@ public class SmallCompositeAMQDataBlock
return frameSize;
}
- public void writePayload(DataOutputStream buffer) throws IOException
+ public void writePayload(ByteBuffer buffer)
{
if (_firstFrame != null)
{
Modified: qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/VersionSpecificRegistry.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/VersionSpecificRegistry.java?rev=1187150&r1=1187149&r2=1187150&view=diff
==============================================================================
--- qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/VersionSpecificRegistry.java (original)
+++ qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/VersionSpecificRegistry.java Fri Oct 21 01:19:00 2011
@@ -20,8 +20,7 @@
*/
package org.apache.qpid.framing;
-import java.io.DataInputStream;
-import java.io.IOException;
+import org.apache.mina.common.ByteBuffer;
import org.apache.qpid.framing.abstraction.ProtocolVersionMethodConverter;
@@ -145,7 +144,7 @@ public class VersionSpecificRegistry
}
- public AMQMethodBody get(short classID, short methodID, DataInputStream in, long size) throws AMQFrameDecodingException, IOException
+ public AMQMethodBody get(short classID, short methodID, ByteBuffer in, long size) throws AMQFrameDecodingException
{
AMQMethodBodyInstanceFactory bodyFactory;
try
Modified: qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/abstraction/ContentChunk.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/abstraction/ContentChunk.java?rev=1187150&r1=1187149&r2=1187150&view=diff
==============================================================================
--- qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/abstraction/ContentChunk.java (original)
+++ qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/abstraction/ContentChunk.java Fri Oct 21 01:19:00 2011
@@ -21,10 +21,12 @@
package org.apache.qpid.framing.abstraction;
+import org.apache.mina.common.ByteBuffer;
+
public interface ContentChunk
{
int getSize();
- byte[] getData();
+ ByteBuffer getData();
void reduceToFit();
}
Modified: qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/abstraction/ProtocolVersionMethodConverter.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/abstraction/ProtocolVersionMethodConverter.java?rev=1187150&r1=1187149&r2=1187150&view=diff
==============================================================================
--- qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/abstraction/ProtocolVersionMethodConverter.java (original)
+++ qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/abstraction/ProtocolVersionMethodConverter.java Fri Oct 21 01:19:00 2011
@@ -23,6 +23,8 @@ package org.apache.qpid.framing.abstract
import org.apache.qpid.framing.AMQBody;
+import java.nio.ByteBuffer;
+
public interface ProtocolVersionMethodConverter extends MessagePublishInfoConverter
{
AMQBody convertToBody(ContentChunk contentBody);
@@ -30,5 +32,5 @@ public interface ProtocolVersionMethodCo
void configure();
- AMQBody convertToBody(byte[] input);
+ AMQBody convertToBody(ByteBuffer buf);
}
Modified: qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/amqp_0_9/MethodConverter_0_9.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/amqp_0_9/MethodConverter_0_9.java?rev=1187150&r1=1187149&r2=1187150&view=diff
==============================================================================
--- qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/amqp_0_9/MethodConverter_0_9.java (original)
+++ qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/amqp_0_9/MethodConverter_0_9.java Fri Oct 21 01:19:00 2011
@@ -21,13 +21,16 @@
package org.apache.qpid.framing.amqp_0_9;
+import org.apache.mina.common.ByteBuffer;
+
import org.apache.qpid.framing.abstraction.AbstractMethodConverter;
import org.apache.qpid.framing.abstraction.ProtocolVersionMethodConverter;
import org.apache.qpid.framing.abstraction.ContentChunk;
import org.apache.qpid.framing.abstraction.MessagePublishInfo;
import org.apache.qpid.framing.abstraction.MessagePublishInfoImpl;
import org.apache.qpid.framing.*;
-
+import org.apache.qpid.framing.amqp_0_9.*;
+import org.apache.qpid.framing.amqp_0_9.BasicPublishBodyImpl;
public class MethodConverter_0_9 extends AbstractMethodConverter implements ProtocolVersionMethodConverter
{
@@ -69,9 +72,9 @@ public class MethodConverter_0_9 extends
}
- public AMQBody convertToBody(byte[] data)
+ public AMQBody convertToBody(java.nio.ByteBuffer buf)
{
- return new ContentBody(data);
+ return new ContentBody(ByteBuffer.wrap(buf));
}
public MessagePublishInfo convertToInfo(AMQMethodBody methodBody)
@@ -113,9 +116,9 @@ public class MethodConverter_0_9 extends
return _contentBodyChunk.getSize();
}
- public byte[] getData()
+ public ByteBuffer getData()
{
- return _contentBodyChunk._payload;
+ return _contentBodyChunk.payload;
}
public void reduceToFit()
Modified: qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/amqp_0_91/MethodConverter_0_91.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/amqp_0_91/MethodConverter_0_91.java?rev=1187150&r1=1187149&r2=1187150&view=diff
==============================================================================
--- qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/amqp_0_91/MethodConverter_0_91.java (original)
+++ qpid/branches/QPID-2519/java/common/src/main/java/org/apache/qpid/framing/amqp_0_91/MethodConverter_0_91.java Fri Oct 21 01:19:00 2011
@@ -21,6 +21,8 @@
package org.apache.qpid.framing.amqp_0_91;
+import org.apache.mina.common.ByteBuffer;
+
import org.apache.qpid.framing.abstraction.AbstractMethodConverter;
import org.apache.qpid.framing.abstraction.ProtocolVersionMethodConverter;
import org.apache.qpid.framing.abstraction.ContentChunk;
@@ -68,9 +70,9 @@ public class MethodConverter_0_91 extend
}
- public AMQBody convertToBody(byte[] data)
+ public AMQBody convertToBody(java.nio.ByteBuffer buf)
{
- return new ContentBody(data);
+ return new ContentBody(ByteBuffer.wrap(buf));
}
public MessagePublishInfo convertToInfo(AMQMethodBody methodBody)
@@ -112,9 +114,9 @@ public class MethodConverter_0_91 extend
return _contentBodyChunk.getSize();
}
- public byte[] getData()
+ public ByteBuffer getData()
{
- return _contentBodyChunk._payload;
+ return _contentBodyChunk.payload;
}
public void reduceToFit()
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org