THRIFT-3569 Add thrift_transport_read_all to facilitate large reads in c_glib.

Client: c_glib
Patch: Chandler May

This closes #849

7 files changed, 438 insertions, 37 deletions

diff --git a/lib/c_glib/src/thrift/c_glib/protocol/thrift_binary_protocol.c b/lib/c_glib/src/thrift/c_glib/protocol/thrift_binary_protocol.c
index 358396cd6..48a2c5c62 100644
--- a/lib/c_glib/src/thrift/c_glib/protocol/thrift_binary_protocol.c
+++ b/lib/c_glib/src/thrift/c_glib/protocol/thrift_binary_protocol.c
@@ -645,8 +645,8 @@ thrift_binary_protocol_read_bool (ThriftProtocol *protocol, gboolean *value,
   g_return_val_if_fail (THRIFT_IS_BINARY_PROTOCOL (protocol), -1);
 
   if ((ret = 
-       thrift_transport_read (protocol->transport,
-                              b, 1, error)) < 0)
+       thrift_transport_read_all (protocol->transport,
+                                  b, 1, error)) < 0)
   {
     return -1;
   }
@@ -664,8 +664,8 @@ thrift_binary_protocol_read_byte (ThriftProtocol *protocol, gint8 *value,
   g_return_val_if_fail (THRIFT_IS_BINARY_PROTOCOL (protocol), -1);
 
   if ((ret =
-       thrift_transport_read (protocol->transport,
-                              b, 1, error)) < 0)
+       thrift_transport_read_all (protocol->transport,
+                                  b, 1, error)) < 0)
   {
     return -1;
   }
@@ -687,8 +687,8 @@ thrift_binary_protocol_read_i16 (ThriftProtocol *protocol, gint16 *value,
   g_return_val_if_fail (THRIFT_IS_BINARY_PROTOCOL (protocol), -1);
 
   if ((ret =
-       thrift_transport_read (protocol->transport,
-                              b.byte_array, 2, error)) < 0)
+       thrift_transport_read_all (protocol->transport,
+                                  b.byte_array, 2, error)) < 0)
   {
     return -1;
   }
@@ -710,8 +710,8 @@ thrift_binary_protocol_read_i32 (ThriftProtocol *protocol, gint32 *value,
   g_return_val_if_fail (THRIFT_IS_BINARY_PROTOCOL (protocol), -1);
 
   if ((ret =
-       thrift_transport_read (protocol->transport,
-                              b.byte_array, 4, error)) < 0)
+       thrift_transport_read_all (protocol->transport,
+                                  b.byte_array, 4, error)) < 0)
   {
     return -1;
   }
@@ -733,8 +733,8 @@ thrift_binary_protocol_read_i64 (ThriftProtocol *protocol, gint64 *value,
   g_return_val_if_fail (THRIFT_IS_BINARY_PROTOCOL (protocol), -1);
 
   if ((ret =
-       thrift_transport_read (protocol->transport,
-                              b.byte_array, 8, error)) < 0)
+       thrift_transport_read_all (protocol->transport,
+                                  b.byte_array, 8, error)) < 0)
   {
     return -1;
   }
@@ -756,8 +756,8 @@ thrift_binary_protocol_read_double (ThriftProtocol *protocol,
   g_return_val_if_fail (THRIFT_IS_BINARY_PROTOCOL (protocol), -1);
 
   if ((ret =
-       thrift_transport_read (protocol->transport,
-                              b.byte_array, 8, error)) < 0)
+       thrift_transport_read_all (protocol->transport,
+                                  b.byte_array, 8, error)) < 0)
   {
     return -1;
   }
