10:41 p.m.
Author: kpvdr
Date: 2009-01-27 16:41:40 -0500 (Tue, 27 Jan 2009)
New Revision: 3075
Modified:
store/trunk/cpp/lib/JournalImpl.cpp
store/trunk/cpp/lib/JournalImpl.h
store/trunk/cpp/lib/MessageStoreImpl.cpp
Log:
Fix for BZ 481822 - "Store: Message content loss during staging/loading if header
size changes in interim".
Modified: store/trunk/cpp/lib/JournalImpl.cpp
===================================================================
--- store/trunk/cpp/lib/JournalImpl.cpp 2009-01-27 15:09:50 UTC (rev 3074)
+++ store/trunk/cpp/lib/JournalImpl.cpp 2009-01-27 21:41:40 UTC (rev 3075)
@@ -237,7 +237,7 @@
#define MAX_AIO_SLEEPS 1000 // 10 sec
#define AIO_SLEEP_TIME 10000 // 10 ms
bool
-JournalImpl::loadMsgContent(u_int64_t rid, std::string& data, size_t offset, size_t
length)
+JournalImpl::loadMsgContent(u_int64_t rid, std::string& data, size_t length)
{
if (_dtok.rid() != rid)
{
@@ -307,6 +307,7 @@
}
if (_external)
return false;
+ u_int32_t offset = qpid::framing::Buffer(static_cast<char*>(_datap),
sizeof(u_int32_t)).getLong() + sizeof(u_int32_t);
if (offset + length > _dlen) {
data.append((const char*)_datap + offset, _dlen - offset);
} else {
Modified: store/trunk/cpp/lib/JournalImpl.h
===================================================================
--- store/trunk/cpp/lib/JournalImpl.h 2009-01-27 15:09:50 UTC (rev 3074)
+++ store/trunk/cpp/lib/JournalImpl.h 2009-01-27 21:41:40 UTC (rev 3075)
@@ -148,7 +148,7 @@
// Temporary fn to read and save last msg read from journal so it can be
assigned
// in chunks. To be replaced when coding to do this direct from the journal
is ready.
// Returns true if the record is extern, false if local.
- bool loadMsgContent(u_int64_t rid, std::string& data, size_t offset,
size_t length);
+ bool loadMsgContent(u_int64_t rid, std::string& data, size_t length);
// Overrides for write inactivity timer
void enqueue_data_record(const void* const data_buff, const size_t
tot_data_len,
Modified: store/trunk/cpp/lib/MessageStoreImpl.cpp
===================================================================
--- store/trunk/cpp/lib/MessageStoreImpl.cpp 2009-01-27 15:09:50 UTC (rev 3074)
+++ store/trunk/cpp/lib/MessageStoreImpl.cpp 2009-01-27 21:41:40 UTC (rev 3075)
@@ -1258,14 +1258,13 @@
u_int32_t length)
{
checkInit();
- u_int64_t realOffset = offset + sizeof(u_int32_t)/*header length*/ +
msg->encodedHeaderSize();
u_int64_t messageId (msg->getPersistenceId());
if (messageId != 0) {
try {
JournalImpl* jc =
static_cast<JournalImpl*>(queue.getExternalQueueStore());
if (jc && jc->is_enqueued(messageId) ) {
- if (jc->loadMsgContent(messageId, data, realOffset, length)) {
+ if (jc->loadMsgContent(messageId, data, length)) {
return;
}
}
@@ -1278,11 +1277,26 @@
Dbt key (&messageId, sizeof(messageId));
char *buffer = new char[length];
Dbt value(buffer, length);
+
+ // Read the first 4 bytes (u_int32_t) which is the header size.
value.set_flags(DB_DBT_USERMEM | DB_DBT_PARTIAL);
+ value.set_ulen(sizeof(u_int32_t));
+ value.set_doff(0);
+ value.set_dlen(sizeof(u_int32_t));
+ int status = messageDb->get(txn.get(), &key, &value, 0);
+ if (status == DB_NOTFOUND) {
+ delete [] buffer;
+ THROW_STORE_EXCEPTION("Cannot load content. Message not known to
store!");
+ }
+ u_int32_t hdr_size = Buffer(buffer, sizeof(u_int32_t)).getLong();
+
+ // Now read the data.
+ u_int64_t realOffset = sizeof(u_int32_t) + hdr_size + offset;
+ value.set_flags(DB_DBT_USERMEM | DB_DBT_PARTIAL);
value.set_ulen(length);
value.set_doff(realOffset);
value.set_dlen(length);
- int status = messageDb->get(txn.get(), &key, &value, 0);
+ status = messageDb->get(txn.get(), &key, &value, 0);
if (status == DB_NOTFOUND) {
delete [] buffer;
THROW_STORE_EXCEPTION("Cannot load content. Message not known to
store!");