Made grpc_byte_buffer_reader able to decompress msgs.

include/grpc/byte_buffer.h

@@ -34,10 +34,17 @@
 #ifndef GRPC_BYTE_BUFFER_H
 #define GRPC_BYTE_BUFFER_H
 
-#include <grpc/grpc.h>
 #include <grpc/support/slice_buffer.h>
 
-typedef enum { GRPC_BB_SLICE_BUFFER } grpc_byte_buffer_type;
+typedef enum {
+  GRPC_BB_SLICE_BUFFER,
+
+  /* Keep the GRPC_BB_COMPRESSED_* entries in the same order as the
+   * grpc_compression_algorithm enum entries. */
+  GRPC_BB_COMPRESSED_NONE, /* for overriding otherwise compressed channels */
+  GRPC_BB_COMPRESSED_DEFLATE,
+  GRPC_BB_COMPRESSED_GZIP
+} grpc_byte_buffer_type;
 
 /* byte buffers are containers for messages passed in from the public api's */
 struct grpc_byte_buffer {
@@ -46,5 +53,55 @@ struct grpc_byte_buffer {
     gpr_slice_buffer slice_buffer;
   } data;
 };
+typedef struct grpc_byte_buffer grpc_byte_buffer;
+
+/** Returns the grpc_compression_algorithm enum value for a
+ * grpc_byte_buffer_type GRPC_BB_COMPRESSED_* value. */
+#define GRPC_COMPRESS_ALGORITHM_FROM_BB_TYPE(bb_type) \
+  (bb_type - GRPC_BB_COMPRESSED_NONE)
+
+/** Returns a byte buffer instance over the given slices (up to \a nslices) of
+ * \a type type.
+ *
+ * Increases the reference count for all \a slices processed.
+ *
+ * The user is responsible for invoking grpc_byte_buffer_destroy on
+ * the returned instance. */
+grpc_byte_buffer *grpc_byte_buffer_typed_create(gpr_slice *slices,
+                                                size_t nslices,
+                                                grpc_byte_buffer_type type);
+/** Convenience method for creating GRPC_BB_SLICE_BUFFER byte buffers. \sa
+ * grpc_byte_buffer_typed_create */
+grpc_byte_buffer *grpc_byte_buffer_create(gpr_slice *slices, size_t nslices);
+
+/** Copies input byte buffer \a bb.
+ *
+ * Increases the reference count of all the source slices. The user is
+ * responsible for calling grpc_byte_buffer_destroy over the returned copy. */
+grpc_byte_buffer *grpc_byte_buffer_copy(grpc_byte_buffer *bb);
+
+/** Returns the size of the given byte buffer, in bytes. */
+size_t grpc_byte_buffer_length(grpc_byte_buffer *bb);
+
+/** Destroys \a byte_buffer deallocating all its memory. */
+void grpc_byte_buffer_destroy(grpc_byte_buffer *byte_buffer);
+
+
+/** Reader for byte buffers. Iterates over slices in the byte buffer */
+struct grpc_byte_buffer_reader;
+typedef struct grpc_byte_buffer_reader grpc_byte_buffer_reader;
+
+/** Initialize \a reader to read over \a buffer */
+void grpc_byte_buffer_reader_init(grpc_byte_buffer_reader *reader,
+                                  grpc_byte_buffer *buffer);
+
+/** Cleanup and destroy \a reader */
+void grpc_byte_buffer_reader_destroy(grpc_byte_buffer_reader *reader);
+
+/** Updates \a slice with the next piece of data from from \a reader and returns
+ * 1. Returns 0 at the end of the stream. Caller is responsible for calling
+ * gpr_slice_unref on the result. */
+int grpc_byte_buffer_reader_next(grpc_byte_buffer_reader *reader,
+                                 gpr_slice *slice);
 
 #endif  /* GRPC_BYTE_BUFFER_H */

include/grpc/byte_buffer_reader.h