@@ -790,8 +790,8 @@ thrift_binary_protocol_read_string (ThriftProtocol *protocol,
     len = (guint32) read_len + 1; /* space for null terminator */
     *str = g_new0 (gchar, len);
     if ((ret =
-         thrift_transport_read (protocol->transport,
-                                *str, read_len, error)) < 0)
+         thrift_transport_read_all (protocol->transport,
+                                    *str, read_len, error)) < 0)
     {
       g_free (*str);
       *str = NULL;
@@ -831,8 +831,8 @@ thrift_binary_protocol_read_binary (ThriftProtocol *protocol,
     *len = (guint32) read_len;
     *buf = g_new (guchar, *len);
     if ((ret =
-         thrift_transport_read (protocol->transport,
-                                *buf, *len, error)) < 0)
+         thrift_transport_read_all (protocol->transport,
+                                    *buf, *len, error)) < 0)
     {
       g_free (*buf);
       *buf = NULL;
diff --git a/lib/c_glib/src/thrift/c_glib/protocol/thrift_compact_protocol.c b/lib/c_glib/src/thrift/c_glib/protocol/thrift_compact_protocol.c
index 5ff33b307..87b6b300f 100644
--- a/lib/c_glib/src/thrift/c_glib/protocol/thrift_compact_protocol.c
+++ b/lib/c_glib/src/thrift/c_glib/protocol/thrift_compact_protocol.c
@@ -269,20 +269,22 @@ thrift_compact_protocol_read_varint64 (ThriftCompactProtocol *protocol,
                                        gint64 *i64,
                                        GError **error)
 {
+  ThriftProtocol *tp;
   gint32 ret;
   gint32 xfer;
   guint64 val;
   gint shift;
   guint8 byte;
 
+  tp = THRIFT_PROTOCOL (protocol);
   xfer = 0;
   val = 0;
   shift = 0;
   byte = 0;
 
   while (TRUE) {
-    if ((ret = thrift_transport_read (THRIFT_PROTOCOL (protocol)->transport,
-                                      (gpointer) &byte, 1, error)) < 0) {
+    if ((ret = thrift_transport_read_all (tp->transport,
+                                          (gpointer) &byte, 1, error)) < 0) {
       return -1;
     }
     ++xfer;
@@ -1243,8 +1245,8 @@ thrift_compact_protocol_read_byte (ThriftProtocol *protocol, gint8 *value,
   g_return_val_if_fail (THRIFT_IS_COMPACT_PROTOCOL (protocol), -1);
 
   if ((ret =
-       thrift_transport_read (protocol->transport,
-                              b, 1, error)) < 0) {
+       thrift_transport_read_all (protocol->transport,
+                                  b, 1, error)) < 0) {
     return -1;
   }
   *value = *(gint8 *) b;
@@ -1344,8 +1346,8 @@ thrift_compact_protocol_read_double (ThriftProtocol *protocol,
   g_return_val_if_fail (THRIFT_IS_COMPACT_PROTOCOL (protocol), -1);
 
   if ((ret =
-       thrift_transport_read (protocol->transport,
-                              u.b, 8, error)) < 0) {
+       thrift_transport_read_all (protocol->transport,
+                                  u.b, 8, error)) < 0) {
     return -1;
   }
   u.bits = GUINT64_FROM_LE (u.bits);
@@ -1390,8 +1392,8 @@ thrift_compact_protocol_read_string (ThriftProtocol *protocol,
     /* allocate the memory as an array of unsigned char for binary data */
     *str = g_new0 (gchar, read_len + 1);
     if ((ret =
-         thrift_transport_read (protocol->transport,
-                                *str, read_len, error)) < 0) {
+         thrift_transport_read_all (protocol->transport,
+                                    *str, read_len, error)) < 0) {
       g_free (*str);
       *str = NULL;
       return -1;
@@ -1452,8 +1454,8 @@ thrift_compact_protocol_read_binary (ThriftProtocol *protocol,
     *len = (guint32) read_len;
     *buf = g_new (guchar, *len);
     if ((ret =
-         thrift_transport_read (protocol->transport,
-                                *buf, *len, error)) < 0) {
+         thrift_transport_read_all (protocol->transport,
+                                    *buf, *len, error)) < 0) {
       g_free (*buf);
       *buf = NULL;
       *len = 0;
diff --git a/lib/c_glib/src/thrift/c_glib/transport/thrift_transport.c b/lib/c_glib/src/thrift/c_glib/transport/thrift_transport.c
index 553343792..9dd267143 100644
--- a/lib/c_glib/src/thrift/c_glib/transport/thrift_transport.c
+++ b/lib/c_glib/src/thrift/c_glib/transport/thrift_transport.c
@@ -85,6 +85,14 @@ thrift_transport_flush (ThriftTransport *transport, GError **error)
   return THRIFT_TRANSPORT_GET_CLASS (transport)->flush (transport, error);
 }
 
+gint32
+thrift_transport_read_all (ThriftTransport *transport, gpointer buf,
+                           guint32 len, GError **error)
+{
+  return THRIFT_TRANSPORT_GET_CLASS (transport)->read_all (transport, buf,
+                                                           len, error);
+}
+
 /* by default, peek returns true if and only if the transport is open */
 static gboolean
 thrift_transport_real_peek (ThriftTransport *transport, GError **error)
@@ -94,6 +102,33 @@ thrift_transport_real_peek (ThriftTransport *transport, GError **error)
   return THRIFT_TRANSPORT_GET_CLASS (transport)->is_open (transport);
 }
 
+static gint32
+thrift_transport_real_read_all (ThriftTransport *transport, gpointer buf,
+                                guint32 len, GError **error)
+{
+  ThriftTransportClass *ttc;
+  guint32 have;
+  gint32 ret;
+  gint8 *bytes;
+
+  THRIFT_UNUSED_VAR (error);
+
+  ttc = THRIFT_TRANSPORT_GET_CLASS (transport);
+  have = 0;
+  ret = 0;
+  bytes = (gint8*) buf;
+
+  while (have < len) {
+    if ((ret = ttc->read (transport, (gpointer) (bytes + have), len - have,
+                          error)) < 0) {
+      return ret;
+    }
+    have += ret;
+  }
+
+  return have;
+}
+
 /* define the GError domain for Thrift transports */
 GQuark
 thrift_transport_error_quark (void)
@@ -115,8 +150,9 @@ thrift_transport_class_init (ThriftTransportClass *cls)
   cls->write_end = thrift_transport_write_end;
   cls->flush = thrift_transport_flush;
 
-  /* provide a default implementation for the peek method */
+  /* provide a default implementation for the peek and read_all methods */
   cls->peek = thrift_transport_real_peek;
+  cls->read_all = thrift_transport_real_read_all;
 }
 
 static void
diff --git a/lib/c_glib/src/thrift/c_glib/transport/thrift_transport.h b/lib/c_glib/src/thrift/c_glib/transport/thrift_transport.h
index 5555a5eb4..94bb6f507 100644
--- a/lib/c_glib/src/thrift/c_glib/transport/thrift_transport.h
+++ b/lib/c_glib/src/thrift/c_glib/transport/thrift_transport.h
@@ -73,6 +73,8 @@ struct _ThriftTransportClass
                    const guint32 len, GError **error);
   gboolean (*write_end) (ThriftTransport *transport, GError **error);
   gboolean (*flush) (ThriftTransport *transport, GError **error);
+  gint32 (*read_all) (ThriftTransport *transport, gpointer buf,
+                      guint32 len, GError **error);
 };
 
 /* used by THRIFT_TYPE_TRANSPORT */
@@ -143,6 +145,13 @@ gboolean thrift_transport_write_end (ThriftTransport *transport,
  */
 gboolean thrift_transport_flush (ThriftTransport *transport, GError **error);
 
+/*!
+ * Read len bytes of data into the buffer buf.
+ * \public \memberof ThriftTransportInterface
+ */
+gint32 thrift_transport_read_all (ThriftTransport *transport, gpointer buf,
+                                  guint32 len, GError **error);
+
 /* define error/exception types */
 typedef enum
 {
diff --git a/lib/c_glib/test/Makefile.am b/lib/c_glib/test/Makefile.am
index 41ed4d13f..3dc54180f 100755
--- a/lib/c_glib/test/Makefile.am
+++ b/lib/c_glib/test/Makefile.am
@@ -104,6 +104,7 @@ testbinaryprotocol_SOURCES = testbinaryprotocol.c
 testbinaryprotocol_LDADD = \
     $(top_builddir)/lib/c_glib/src/thrift/c_glib/protocol/libthrift_c_glib_la-thrift_protocol.o \
     $(top_builddir)/lib/c_glib/src/thrift/c_glib/transport/libthrift_c_glib_la-thrift_transport.o \
+    $(top_builddir)/lib/c_glib/src/thrift/c_glib/transport/libthrift_c_glib_la-thrift_framed_transport.o \
     $(top_builddir)/lib/c_glib/src/thrift/c_glib/transport/libthrift_c_glib_la-thrift_socket.o \
     $(top_builddir)/lib/c_glib/src/thrift/c_glib/transport/libthrift_c_glib_la-thrift_server_transport.o \
     $(top_builddir)/lib/c_glib/src/thrift/c_glib/transport/libthrift_c_glib_la-thrift_server_socket.o
@@ -112,6 +113,7 @@ testcompactprotocol_SOURCES = testcompactprotocol.c
 testcompactprotocol_LDADD = \
     $(top_builddir)/lib/c_glib/src/thrift/c_glib/protocol/libthrift_c_glib_la-thrift_protocol.o \
     $(top_builddir)/lib/c_glib/src/thrift/c_glib/transport/libthrift_c_glib_la-thrift_transport.o \
+    $(top_builddir)/lib/c_glib/src/thrift/c_glib/transport/libthrift_c_glib_la-thrift_framed_transport.o \
     $(top_builddir)/lib/c_glib/src/thrift/c_glib/transport/libthrift_c_glib_la-thrift_socket.o \
     $(top_builddir)/lib/c_glib/src/thrift/c_glib/transport/libthrift_c_glib_la-thrift_server_transport.o \
     $(top_builddir)/lib/c_glib/src/thrift/c_glib/transport/libthrift_c_glib_la-thrift_server_socket.o
diff --git a/lib/c_glib/test/testbinaryprotocol.c b/lib/c_glib/test/testbinaryprotocol.c
index 349460a0a..7ca5150ae 100755
--- a/lib/c_glib/test/testbinaryprotocol.c
+++ b/lib/c_glib/test/testbinaryprotocol.c
@@ -36,6 +36,7 @@
 #include <thrift/c_glib/protocol/thrift_protocol.h>
 #include <thrift/c_glib/transport/thrift_socket.h>
 #include <thrift/c_glib/transport/thrift_server_socket.h>
+#include <thrift/c_glib/transport/thrift_framed_transport.h>
 
 #define TEST_BOOL TRUE
 #define TEST_BYTE 123
@@ -50,14 +51,14 @@ static int transport_read_count = 0;
 static int transport_read_error = 0;
 static int transport_read_error_at = -1;
 gint32
-my_thrift_transport_read (ThriftTransport *transport, gpointer buf,
-                          guint32 len, GError **error)
+my_thrift_transport_read_all (ThriftTransport *transport, gpointer buf,
+                              guint32 len, GError **error)
 {
   if (transport_read_count != transport_read_error_at
       && transport_read_error == 0)
   {
     transport_read_count++;
-    return thrift_transport_read (transport, buf, len, error);
+    return thrift_transport_read_all (transport, buf, len, error);
   }
   return -1;
 }
@@ -78,14 +79,15 @@ my_thrift_transport_write (ThriftTransport *transport, const gpointer buf,
   return FALSE;
 }
 
-#define thrift_transport_read my_thrift_transport_read
+#define thrift_transport_read_all my_thrift_transport_read_all
 #define thrift_transport_write my_thrift_transport_write
 #include "../src/thrift/c_glib/protocol/thrift_binary_protocol.c"
-#undef thrift_transport_read
+#undef thrift_transport_read_all
 #undef thrift_transport_write
 
 static void thrift_server_primitives (const int port);
 static void thrift_server_complex_types (const int port);
+static void thrift_server_many_frames (const int port);
 
 static void
 test_create_and_destroy(void)
@@ -389,6 +391,91 @@ test_read_and_write_complex_types (void)
   }
 }
 
+static void
+test_read_and_write_many_frames (void)
+{
+  int status;
+  pid_t pid;
+  ThriftSocket *tsocket = NULL;
+  ThriftTransport *transport = NULL;
+  ThriftFramedTransport *ft = NULL;
+  ThriftBinaryProtocol *tb = NULL;
+  ThriftProtocol *protocol = NULL;
+  gpointer binary = (gpointer *) TEST_STRING;
+  const guint32 len = strlen (TEST_STRING);
+  int port = TEST_PORT;
+
+  /* fork a server from the client */
+  pid = fork ();
+  assert (pid >= 0);
+
+  if (pid == 0)
+  {
+    /* child listens */
+    thrift_server_many_frames (port);
+    exit (0);
+  } else {
+    /* parent.  wait a bit for the socket to be created. */
+    sleep (1);
+
+    /* create a ThriftSocket */
+    tsocket = g_object_new (THRIFT_TYPE_SOCKET, "hostname", "localhost",
+                            "port", port, NULL);
+    assert (tsocket != NULL);
+    transport = THRIFT_TRANSPORT (tsocket);
+
+    /* wrap in a framed transport */
+    ft = g_object_new (THRIFT_TYPE_FRAMED_TRANSPORT, "transport", transport,
+                       "w_buf_size", 1, NULL);
+    assert (ft != NULL);
+    transport = THRIFT_TRANSPORT (ft);
+
+    thrift_transport_open (transport, NULL);
+    assert (thrift_transport_is_open (transport));
+
+    /* create a binary protocol */
+    tb = g_object_new (THRIFT_TYPE_BINARY_PROTOCOL, "transport",
+                       transport, NULL);
+    protocol = THRIFT_PROTOCOL (tb);
+    assert (protocol != NULL);
+
+    /* write a bunch of primitives */
+    assert (thrift_binary_protocol_write_bool (protocol, TEST_BOOL, NULL) > 0);
+    thrift_transport_flush (transport, NULL);
+    assert (thrift_binary_protocol_write_byte (protocol, TEST_BYTE, NULL) > 0);
+    thrift_transport_flush (transport, NULL);
+    assert (thrift_binary_protocol_write_i16 (protocol, TEST_I16, NULL) > 0);
+    thrift_transport_flush (transport, NULL);
+    assert (thrift_binary_protocol_write_i32 (protocol, TEST_I32, NULL) > 0);
+    thrift_transport_flush (transport, NULL);
+    assert (thrift_binary_protocol_write_i64 (protocol, TEST_I64, NULL) > 0);
+    thrift_transport_flush (transport, NULL);
+    assert (thrift_binary_protocol_write_double (protocol,
+                                                 TEST_DOUBLE, NULL) > 0);
+    thrift_transport_flush (transport, NULL);
+    assert (thrift_binary_protocol_write_string (protocol,
+                                                 TEST_STRING, NULL) > 0);
+    thrift_transport_flush (transport, NULL);
+    assert (thrift_binary_protocol_write_binary (protocol, binary,
+                                                 len, NULL) > 0);
+    thrift_transport_flush (transport, NULL);
+    assert (thrift_binary_protocol_write_binary (protocol, NULL, 0, NULL) > 0);
+    thrift_transport_flush (transport, NULL);
+    assert (thrift_binary_protocol_write_binary (protocol, binary,
+                                                 len, NULL) > 0);
+    thrift_transport_flush (transport, NULL);
+
+    /* clean up */
+    thrift_transport_write_end (transport, NULL);
+    thrift_transport_close (transport, NULL);
+    g_object_unref (ft);
+    g_object_unref (tsocket);
+    g_object_unref (tb);
+    assert (wait (&status) == pid);
+    assert (status == 0);
+  }
+}
+
 
 static void
 thrift_server_primitives (const int port)
@@ -666,6 +753,71 @@ thrift_server_complex_types (const int port)
   g_object_unref (tsocket);
 }
 
