From 6ab3b0ebc020bddfa0166d0d754a6c41219f17af Mon Sep 17 00:00:00 2001 From: Alexey Rybalchenko Date: Mon, 21 Mar 2016 09:59:00 +0100 Subject: [PATCH] Update multi-part features (nanomsg) and various fixes - Implement nanomsg multipart with MessagePack. - Use the MessagePack from FairSoft and handle not found case. - Update splitter, merger and proxy devices to handle multi-part. - Let FairMQParts.At() return pointer reference (can be used for moving). - Add missing const specifier in the message interface. - Add transmit kernel size setting to channels (ZMQ_SNDBUF). - Remove FairMQBuffer device. - Remove old multi-part methods from Tutorial3 example (to be replaced with Parts API). - Make callback mandatory for newMsg(data, size, callback). - Add missing include in FairMQSocket. --- examples/MQ/4-copypush/FairMQExample4Sampler.cxx | 5 ++++- examples/MQ/8-multipart/FairMQExample8Sampler.cxx | 11 ++++++++--- examples/MQ/8-multipart/FairMQExample8Sink.cxx | 4 ++-- 3 files changed, 14 insertions(+), 6 deletions(-) diff --git a/examples/MQ/4-copypush/FairMQExample4Sampler.cxx b/examples/MQ/4-copypush/FairMQExample4Sampler.cxx index aedf70bd..515b7870 100644 --- a/examples/MQ/4-copypush/FairMQExample4Sampler.cxx +++ b/examples/MQ/4-copypush/FairMQExample4Sampler.cxx @@ -34,7 +34,10 @@ void FairMQExample4Sampler::Run() uint64_t* number = new uint64_t(counter); - std::unique_ptr msg(NewMessage(number, sizeof(uint64_t))); + std::unique_ptr msg(NewMessage(number, // data pointer + sizeof(uint64_t), // data size + [](void* data, void* hint){ delete static_cast(data); } // callback to deallocate after the transfer + )); LOG(INFO) << "Sending \"" << counter << "\""; diff --git a/examples/MQ/8-multipart/FairMQExample8Sampler.cxx b/examples/MQ/8-multipart/FairMQExample8Sampler.cxx index 6e4123f7..16edda50 100644 --- a/examples/MQ/8-multipart/FairMQExample8Sampler.cxx +++ b/examples/MQ/8-multipart/FairMQExample8Sampler.cxx @@ -39,12 +39,17 @@ void FairMQExample8Sampler::Run() // Set stopFlag to 1 for the first 4 messages, and to 0 for the 5th. counter < 5 ? header->stopFlag = 0 : header->stopFlag = 1; + LOG(INFO) << "Sending header with stopFlag: " << header->stopFlag; + FairMQParts parts; - parts.AddPart(NewMessage(header, sizeof(Ex8Header))); + + parts.AddPart(NewMessage(header, // data pointer + sizeof(Ex8Header), // data size + [](void* data, void* hint) { delete static_cast(data); } // callback to deallocate after the transfer + )); parts.AddPart(NewMessage(1000)); - LOG(INFO) << "Sending header with stopFlag: " << header->stopFlag; - LOG(INFO) << "Sending body of size: " << parts.At(1).GetSize(); + LOG(INFO) << "Sending body of size: " << parts.At(1)->GetSize(); Send(parts, "data-out"); diff --git a/examples/MQ/8-multipart/FairMQExample8Sink.cxx b/examples/MQ/8-multipart/FairMQExample8Sink.cxx index 0791eb4a..05216d27 100644 --- a/examples/MQ/8-multipart/FairMQExample8Sink.cxx +++ b/examples/MQ/8-multipart/FairMQExample8Sink.cxx @@ -37,9 +37,9 @@ void FairMQExample8Sink::Run() if (Receive(parts, "data-in") >= 0) { Ex8Header header; - header.stopFlag = (static_cast(parts.At(0).GetData()))->stopFlag; + header.stopFlag = (static_cast(parts.At(0)->GetData()))->stopFlag; LOG(INFO) << "Received header with stopFlag: " << header.stopFlag; - LOG(INFO) << "Received body of size: " << parts.At(1).GetSize(); + LOG(INFO) << "Received body of size: " << parts.At(1)->GetSize(); if (header.stopFlag == 1) { LOG(INFO) << "Flag is 0, exiting Run()";