cluster: add Event size to encoded header.

git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid@722728 13f79535-47bb-0310-9956-ffa450edef68

3 files changed, 11 insertions, 8 deletions

diff --git a/cpp/src/qpid/cluster/Cluster.cpp b/cpp/src/qpid/cluster/Cluster.cpp
index 0ac0da2be4..6cad003605 100644
--- a/cpp/src/qpid/cluster/Cluster.cpp
+++ b/cpp/src/qpid/cluster/Cluster.cpp
@@ -261,7 +261,8 @@ void Cluster::deliver(
 {
     Mutex::ScopedLock l(lock);
     MemberId from(nodeid, pid);
-    deliver(Event::delivered(from, msg, msg_len), l);
+    framing::Buffer buf(static_cast<char*>(msg), msg_len);
+    deliver(Event::decode(from, buf), l);
 }
 
 void Cluster::deliver(const Event& e, Lock&) {
diff --git a/cpp/src/qpid/cluster/Event.cpp b/cpp/src/qpid/cluster/Event.cpp
index 0e55d7bce3..5ac63c86f5 100644
--- a/cpp/src/qpid/cluster/Event.cpp
+++ b/cpp/src/qpid/cluster/Event.cpp
@@ -32,20 +32,21 @@ namespace cluster {
 
 using framing::Buffer;
 
-const size_t Event::OVERHEAD = sizeof(uint8_t) + sizeof(uint64_t) + sizeof(uint32_t);
+const size_t Event::OVERHEAD = sizeof(uint8_t) + sizeof(uint64_t) + sizeof(uint32_t) + sizeof(uint32_t);
 
 Event::Event(EventType t, const ConnectionId& c,  size_t s, uint32_t i)
     : type(t), connectionId(c), size(s), data(RefCountedBuffer::create(s)), id(i) {}
 
-Event Event::delivered(const MemberId& m, void* d, size_t s) {
-    Buffer buf(static_cast<char*>(d), s);
+Event Event::decode(const MemberId& m, framing::Buffer& buf) {
+    assert(buf.available() > OVERHEAD);
     EventType type((EventType)buf.getOctet());
     assert(type == DATA || type == CONTROL);
     ConnectionId connection(m, reinterpret_cast<Connection*>(buf.getLongLong()));
     uint32_t id = buf.getLong();
-    assert(buf.getPosition() == OVERHEAD);
-    Event e(type, connection, s-OVERHEAD, id);
-    memcpy(e.getData(), static_cast<char*>(d)+OVERHEAD, s-OVERHEAD);
+    uint32_t size = buf.getLong();
+    Event e(type, connection, size, id);
+    assert(buf.available() >= size);
+    memcpy(e.getData(), buf.getPointer() + buf.getPosition(), size);
     return e;
 }
 
@@ -63,6 +64,7 @@ bool Event::mcast (Cpg& cpg) const {
     b.putOctet(type);
     b.putLongLong(reinterpret_cast<uint64_t>(connectionId.getPointer()));
     b.putLong(id);
+    b.putLong(size);
     assert(b.getPosition() == OVERHEAD);
     iovec iov[] = { { header, OVERHEAD }, { const_cast<char*>(getData()), getSize() } };
     return cpg.mcast(iov, sizeof(iov)/sizeof(*iov));
diff --git a/cpp/src/qpid/cluster/Event.h b/cpp/src/qpid/cluster/Event.h
index b61ce0e60d..e046990747 100644
--- a/cpp/src/qpid/cluster/Event.h
+++ b/cpp/src/qpid/cluster/Event.h
@@ -46,7 +46,7 @@ class Event {
     Event(EventType t=DATA, const ConnectionId& c=ConnectionId(), size_t size=0, uint32_t id=0);
 
     /** Create an event copied from delivered data. */
-    static Event delivered(const MemberId& m, void* data, size_t size);
+    static Event decode(const MemberId& m, framing::Buffer&);
 
     /** Create an event containing a control */
     static Event control(const framing::AMQBody&, const ConnectionId&, uint32_t id=0);