+static void
+thrift_server_many_frames (const int port)
+{
+  ThriftServerTransport *transport = NULL;
+  ThriftTransport *client = NULL;
+  ThriftBinaryProtocol *tbp = NULL;
+  ThriftProtocol *protocol = NULL;
+  ThriftServerSocket *tsocket = NULL;
+  gboolean value_boolean = FALSE;
+  gint8 value_byte = 0;
+  gint16 value_16 = 0;
+  gint32 value_32 = 0;
+  gint64 value_64 = 0;
+  gdouble value_double = 0;
+  gchar *string = NULL;
+  gpointer binary = NULL;
+  guint32 len = 0;
+  void *comparator = (void *) TEST_STRING;
+
+  tsocket = g_object_new (THRIFT_TYPE_SERVER_SOCKET, "port", port, NULL);
+  transport = THRIFT_SERVER_TRANSPORT (tsocket);
+  thrift_server_transport_listen (transport, NULL);
+
+  /* wrap the client in a framed transport */
+  client = g_object_new (THRIFT_TYPE_FRAMED_TRANSPORT, "transport",
+                         thrift_server_transport_accept (transport, NULL),
+                         "r_buf_size", 1, NULL);
+  assert (client != NULL);
+
+  tbp = g_object_new (THRIFT_TYPE_BINARY_PROTOCOL, "transport",
+                      client, NULL);
+  protocol = THRIFT_PROTOCOL (tbp);
+
+  assert (thrift_binary_protocol_read_bool (protocol,
+                                            &value_boolean, NULL) > 0);
+  assert (thrift_binary_protocol_read_byte (protocol, &value_byte, NULL) > 0);
+  assert (thrift_binary_protocol_read_i16 (protocol, &value_16, NULL) > 0);
+  assert (thrift_binary_protocol_read_i32 (protocol, &value_32, NULL) > 0);
+  assert (thrift_binary_protocol_read_i64 (protocol, &value_64, NULL) > 0);
+  assert (thrift_binary_protocol_read_double (protocol,
+                                              &value_double, NULL) > 0);
+  assert (thrift_binary_protocol_read_string (protocol, &string, NULL) > 0);
+  assert (thrift_binary_protocol_read_binary (protocol, &binary,
+                                              &len, NULL) > 0);
+
+  assert (value_boolean == TEST_BOOL);
+  assert (value_byte == TEST_BYTE);
+  assert (value_16 == TEST_I16);
+  assert (value_32 == TEST_I32);
+  assert (value_64 == TEST_I64);
+  assert (value_double == TEST_DOUBLE);
+  assert (strcmp (TEST_STRING, string) == 0);
+  assert (memcmp (comparator, binary, len) == 0);
+
+  g_free (string);
+  g_free (binary);
+
+  thrift_transport_read_end (client, NULL);
+  thrift_transport_close (client, NULL);
+
+  g_object_unref (tbp);
+  g_object_unref (client);
+  g_object_unref (tsocket);
+}
+
 int
 main(int argc, char *argv[])
 {
@@ -679,6 +831,8 @@ main(int argc, char *argv[])
   g_test_add_func ("/testbinaryprotocol/Initialize", test_initialize);
   g_test_add_func ("/testbinaryprotocol/ReadAndWritePrimitives", test_read_and_write_primitives);
   g_test_add_func ("/testbinaryprotocol/ReadAndWriteComplexTypes", test_read_and_write_complex_types);
+  g_test_add_func ("/testbinaryprotocol/ReadAndWriteManyFrames",
+                   test_read_and_write_many_frames);
 
   return g_test_run ();
 }
