Merge pull request #515 from edsiper/unpacker_size

unpack: new msgpack_unpacker_next_with_size() function

include/msgpack/unpack.h

@@ -146,6 +146,18 @@ static inline void   msgpack_unpacker_buffer_consumed(msgpack_unpacker* mpac, si
 MSGPACK_DLLEXPORT
 msgpack_unpack_return msgpack_unpacker_next(msgpack_unpacker* mpac, msgpack_unpacked* pac);
 
+/**
+ * Deserializes one object and set the number of parsed bytes involved.
+ * Returns true if it successes. Otherwise false is returned.
+ * @param mpac    pointer to an initialized msgpack_unpacker object.
+ * @param result  pointer to an initialized msgpack_unpacked object.
+ * @param p_bytes pointer to variable that will be set with the number of parsed bytes.
+ */
+MSGPACK_DLLEXPORT
+msgpack_unpack_return msgpack_unpacker_next_with_size(msgpack_unpacker* mpac,
+                                                      msgpack_unpacked* result,
+                                                      size_t *p_bytes);
+
 /**
  * Initializes a msgpack_unpacked object.
  * The initialized object must be destroyed by msgpack_unpacked_destroy(msgpack_unpacker*).
@@ -267,4 +279,3 @@ static inline msgpack_zone* msgpack_unpacked_release_zone(msgpack_unpacked* resu
 #endif
 
 #endif /* msgpack/unpack.h */
-

src/unpack.c

@@ -510,7 +510,8 @@ void msgpack_unpacker_reset(msgpack_unpacker* mpac)
     mpac->parsed = 0;
 }
 
-msgpack_unpack_return msgpack_unpacker_next(msgpack_unpacker* mpac, msgpack_unpacked* result)
+static inline msgpack_unpack_return unpacker_next(msgpack_unpacker* mpac,
+                                                  msgpack_unpacked* result)
 {
     int ret;
 
@@ -529,11 +530,40 @@ msgpack_unpack_return msgpack_unpacker_next(msgpack_unpacker* mpac, msgpack_unpa
     }
     result->zone = msgpack_unpacker_release_zone(mpac);
     result->data = msgpack_unpacker_data(mpac);
-    msgpack_unpacker_reset(mpac);
 
     return MSGPACK_UNPACK_SUCCESS;
 }
 
+msgpack_unpack_return msgpack_unpacker_next(msgpack_unpacker* mpac,
+                                            msgpack_unpacked* result)
+{
+    int ret;
+
+    ret = unpacker_next(mpac, result);
+    if (ret == MSGPACK_UNPACK_SUCCESS) {
+        msgpack_unpacker_reset(mpac);
+    }
+
+    return ret;
+}
+
+msgpack_unpack_return
+msgpack_unpacker_next_with_size(msgpack_unpacker* mpac,
+                                msgpack_unpacked* result, size_t *p_bytes)
+{
+    int ret;
+
+    ret = unpacker_next(mpac, result);
+    if (ret == MSGPACK_UNPACK_SUCCESS || ret == MSGPACK_UNPACK_CONTINUE) {
+        *p_bytes = mpac->parsed;
+    }
+
+    if (ret == MSGPACK_UNPACK_SUCCESS) {
+        msgpack_unpacker_reset(mpac);
+    }
+
+    return ret;
+}
 
 msgpack_unpack_return
 msgpack_unpack(const char* data, size_t len, size_t* off,

test/streaming_c.cpp

@@ -120,3 +120,64 @@ TEST(streaming, basic)
     msgpack_unpacked_destroy(&result);
     msgpack_sbuffer_free(buffer);
 }
+
+TEST(streaming, basic_with_size)
+{
+    int ret;
+    size_t bytes;
+    size_t parsed = 0;
+    msgpack_sbuffer* buffer = msgpack_sbuffer_new();
+    msgpack_packer* pk = msgpack_packer_new(buffer, msgpack_sbuffer_write);
+    msgpack_unpacked result;
+    msgpack_unpacker *unp;
+
+    // 1, 2, 3, "str", ["str_data"], "bin", ["bin_data"], {0.3: 0.4}
+    msgpack_pack_int(pk, 1);
+    msgpack_pack_int(pk, 2);
+    msgpack_pack_int(pk, 3);
+    msgpack_pack_str(pk, 3);
+    msgpack_pack_str_body(pk, "str", 3);
+    msgpack_pack_array(pk, 1);
+    msgpack_pack_str(pk, 8);
+    msgpack_pack_str_body(pk, "str_data", 8);
+    msgpack_pack_bin(pk, 3);
+    msgpack_pack_bin_body(pk, "bin", 3);
+    msgpack_pack_array(pk, 1);
+    msgpack_pack_bin(pk, 8);
+    msgpack_pack_bin_body(pk, "bin_data", 8);
+    msgpack_pack_map(pk, 1);
+    msgpack_pack_float(pk, 0.4f);
+    msgpack_pack_double(pk, 0.8);
+    msgpack_packer_free(pk);
+
+    unp = msgpack_unpacker_new(32 * 1024);
+    msgpack_unpacked_init(&result);
+
+    const char* input = buffer->data;
+
+    while (parsed < buffer->size) {
+        memcpy(msgpack_unpacker_buffer(unp), input, 1);
+        msgpack_unpacker_buffer_consumed(unp, 1);
+        input += 1;
+
+        bytes = 0;
+        ret = msgpack_unpacker_next_with_size(unp, &result, &bytes);
+        if (ret == MSGPACK_UNPACK_CONTINUE) {
+            EXPECT_GT(bytes, 0);
+            continue;
+        }
+
+        while (ret == MSGPACK_UNPACK_SUCCESS) {
+            EXPECT_GT(bytes, 0);
+            parsed += bytes;
+            ret = msgpack_unpacker_next_with_size(unp, &result, &bytes);
+        }
+
+    }
+
+    EXPECT_EQ(parsed, buffer->size);
+
+    msgpack_unpacked_destroy(&result);
+    msgpack_unpacker_free(unp);
+    msgpack_sbuffer_free(buffer);
+}