MessagePack 學習筆記

封裝和解析類似json的  key-value 示例

{"ID" = 333,"name"="zds","3333"="ende"}

    msgpack::sbuffer sBuf;
    msgpack::packer<msgpack::sbuffer> pker(&sBuf);

    pker.pack_map(3);
    pker.pack(std::string("ID"));
    pker.pack(333);
    pker.pack(std::string("name"));
    pker.pack(std::string("zds"));
    pker.pack(std::string("333"));
    pker.pack(std::string("ende"));


    //unserilized
    msgpack::unpacked unpack;
    msgpack::unpack(unpack, sBuf.data(), sBuf.size());

    msgpack::object obj = unpack.get();
    std::cout << obj << std::endl;


    if (obj.type == msgpack::type::ARRAY)
        std::cout << "是array" << std::endl;
    else if (obj.type == msgpack::type::MAP)
        std::cout << "是map" << std::endl;




    if(obj.via.map.size > 0)
    {
        auto pkv = obj.via.map.ptr;
        auto pkv_end = obj.via.map.ptr + obj.via.map.size;

        do 
        {
            auto key = pkv->key;
            auto val = pkv->val;
            std::cout << "key:" << key << " value:" << val << std::endl;
            ++pkv;
        } while (pkv < pkv_end);
    }

解析Socket示例

各類數據結構:

msgpack::object 他是一個引用，拷貝他的代價少，因為他是淺拷貝
msgpack::object_handle  他管理了一個對象的生命周期。他如果釋放了，所有從他生成的object都是無效的引用

 

解析Socket示例

下列代碼解析socke收包數據

unpacker.reserve_buffer 分配要收的數據的內存字節數
unpacker..buffer() 返回數據地址
unpacker.buffer_consumed() 設置實際收到的數據
unpacker.next(object_handle& oh) 循環解析數據

int main() {
    boost::asio::io_service ios;
    std::uint16_t const port = 12345;

    // Server
    std::size_t const window_size = 10;
    boost::asio::ip::tcp::acceptor ac(ios, boost::asio::ip::tcp::endpoint(boost::asio::ip::tcp::v4(), port));
    boost::asio::ip::tcp::socket ss(ios);
    std::function<void()> do_accept;
    std::function<void()> do_async_read_some;

    msgpack::unpacker unp;

    do_accept = [&] {
        ac.async_accept(
            ss,
            [&]
            (boost::system::error_code const& e) {
                if (e) {
                    std::cout << __LINE__ << ":" << e.message() << std::endl;
                    return;
                }
                unp.reserve_buffer(window_size);
                do_async_read_some = [&] {
                    ss.async_read_some(
                        boost::asio::buffer(unp.buffer(), window_size),
                        [&](boost::system::error_code const& e, std::size_t bytes_transferred) {
                            if (e) {
                                std::cout << __LINE__ << ":" << e.message() << std::endl;
                                return;
                            }
                            std::cout << bytes_transferred << " bytes read." << std::endl;
                            unp.buffer_consumed(bytes_transferred);
                            msgpack::object_handle oh;
                            while (unp.next(oh)) {
                                std::cout << oh.get() << std::endl;
                                // In order to finish the program,
                                // return if one complete msgpack is processed.
                                // In actual server, don't return here.
                                return;
                            }
                            do_async_read_some();
                        }
                    );
                };
                do_async_read_some();
            }
        );
    };
    do_accept();

    // Client
    auto host = "localhost";
    boost::asio::ip::tcp::resolver r(ios);
    boost::asio::ip::tcp::resolver::query q(host, boost::lexical_cast<std::string>(port));
    auto it = r.resolve(q);
    boost::asio::ip::tcp::socket cs(ios);
    boost::asio::async_connect(
        cs,

        it,
        [&]
        (boost::system::error_code const& e, boost::asio::ip::tcp::resolver::iterator) {
            if (e) {
                std::cout << __LINE__ << ":" << e.message() << std::endl;
                return;
            }
            std::cout << __LINE__ << ":client connected" << std::endl;
            msgpack::sbuffer sb;
            msgpack::pack(sb, std::make_tuple(42, false, "hello world", 12.3456));
            write(cs, boost::asio::buffer(sb.data(), sb.size()));
        }
    );

    // Start
    ios.run();
}

詳解：

msgpack controls a buffer

msgpack provides a buffer management functionality named msgpack::unpacker. msgpack::unpacker is sutable for the following motivations:

• msgpack data is chopped, and the client doesn't know when it will complete. This is a typical situation when you develop streaming applications.
• You want to minimize copy opperations without careful memory management.

Here is the basic (not all) interface of msgpack::unpacker:

#ifndef MSGPACK_UNPACKER_INIT_BUFFER_SIZE
#define MSGPACK_UNPACKER_INIT_BUFFER_SIZE (64*1024) #endif #ifndef MSGPACK_UNPACKER_RESERVE_SIZE #define MSGPACK_UNPACKER_RESERVE_SIZE (32*1024) #endif class unpacker { public: unpacker(unpack_reference_func f = &unpacker::default_reference_func, void* user_data = nullptr, std::size_t init_buffer_size = MSGPACK_UNPACKER_INIT_BUFFER_SIZE, unpack_limit const& limit = unpack_limit()); void reserve_buffer(std::size_t size = MSGPACK_UNPACKER_RESERVE_SIZE); char* buffer(); void buffer_consumed(std::size_t size); bool next(unpacked& result); };

Here is a basic pattern using msgpack::unpacker:

// The size may decided by receive performance, transmit layer's protocol and so on. std::size_t const try_read_size = 100; msgpack::unpacker unp; // Message receive loop while (/* block until input becomes readable */) { unp.reserve_buffer(try_read_size); // unp has at least try_read_size buffer on this point. // input is a kind of I/O library object. // read message to msgpack::unpacker's internal buffer directly. std::size_t actual_read_size = input.readsome(unp.buffer(), try_read_size); // tell msgpack::unpacker actual consumed size. unp.buffer_consumed(actual_read_size); msgpack::unpacked result; // Message pack data loop while(unp.next(result)) { msgpack::object obj(result.get()); // Use obj } // All complete msgpack message is proccessed at this point, // then continue to read addtional message. }

msgpack::unpacker::next() returns true if one complete msgpack messege is proccessed. If msgpack message is correct but insufficient, it returns false. However, parsing proccess is proceeded and the context information is preserved in the msgpack::unpacker. It helps leveling the load of parse.

When msgpack message contains binary data, string data, or ext data, they are not copied but referenced from msgpack::object by default. See the following implementation:

inline bool unpacker::default_reference_func(type::object_type type, uint64_t len, void*) { return true; }

You can also customize unpack_reference_func. Even if you use references, you don't need to control buffer's lifetime. The buffers' lifetime is controled by msgpack using msgpack::zone's finalizer_array and msgpack::unpacker's reference counting mechanism.

So, in most cases, the default behavior is enough. If you want to control the peak of memory consumption when receiving msgpack data patterns are predictable, customizing unpack_reference_func might be useful.

You can get a reference information from msgpack::unpacker::next() using the following function:

    bool next(unpacked& result, bool& referenced);

However, mostly you don't need to use that version of next() because referenced memories are managed by unpacker.