diff --git a/lib/c_glib/test/testcompactprotocol.c b/lib/c_glib/test/testcompactprotocol.c
index 3d16dc074..9b57a8cc0 100755
--- a/lib/c_glib/test/testcompactprotocol.c
+++ b/lib/c_glib/test/testcompactprotocol.c
@@ -40,6 +40,7 @@
 #include <thrift/c_glib/protocol/thrift_protocol.h>
 #include <thrift/c_glib/transport/thrift_socket.h>
 #include <thrift/c_glib/transport/thrift_server_socket.h>
+#include <thrift/c_glib/transport/thrift_framed_transport.h>
 
 #define TEST_BOOL TRUE
 #define TEST_BYTE 123
@@ -57,14 +58,14 @@ static int transport_read_count = 0;
 static int transport_read_error = 0;
 static int transport_read_error_at = -1;
 gint32
-my_thrift_transport_read (ThriftTransport *transport, gpointer buf,
-                          guint32 len, GError **error)
+my_thrift_transport_read_all (ThriftTransport *transport, gpointer buf,
+                              guint32 len, GError **error)
 {
   if (transport_read_count != transport_read_error_at
       && transport_read_error == 0)
   {
     transport_read_count++;
-    return thrift_transport_read (transport, buf, len, error);
+    return thrift_transport_read_all (transport, buf, len, error);
   }
   return -1;
 }