@@ -38,7 +38,8 @@
 #include <grpc/byte_buffer.h>
 
 struct grpc_byte_buffer_reader {
-  grpc_byte_buffer *buffer;
+  grpc_byte_buffer *buffer_in;
+  grpc_byte_buffer *buffer_out;
   /* Different current objects correspond to different types of byte buffers */
   union {
     /* Index into a slice buffer's array of slices */

include/grpc/grpc.h

@@ -37,6 +37,7 @@
 #include <grpc/status.h>
 
 #include <stddef.h>
+#include <grpc/byte_buffer.h>
 #include <grpc/support/slice.h>
 #include <grpc/support/time.h>
 
@@ -155,34 +156,6 @@ typedef enum grpc_call_error {
    (start_write/add_metadata). Illegal on invoke/accept. */
 #define GRPC_WRITE_NO_COMPRESS (0x00000002u)
 
-/* A buffer of bytes */
-struct grpc_byte_buffer;
-typedef struct grpc_byte_buffer grpc_byte_buffer;
-
-/* Sample helpers to obtain byte buffers (these will certainly move
-   someplace else) */
-grpc_byte_buffer *grpc_byte_buffer_create(gpr_slice *slices, size_t nslices);
-grpc_byte_buffer *grpc_byte_buffer_copy(grpc_byte_buffer *bb);
-size_t grpc_byte_buffer_length(grpc_byte_buffer *bb);
-void grpc_byte_buffer_destroy(grpc_byte_buffer *byte_buffer);
-
-/* Reader for byte buffers. Iterates over slices in the byte buffer */
-struct grpc_byte_buffer_reader;
-typedef struct grpc_byte_buffer_reader grpc_byte_buffer_reader;
-
-/** Initialize \a reader to read over \a buffer */
-void grpc_byte_buffer_reader_init(grpc_byte_buffer_reader *reader,
-                                  grpc_byte_buffer *buffer);
-
-/** Cleanup and destroy \a reader */
-void grpc_byte_buffer_reader_destroy(grpc_byte_buffer_reader *reader);
-
-/* At the end of the stream, returns 0. Otherwise, returns 1 and sets slice to
-   be the returned slice. Caller is responsible for calling gpr_slice_unref on
-   the result. */
-int grpc_byte_buffer_reader_next(grpc_byte_buffer_reader *reader,
-                                 gpr_slice *slice);
-
 /* A single metadata element */
 typedef struct grpc_metadata {
   const char *key;

src/core/surface/byte_buffer.c

@@ -36,10 +36,16 @@
 #include <grpc/support/log.h>
 
 grpc_byte_buffer *grpc_byte_buffer_create(gpr_slice *slices, size_t nslices) {
+  return grpc_byte_buffer_typed_create(slices, nslices, GRPC_BB_SLICE_BUFFER);
+}
+
+grpc_byte_buffer *grpc_byte_buffer_typed_create(gpr_slice *slices,
+                                               size_t nslices,
+                                               grpc_byte_buffer_type type) {
   size_t i;
   grpc_byte_buffer *bb = malloc(sizeof(grpc_byte_buffer));
 
-  bb->type = GRPC_BB_SLICE_BUFFER;
+  bb->type = type;
   gpr_slice_buffer_init(&bb->data.slice_buffer);
   for (i = 0; i < nslices; i++) {
     gpr_slice_ref(slices[i]);
@@ -49,9 +55,13 @@ grpc_byte_buffer *grpc_byte_buffer_create(gpr_slice *slices, size_t nslices) {
   return bb;
 }
 
+
 grpc_byte_buffer *grpc_byte_buffer_copy(grpc_byte_buffer *bb) {
   switch (bb->type) {
     case GRPC_BB_SLICE_BUFFER:
+    case GRPC_BB_COMPRESSED_NONE:
+    case GRPC_BB_COMPRESSED_DEFLATE:
+    case GRPC_BB_COMPRESSED_GZIP:
       return grpc_byte_buffer_create(bb->data.slice_buffer.slices,
                                      bb->data.slice_buffer.count);
   }
@@ -64,6 +74,9 @@ void grpc_byte_buffer_destroy(grpc_byte_buffer *bb) {
   if (!bb) return;
   switch (bb->type) {
     case GRPC_BB_SLICE_BUFFER:
+    case GRPC_BB_COMPRESSED_NONE:
+    case GRPC_BB_COMPRESSED_DEFLATE:
+    case GRPC_BB_COMPRESSED_GZIP:
       gpr_slice_buffer_destroy(&bb->data.slice_buffer);
       break;
   }
@@ -73,6 +86,9 @@ void grpc_byte_buffer_destroy(grpc_byte_buffer *bb) {
 size_t grpc_byte_buffer_length(grpc_byte_buffer *bb) {
   switch (bb->type) {
     case GRPC_BB_SLICE_BUFFER:
+    case GRPC_BB_COMPRESSED_NONE:
+    case GRPC_BB_COMPRESSED_DEFLATE:
+    case GRPC_BB_COMPRESSED_GZIP:
       return bb->data.slice_buffer.length;
   }
   gpr_log(GPR_ERROR, "should never reach here");

src/core/surface/byte_buffer_reader.c

@@ -39,35 +39,68 @@
 #include <grpc/support/slice_buffer.h>
 #include <grpc/byte_buffer.h>
 
+#include "src/core/compression/algorithm.h"
+#include "src/core/compression/message_compress.h"
+
 void grpc_byte_buffer_reader_init(grpc_byte_buffer_reader *reader,
                                   grpc_byte_buffer *buffer) {
-  reader->buffer = buffer;
+  grpc_compression_algorithm compress_algo;
+  gpr_slice_buffer decompressed_slices_buffer;
+  reader->buffer_in = buffer;
   switch (buffer->type) {
+    case GRPC_BB_COMPRESSED_DEFLATE:
+    case GRPC_BB_COMPRESSED_GZIP:
+      compress_algo =
+          GRPC_COMPRESS_ALGORITHM_FROM_BB_TYPE(reader->buffer_in->type);
+      gpr_slice_buffer_init(&decompressed_slices_buffer);
+      grpc_msg_decompress(compress_algo, &reader->buffer_in->data.slice_buffer,
+                          &decompressed_slices_buffer);
+      /* the output buffer is a regular GRPC_BB_SLICE_BUFFER */
+      reader->buffer_out = grpc_byte_buffer_create(
+          decompressed_slices_buffer.slices,
+          decompressed_slices_buffer.count);
+      gpr_slice_buffer_destroy(&decompressed_slices_buffer);
+    /* fallthrough */
     case GRPC_BB_SLICE_BUFFER:
+    case GRPC_BB_COMPRESSED_NONE:
       reader->current.index = 0;
   }
 }
 
 void grpc_byte_buffer_reader_destroy(grpc_byte_buffer_reader *reader) {
-  /* no-op: the user is responsible for memory deallocation.
-   * Other cleanup operations would go here if needed. */
+  switch (reader->buffer_in->type) {
+    case GRPC_BB_COMPRESSED_DEFLATE:
+    case GRPC_BB_COMPRESSED_GZIP:
+      grpc_byte_buffer_destroy(reader->buffer_out);
+      break;
+    case GRPC_BB_SLICE_BUFFER:
+    case GRPC_BB_COMPRESSED_NONE:
+      ; /* no-op */
+  }
 }
 
 int grpc_byte_buffer_reader_next(grpc_byte_buffer_reader *reader,
                                  gpr_slice *slice) {
-  grpc_byte_buffer *buffer = reader->buffer;
   gpr_slice_buffer *slice_buffer;
-  switch (buffer->type) {
+  grpc_byte_buffer *buffer = NULL;
+
+  /* Pick the right buffer based on the input type */
+  switch (reader->buffer_in->type) {
     case GRPC_BB_SLICE_BUFFER:
-      slice_buffer = &buffer->data.slice_buffer;
-      if (reader->current.index < slice_buffer->count) {
-        *slice = gpr_slice_ref(slice_buffer->slices[reader->current.index]);
-        reader->current.index += 1;
-        return 1;
-      } else {
-        return 0;
-      }
+    case GRPC_BB_COMPRESSED_NONE:
+      buffer = reader->buffer_in;
       break;
+    case GRPC_BB_COMPRESSED_DEFLATE:
+    case GRPC_BB_COMPRESSED_GZIP:
+      buffer = reader->buffer_out;
+      break;
+  }
+  GPR_ASSERT(buffer);
+  slice_buffer = &buffer->data.slice_buffer;
+  if (reader->current.index < slice_buffer->count) {
+    *slice = gpr_slice_ref(slice_buffer->slices[reader->current.index]);
+    reader->current.index += 1;
+    return 1;
   }
   return 0;
 }

src/core/surface/call.c

@@ -806,6 +806,9 @@ static void copy_byte_buffer_to_stream_ops(grpc_byte_buffer *byte_buffer,
 
   switch (byte_buffer->type) {
     case GRPC_BB_SLICE_BUFFER:
+    case GRPC_BB_COMPRESSED_NONE:
+    case GRPC_BB_COMPRESSED_DEFLATE:
+    case GRPC_BB_COMPRESSED_GZIP:
       for (i = 0; i < byte_buffer->data.slice_buffer.count; i++) {
         gpr_slice slice = byte_buffer->data.slice_buffer.slices[i];
         gpr_slice_ref(slice);

test/core/surface/byte_buffer_reader_test.c

@@ -42,6 +42,8 @@
 #include <grpc/support/time.h>
 #include "test/core/util/test_config.h"
 
+#include "src/core/compression/message_compress.h"
+
 #include <string.h>
 
 #define LOG_TEST(x) gpr_log(GPR_INFO, "%s", x)
@@ -89,9 +91,82 @@ static void test_read_one_slice_malloc(void) {
   grpc_byte_buffer_destroy(buffer);
 }
 
+static void test_read_none_compressed_slice(void) {
+  gpr_slice slice;
+  grpc_byte_buffer *buffer;
+  grpc_byte_buffer_reader reader;
+  gpr_slice first_slice, second_slice;
+  int first_code, second_code;
+
+  LOG_TEST("test_read_none_compressed_slice");
+  slice = gpr_slice_from_copied_string("test");
+  buffer = grpc_byte_buffer_typed_create(&slice, 1, GRPC_BB_COMPRESSED_NONE);
+  gpr_slice_unref(slice);
+  grpc_byte_buffer_reader_init(&reader, buffer);
+  first_code = grpc_byte_buffer_reader_next(&reader, &first_slice);
+  GPR_ASSERT(first_code != 0);
+  GPR_ASSERT(memcmp(GPR_SLICE_START_PTR(first_slice), "test", 4) == 0);
+  gpr_slice_unref(first_slice);
+  second_code = grpc_byte_buffer_reader_next(&reader, &second_slice);
+  GPR_ASSERT(second_code == 0);
+  grpc_byte_buffer_destroy(buffer);
+}
+
+static void read_compressed_slice(grpc_compression_algorithm algorithm,
+                                  int input_size) {
+  gpr_slice input_slice;
+  gpr_slice_buffer sliceb_in;
+  gpr_slice_buffer sliceb_out;
+  grpc_byte_buffer *buffer;
+  grpc_byte_buffer_reader reader;
+  gpr_slice read_slice;
+  int read_count = 0;
+
+  gpr_slice_buffer_init(&sliceb_in);
+  gpr_slice_buffer_init(&sliceb_out);
+
+  input_slice = gpr_slice_malloc(input_size);
+  memset(GPR_SLICE_START_PTR(input_slice), 'a', input_size);
+  gpr_slice_buffer_add(&sliceb_in, input_slice);  /* takes ownership */
+  GPR_ASSERT(grpc_msg_compress(algorithm, &sliceb_in, &sliceb_out));
+
+  buffer = grpc_byte_buffer_typed_create(sliceb_out.slices, sliceb_out.count,
+                                         GRPC_BB_COMPRESSED_NONE + algorithm);
+  grpc_byte_buffer_reader_init(&reader, buffer);
+
+  while (grpc_byte_buffer_reader_next(&reader, &read_slice)) {
+    GPR_ASSERT(memcmp(GPR_SLICE_START_PTR(read_slice),
+                      GPR_SLICE_START_PTR(input_slice) + read_count,
+                      GPR_SLICE_LENGTH(read_slice)) == 0);
+    read_count += GPR_SLICE_LENGTH(read_slice);
+    gpr_slice_unref(read_slice);
+  }
+  GPR_ASSERT(read_count == input_size);
+  grpc_byte_buffer_reader_destroy(&reader);
+  grpc_byte_buffer_destroy(buffer);
+  gpr_slice_buffer_destroy(&sliceb_out);
+  gpr_slice_buffer_destroy(&sliceb_in);
+}
+
+static void test_read_gzip_compressed_slice(void) {
+  const int INPUT_SIZE = 2048;
+  LOG_TEST("test_read_gzip_compressed_slice");
+  read_compressed_slice(GRPC_COMPRESS_GZIP, INPUT_SIZE);
+}
+
+static void test_read_deflate_compressed_slice(void) {
+  const int INPUT_SIZE = 2048;
+  LOG_TEST("test_read_deflate_compressed_slice");
+  read_compressed_slice(GRPC_COMPRESS_DEFLATE, INPUT_SIZE);
+}
+
 int main(int argc, char **argv) {
   grpc_test_init(argc, argv);
   test_read_one_slice();
   test_read_one_slice_malloc();
+  test_read_none_compressed_slice();
+  test_read_gzip_compressed_slice();
+  test_read_deflate_compressed_slice();
+
   return 0;
 }