diff --git a/components/codecs/include/ftl/codecs/writer.hpp b/components/codecs/include/ftl/codecs/writer.hpp index 044624ba837f17c83e4a22a16787ec86edd4ad67..94d82f1ecbfb697bdbad35acc5a4cf29dad22410 100644 --- a/components/codecs/include/ftl/codecs/writer.hpp +++ b/components/codecs/include/ftl/codecs/writer.hpp @@ -5,6 +5,7 @@ #include <msgpack.hpp> //#include <Eigen/Eigen> +#include <ftl/threads.hpp> #include <ftl/codecs/packet.hpp> namespace ftl { @@ -24,6 +25,7 @@ class Writer { msgpack::sbuffer buffer_; int64_t timestart_; bool active_; + MUTEX mutex_; }; } diff --git a/components/codecs/src/reader.cpp b/components/codecs/src/reader.cpp index 97a74acd1a16c29e7d5c926382759aaf765b1818..ba68f26c9666da9a84ea86632caa1035bb6c5c88 100644 --- a/components/codecs/src/reader.cpp +++ b/components/codecs/src/reader.cpp @@ -37,6 +37,22 @@ bool Reader::begin() { return true; } +/*static void printMsgpack(msgpack::object &obj) { + switch (obj.type) { + case msgpack::type::NIL: return; + case msgpack::type::BOOLEAN: LOG(INFO) << "BOOL " << obj.as<bool>(); return; + case msgpack::type::POSITIVE_INTEGER: + case msgpack::type::NEGATIVE_INTEGER: LOG(INFO) << "INT " << obj.as<int>(); return; + case msgpack::type::FLOAT32: LOG(INFO) << "FLOAT " << obj.as<float>(); return; + case msgpack::type::FLOAT64: LOG(INFO) << "DOUBLE " << obj.as<double>(); return; + case msgpack::type::STR: LOG(INFO) << "STRING " << obj.as<std::string>(); return; + case msgpack::type::BIN: return; + case msgpack::type::EXT: return; + case msgpack::type::ARRAY: LOG(INFO) << "ARRAY: "; return; + case msgpack::type::MAP: LOG(INFO) << "MAP: "; return; + } +}*/ + bool Reader::read(int64_t ts, const std::function<void(const ftl::codecs::StreamPacket &, ftl::codecs::Packet &)> &f) { //UNIQUE_LOCK(mtx_, lk); std::unique_lock<std::mutex> lk(mtx_, std::defer_lock); @@ -75,10 +91,13 @@ bool Reader::read(int64_t ts, const std::function<void(const ftl::codecs::Stream //std::tuple<StreamPacket,Packet> data; msgpack::object obj = msg.get(); + + //printMsgpack(obj); + try { obj.convert(data_.emplace_back()); } catch (std::exception &e) { - LOG(INFO) << "Corrupt message: " << buffer_.nonparsed_size(); + LOG(INFO) << "Corrupt message: " << buffer_.nonparsed_size() << " - " << e.what(); //partial = true; //continue; return false; diff --git a/components/codecs/src/writer.cpp b/components/codecs/src/writer.cpp index d7582f11866b701796ab311016d2a01a5218595a..1e3841a8e05f840f808aaddf127833392212af05 100644 --- a/components/codecs/src/writer.cpp +++ b/components/codecs/src/writer.cpp @@ -41,7 +41,8 @@ bool Writer::write(const ftl::codecs::StreamPacket &s, const ftl::codecs::Packet auto data = std::make_tuple(s2,p); msgpack::sbuffer buffer; msgpack::pack(buffer, data); + + UNIQUE_LOCK(mutex_, lk); (*stream_).write(buffer.data(), buffer.size()); - //buffer_.clear(); return true; }