@@ -85,14 +86,15 @@ my_thrift_transport_write (ThriftTransport *transport, const gpointer buf,
   return FALSE;
 }
 
-#define thrift_transport_read my_thrift_transport_read
+#define thrift_transport_read_all my_thrift_transport_read_all
 #define thrift_transport_write my_thrift_transport_write
 #include "../src/thrift/c_glib/protocol/thrift_compact_protocol.c"
-#undef thrift_transport_read
+#undef thrift_transport_read_all
 #undef thrift_transport_write
 
 static void thrift_server_primitives (const int port);
 static void thrift_server_complex_types (const int port);
+static void thrift_server_many_frames (const int port);
 
 static void
 test_create_and_destroy (void)
@@ -549,6 +551,112 @@ test_read_and_write_complex_types (void)
   }
 }
 
+static void
+test_read_and_write_many_frames (void)
+{
+  int status;
+  pid_t pid;
+  ThriftSocket *tsocket = NULL;
+  ThriftTransport *transport = NULL;
+  ThriftFramedTransport *ft = NULL;
+  ThriftCompactProtocol *tc = NULL;
+  ThriftProtocol *protocol = NULL;
+  gpointer binary = (gpointer *) TEST_STRING;
+  const guint32 len = strlen (TEST_STRING);
+  int port = TEST_PORT;
+
+  /* fork a server from the client */
+  pid = fork ();
+  assert (pid >= 0);
+
+  if (pid == 0)
+  {
+    /* child listens */
+    thrift_server_many_frames (port);
+    exit (0);
+  } else {
+    /* parent.  wait a bit for the socket to be created. */
+    sleep (1);
+
+    /* create a ThriftSocket */
+    tsocket = g_object_new (THRIFT_TYPE_SOCKET, "hostname", "localhost",
+                            "port", port, NULL);
+    assert (tsocket != NULL);
+    transport = THRIFT_TRANSPORT (tsocket);
+
+    /* wrap in a framed transport */
+    ft = g_object_new (THRIFT_TYPE_FRAMED_TRANSPORT, "transport", transport,
+                       "w_buf_size", 1, NULL);
+    assert (ft != NULL);
+    transport = THRIFT_TRANSPORT (ft);
+
+    thrift_transport_open (transport, NULL);
+    assert (thrift_transport_is_open (transport));
+
+    /* create a compact protocol */
+    tc = g_object_new (THRIFT_TYPE_COMPACT_PROTOCOL, "transport",
+                       transport, NULL);
+    protocol = THRIFT_PROTOCOL (tc);
+    assert (protocol != NULL);
+
+    /* write a bunch of primitives */
+    assert (thrift_compact_protocol_write_bool (protocol, TEST_BOOL,
+                                                NULL) > 0);
+    thrift_transport_flush (transport, NULL);
+    assert (thrift_compact_protocol_write_byte (protocol, TEST_BYTE,
+                                                NULL) > 0);
+    thrift_transport_flush (transport, NULL);
+    assert (thrift_compact_protocol_write_i16 (protocol, TEST_I16, NULL) > 0);
+    thrift_transport_flush (transport, NULL);
+    assert (thrift_compact_protocol_write_i32 (protocol, TEST_I32, NULL) > 0);
+    thrift_transport_flush (transport, NULL);
+    assert (thrift_compact_protocol_write_i64 (protocol, TEST_I64, NULL) > 0);
+    thrift_transport_flush (transport, NULL);
+    assert (thrift_compact_protocol_write_i16 (protocol, TEST_NI16, NULL) > 0);
+    thrift_transport_flush (transport, NULL);
+    assert (thrift_compact_protocol_write_i32 (protocol, TEST_NI32, NULL) > 0);
+    thrift_transport_flush (transport, NULL);
+    assert (thrift_compact_protocol_write_i64 (protocol, TEST_NI64, NULL) > 0);
+    thrift_transport_flush (transport, NULL);
+    assert (thrift_compact_protocol_write_i16 (protocol, 2, NULL) > 0);
+    thrift_transport_flush (transport, NULL);
+    assert (thrift_compact_protocol_write_i32 (protocol, 2, NULL) > 0);
+    thrift_transport_flush (transport, NULL);
+    assert (thrift_compact_protocol_write_i64 (protocol, 2, NULL) > 0);
+    thrift_transport_flush (transport, NULL);
+    assert (thrift_compact_protocol_write_i16 (protocol, -2, NULL) > 0);
+    thrift_transport_flush (transport, NULL);
+    assert (thrift_compact_protocol_write_i32 (protocol, -2, NULL) > 0);
+    thrift_transport_flush (transport, NULL);
+    assert (thrift_compact_protocol_write_i64 (protocol, -2, NULL) > 0);
+    thrift_transport_flush (transport, NULL);
+    assert (thrift_compact_protocol_write_double (protocol,
+                                                 TEST_DOUBLE, NULL) > 0);
+    thrift_transport_flush (transport, NULL);
+    assert (thrift_compact_protocol_write_string (protocol,
+                                                 TEST_STRING, NULL) > 0);
+    thrift_transport_flush (transport, NULL);
+    assert (thrift_compact_protocol_write_binary (protocol, binary,
+                                                 len, NULL) > 0);
+    thrift_transport_flush (transport, NULL);
+    assert (thrift_compact_protocol_write_binary (protocol, NULL,
+                                                  0, NULL) > 0);
+    thrift_transport_flush (transport, NULL);
+    assert (thrift_compact_protocol_write_binary (protocol, binary,
+                                                 len, NULL) > 0);
+    thrift_transport_flush (transport, NULL);
+
+    /* clean up */
+    thrift_transport_write_end (transport, NULL);
+    thrift_transport_close (transport, NULL);
+    g_object_unref (ft);
+    g_object_unref (tsocket);
+    g_object_unref (tc);
+    assert (wait (&status) == pid);
+    assert (status == 0);
+  }
+}
+
 
 static void
 thrift_server_primitives (const int port)
