QPID-7791: Recover metadata into direct memory, if it will fit into a pooled buffer (least intrusive change for branch).
diff --git a/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuple/MessageMetaDataBinding.java b/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuple/MessageMetaDataBinding.java
index 4f10557..1bc4944 100644
--- a/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuple/MessageMetaDataBinding.java
+++ b/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/tuple/MessageMetaDataBinding.java
@@ -51,13 +51,24 @@
@Override
public StorableMessageMetaData entryToObject(DatabaseEntry entry)
{
- QpidByteBuffer buf = QpidByteBuffer.wrap(entry.getData(), entry.getOffset(), entry.getSize());
+ QpidByteBuffer buf;
+ if (entry.getSize() > QpidByteBuffer.getPooledBufferSize())
+ {
+ buf = QpidByteBuffer.wrap(entry.getData(), entry.getOffset(), entry.getSize());
+ }
+ else
+ {
+ buf = QpidByteBuffer.allocateDirect(entry.getSize());
+ buf.put(entry.getData(), entry.getOffset(), entry.getSize());
+ buf.flip();
+ }
final int bodySize = buf.getInt() ^ 0x80000000;
final int metaDataType = buf.get() & 0xff;
- buf = buf.slice();
- buf.limit(bodySize-1);
+ QpidByteBuffer slice = buf.slice();
+ slice.limit(bodySize-1);
MessageMetaDataType type = MessageMetaDataTypeRegistry.fromOrdinal(metaDataType);
- final StorableMessageMetaData metaData = type.createMetaData(buf);
+ final StorableMessageMetaData metaData = type.createMetaData(slice);
+ slice.dispose();
buf.dispose();
return metaData;
}
diff --git a/broker-core/src/main/java/org/apache/qpid/server/store/AbstractJDBCMessageStore.java b/broker-core/src/main/java/org/apache/qpid/server/store/AbstractJDBCMessageStore.java
index 18a5da3..6b95012 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/store/AbstractJDBCMessageStore.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/store/AbstractJDBCMessageStore.java
@@ -1002,14 +1002,7 @@
if(rs.next())
{
byte[] dataAsBytes = getBlobAsBytes(rs, 1);
- QpidByteBuffer buf = QpidByteBuffer.wrap(dataAsBytes);
- buf.position(1);
- buf = buf.slice();
- int typeOrdinal = dataAsBytes[0] & 0xff;;
- MessageMetaDataType type = MessageMetaDataTypeRegistry.fromOrdinal(typeOrdinal);
- StorableMessageMetaData metaData = type.createMetaData(buf);
- buf.dispose();
- return metaData;
+ return getStorableMessageMetaData(dataAsBytes);
}
else
{
@@ -1761,14 +1754,8 @@
if (rs.next())
{
byte[] dataAsBytes = getBlobAsBytes(rs, 2);
- QpidByteBuffer buf = QpidByteBuffer.wrap(dataAsBytes);
- buf.position(1);
- buf = buf.slice();
- MessageMetaDataType<?> type = MessageMetaDataTypeRegistry.fromOrdinal(dataAsBytes[0]);
- StorableMessageMetaData metaData = type.createMetaData(buf);
- buf.dispose();
+ StorableMessageMetaData metaData = getStorableMessageMetaData(dataAsBytes);
message = createStoredJDBCMessage(messageId, metaData, true);
-
}
else
{
@@ -2040,6 +2027,29 @@
}
}
+ private StorableMessageMetaData getStorableMessageMetaData(final byte[] dataAsBytes)
+ {
+ final QpidByteBuffer buf;
+ if (dataAsBytes.length > QpidByteBuffer.getPooledBufferSize())
+ {
+ buf = QpidByteBuffer.wrap(dataAsBytes);
+ }
+ else
+ {
+ buf = QpidByteBuffer.allocateDirect(dataAsBytes.length);
+ buf.put(dataAsBytes);
+ buf.flip();
+ }
+ buf.position(1);
+ QpidByteBuffer slice = buf.slice();
+ int typeOrdinal = dataAsBytes[0] & 0xff;
+ MessageMetaDataType type = MessageMetaDataTypeRegistry.fromOrdinal(typeOrdinal);
+ StorableMessageMetaData metaData = type.createMetaData(slice);
+ slice.dispose();
+ buf.dispose();
+ return metaData;
+ }
+
protected abstract void storedSizeChange(int storeSizeIncrease);
@Override