@@ -1053,6 +1161,94 @@ thrift_server_complex_types (const int port)
   g_object_unref (tsocket);
 }
 
+static void
+thrift_server_many_frames (const int port)
+{
+  ThriftServerTransport *transport = NULL;
+  ThriftTransport *client = NULL;
+  ThriftCompactProtocol *tcp = NULL;
+  ThriftProtocol *protocol = NULL;
+  ThriftServerSocket *tsocket = NULL;
+  gboolean value_boolean = FALSE;
+  gint8 value_byte = 0,
+    zigzag_p16 = 0, zigzag_p32 = 0, zigzag_p64 = 0,
+    zigzag_n16 = 0, zigzag_n32 = 0, zigzag_n64 = 0;
+  gint16 value_16 = 0;
+  gint32 value_32 = 0;
+  gint64 value_64 = 0;
+  gint16 value_n16 = 0;
+  gint32 value_n32 = 0;
+  gint64 value_n64 = 0;
+  gdouble value_double = 0;
+  gchar *string = NULL;
+  gpointer binary = NULL;
+  guint32 len = 0;
+  void *comparator = (void *) TEST_STRING;
+
+  tsocket = g_object_new (THRIFT_TYPE_SERVER_SOCKET, "port", port, NULL);
+  transport = THRIFT_SERVER_TRANSPORT (tsocket);
+  thrift_server_transport_listen (transport, NULL);
+
+  /* wrap the client in a framed transport */
+  client = g_object_new (THRIFT_TYPE_FRAMED_TRANSPORT, "transport",
+                         thrift_server_transport_accept (transport, NULL),
+                         "r_buf_size", 1, NULL);
+  assert (client != NULL);
+
+  tcp = g_object_new (THRIFT_TYPE_COMPACT_PROTOCOL, "transport",
+                      client, NULL);
+  protocol = THRIFT_PROTOCOL (tcp);
+
+  assert (thrift_compact_protocol_read_bool (protocol,
+                                            &value_boolean, NULL) > 0);
+  assert (thrift_compact_protocol_read_byte (protocol, &value_byte, NULL) > 0);
+  assert (thrift_compact_protocol_read_i16 (protocol, &value_16, NULL) > 0);
+  assert (thrift_compact_protocol_read_i32 (protocol, &value_32, NULL) > 0);
+  assert (thrift_compact_protocol_read_i64 (protocol, &value_64, NULL) > 0);
+  assert (thrift_compact_protocol_read_i16 (protocol, &value_n16, NULL) > 0);
+  assert (thrift_compact_protocol_read_i32 (protocol, &value_n32, NULL) > 0);
+  assert (thrift_compact_protocol_read_i64 (protocol, &value_n64, NULL) > 0);
+  assert (thrift_compact_protocol_read_byte (protocol, &zigzag_p16, NULL) > 0);
+  assert (thrift_compact_protocol_read_byte (protocol, &zigzag_p32, NULL) > 0);
+  assert (thrift_compact_protocol_read_byte (protocol, &zigzag_p64, NULL) > 0);
+  assert (thrift_compact_protocol_read_byte (protocol, &zigzag_n16, NULL) > 0);
+  assert (thrift_compact_protocol_read_byte (protocol, &zigzag_n32, NULL) > 0);
+  assert (thrift_compact_protocol_read_byte (protocol, &zigzag_n64, NULL) > 0);
+  assert (thrift_compact_protocol_read_double (protocol,
+                                              &value_double, NULL) > 0);
+  assert (thrift_compact_protocol_read_string (protocol, &string, NULL) > 0);
+  assert (thrift_compact_protocol_read_binary (protocol, &binary,
+                                              &len, NULL) > 0);
+
+  assert (value_boolean == TEST_BOOL);
+  assert (value_byte == TEST_BYTE);
+  assert (value_16 == TEST_I16);
+  assert (value_32 == TEST_I32);
+  assert (value_64 == TEST_I64);
+  assert (value_n16 == TEST_NI16);
+  assert (value_n32 == TEST_NI32);
+  assert (value_n64 == TEST_NI64);
+  assert (zigzag_p16 == 4);
+  assert (zigzag_p32 == 4);
+  assert (zigzag_p64 == 4);
+  assert (zigzag_n16 == 3);
+  assert (zigzag_n32 == 3);
+  assert (zigzag_n64 == 3);
+  assert (value_double == TEST_DOUBLE);
+  assert (strcmp (TEST_STRING, string) == 0);
+  assert (memcmp (comparator, binary, len) == 0);
+
+  g_free (string);
+  g_free (binary);
+
+  thrift_transport_read_end (client, NULL);
+  thrift_transport_close (client, NULL);
+
+  g_object_unref (tcp);
+  g_object_unref (client);
+  g_object_unref (tsocket);
+}
+
 int
 main (int argc, char *argv[])
 {
@@ -1069,6 +1265,8 @@ main (int argc, char *argv[])
                    test_read_and_write_primitives);
   g_test_add_func ("/testcompactprotocol/ReadAndWriteComplexTypes",
                    test_read_and_write_complex_types);
+  g_test_add_func ("/testcompactprotocol/ReadAndWriteManyFrames",
+                   test_read_and_write_many_frames);
 
   return g_test_run ();
 }