Merge pull request #1963 from dgquintas/compression

Made grpc_byte_buffer_reader able to decompress msgs.

BUILD

@@ -347,6 +347,7 @@ cc_library(
     "include/grpc/grpc_security.h",
     "include/grpc/byte_buffer.h",
     "include/grpc/byte_buffer_reader.h",
+    "include/grpc/compression.h",
     "include/grpc/grpc.h",
     "include/grpc/status.h",
     "include/grpc/census.h",
@@ -545,6 +546,7 @@ cc_library(
   hdrs = [
     "include/grpc/byte_buffer.h",
     "include/grpc/byte_buffer_reader.h",
+    "include/grpc/compression.h",
     "include/grpc/grpc.h",
     "include/grpc/status.h",
     "include/grpc/census.h",

Makefile

@@ -3044,6 +3044,7 @@ PUBLIC_HEADERS_C += \
     include/grpc/grpc_security.h \
     include/grpc/byte_buffer.h \
     include/grpc/byte_buffer_reader.h \
+    include/grpc/compression.h \
     include/grpc/grpc.h \
     include/grpc/status.h \
     include/grpc/census.h \
@@ -3286,6 +3287,7 @@ LIBGRPC_UNSECURE_SRC = \
 PUBLIC_HEADERS_C += \
     include/grpc/byte_buffer.h \
     include/grpc/byte_buffer_reader.h \
+    include/grpc/compression.h \
     include/grpc/grpc.h \
     include/grpc/status.h \
     include/grpc/census.h \

build.json

@@ -101,6 +101,7 @@
       "public_headers": [
         "include/grpc/byte_buffer.h",
         "include/grpc/byte_buffer_reader.h",
+        "include/grpc/compression.h",
         "include/grpc/grpc.h",
         "include/grpc/status.h"
       ],

include/grpc++/byte_buffer.h

@@ -35,6 +35,7 @@
 #define GRPCXX_BYTE_BUFFER_H
 
 #include <grpc/grpc.h>
+#include <grpc/byte_buffer.h>
 #include <grpc/support/log.h>
 #include <grpc++/config.h>
 #include <grpc++/slice.h>

include/grpc/byte_buffer.h

@@ -34,22 +34,74 @@
 #ifndef GRPC_BYTE_BUFFER_H
 #define GRPC_BYTE_BUFFER_H
 
-#include <grpc/grpc.h>
+#include <grpc/compression.h>
 #include <grpc/support/slice_buffer.h>
 
 #ifdef __cplusplus
 extern "C" {
 #endif
 
-typedef enum { GRPC_BB_SLICE_BUFFER } grpc_byte_buffer_type;
+typedef enum {
+  GRPC_BB_RAW
+  /* Future types may include GRPC_BB_PROTOBUF, etc. */
+} grpc_byte_buffer_type;
 
-/* byte buffers are containers for messages passed in from the public api's */
 struct grpc_byte_buffer {
   grpc_byte_buffer_type type;
   union {
-    gpr_slice_buffer slice_buffer;
+    struct {
+      grpc_compression_algorithm compression;
+      gpr_slice_buffer slice_buffer;
+    } raw;
   } data;
 };
+typedef struct grpc_byte_buffer grpc_byte_buffer;
+
+/** Returns a RAW byte buffer instance over the given slices (up to \a nslices).
+ *
+ * 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_raw_byte_buffer_create(gpr_slice *slices,
+                                              size_t nslices);
+
+/** Returns a *compressed* RAW byte buffer instance over the given slices (up to
+ * \a nslices). The \a compression argument defines the compression algorithm
+ * used to generate the data in \a slices.
+ *
+ * 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_raw_compressed_byte_buffer_create(
+    gpr_slice *slices, size_t nslices, grpc_compression_algorithm compression);
+
+/** 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);
 
 #ifdef __cplusplus
 }

include/grpc/byte_buffer_reader.h

@@ -42,7 +42,8 @@ extern "C" {
 #endif
 
 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/compression.h

@@ -31,8 +31,8 @@
  *
  */
 
-#ifndef GRPC_INTERNAL_CORE_COMPRESSION_ALGORITHM_H
-#define GRPC_INTERNAL_CORE_COMPRESSION_ALGORITHM_H
+#ifndef GRPC_COMPRESSION_H
+#define GRPC_COMPRESSION_H
 
 /* The various compression algorithms supported by GRPC */
 typedef enum {
@@ -46,4 +46,4 @@ typedef enum {
 const char *grpc_compression_algorithm_name(
     grpc_compression_algorithm algorithm);
 
-#endif  /* GRPC_INTERNAL_CORE_COMPRESSION_ALGORITHM_H */
+#endif  /* GRPC_COMPRESSION_H */

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/compression/algorithm.c

@@ -31,7 +31,7 @@
  *
  */
 
-#include "src/core/compression/algorithm.h"
+#include <grpc/compression.h>
 
 const char *grpc_compression_algorithm_name(
     grpc_compression_algorithm algorithm) {

src/core/compression/message_compress.h

@@ -34,7 +34,7 @@
 #ifndef GRPC_INTERNAL_CORE_COMPRESSION_MESSAGE_COMPRESS_H
 #define GRPC_INTERNAL_CORE_COMPRESSION_MESSAGE_COMPRESS_H
 
-#include "src/core/compression/algorithm.h"
+#include <grpc/compression.h>
 #include <grpc/support/slice_buffer.h>
 
 /* compress 'input' to 'output' using 'algorithm'.

src/core/surface/byte_buffer.c

@@ -35,25 +35,31 @@
 #include <grpc/support/alloc.h>
 #include <grpc/support/log.h>
 
-grpc_byte_buffer *grpc_byte_buffer_create(gpr_slice *slices, size_t nslices) {
+grpc_byte_buffer *grpc_raw_byte_buffer_create(gpr_slice *slices,
+                                              size_t nslices) {
+  return grpc_raw_compressed_byte_buffer_create(slices, nslices,
+                                                GRPC_COMPRESS_NONE);
+}
+
+grpc_byte_buffer *grpc_raw_compressed_byte_buffer_create(
+    gpr_slice *slices, size_t nslices, grpc_compression_algorithm compression) {
   size_t i;
   grpc_byte_buffer *bb = malloc(sizeof(grpc_byte_buffer));
-
-  bb->type = GRPC_BB_SLICE_BUFFER;
-  gpr_slice_buffer_init(&bb->data.slice_buffer);
+  bb->type = GRPC_BB_RAW;
+  bb->data.raw.compression = compression;
+  gpr_slice_buffer_init(&bb->data.raw.slice_buffer);
   for (i = 0; i < nslices; i++) {
     gpr_slice_ref(slices[i]);
-    gpr_slice_buffer_add(&bb->data.slice_buffer, slices[i]);
+    gpr_slice_buffer_add(&bb->data.raw.slice_buffer, slices[i]);
   }
-
   return bb;
 }
 
 grpc_byte_buffer *grpc_byte_buffer_copy(grpc_byte_buffer *bb) {
   switch (bb->type) {
-    case GRPC_BB_SLICE_BUFFER:
-      return grpc_byte_buffer_create(bb->data.slice_buffer.slices,
-                                     bb->data.slice_buffer.count);
+    case GRPC_BB_RAW:
+      return grpc_raw_byte_buffer_create(bb->data.raw.slice_buffer.slices,
+                                         bb->data.raw.slice_buffer.count);
   }
   gpr_log(GPR_INFO, "should never get here");
   abort();
@@ -63,8 +69,8 @@ grpc_byte_buffer *grpc_byte_buffer_copy(grpc_byte_buffer *bb) {
 void grpc_byte_buffer_destroy(grpc_byte_buffer *bb) {
   if (!bb) return;
   switch (bb->type) {
-    case GRPC_BB_SLICE_BUFFER:
-      gpr_slice_buffer_destroy(&bb->data.slice_buffer);
+    case GRPC_BB_RAW:
+      gpr_slice_buffer_destroy(&bb->data.raw.slice_buffer);
       break;
   }
   free(bb);
@@ -72,8 +78,8 @@ 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:
-      return bb->data.slice_buffer.length;
+    case GRPC_BB_RAW:
+      return bb->data.raw.slice_buffer.length;
   }
   gpr_log(GPR_ERROR, "should never reach here");
   abort();

src/core/surface/byte_buffer_reader.c

@@ -33,41 +33,73 @@
 
 #include <grpc/byte_buffer_reader.h>
 
+#include <grpc/compression.h>
 #include <grpc/grpc.h>
 #include <grpc/support/alloc.h>
 #include <grpc/support/log.h>
 #include <grpc/support/slice_buffer.h>
 #include <grpc/byte_buffer.h>
 
+#include "src/core/compression/message_compress.h"
+
+static int is_compressed(grpc_byte_buffer *buffer) {
+  switch (buffer->type) {
+    case GRPC_BB_RAW:
+      if (buffer->data.raw.compression == GRPC_COMPRESS_NONE) {
+        return 0 /* GPR_FALSE */;
+      }
+      break;
+  }
+  return 1 /* GPR_TRUE */;
+}
+
 void grpc_byte_buffer_reader_init(grpc_byte_buffer_reader *reader,
                                   grpc_byte_buffer *buffer) {
-  reader->buffer = buffer;
-  switch (buffer->type) {
-    case GRPC_BB_SLICE_BUFFER:
+  gpr_slice_buffer decompressed_slices_buffer;
+  reader->buffer_in = buffer;
+  switch (reader->buffer_in->type) {
+    case GRPC_BB_RAW:
+      gpr_slice_buffer_init(&decompressed_slices_buffer);
+      if (is_compressed(reader->buffer_in)) {
+        grpc_msg_decompress(reader->buffer_in->data.raw.compression,
+                            &reader->buffer_in->data.raw.slice_buffer,
+                            &decompressed_slices_buffer);
+        reader->buffer_out = grpc_raw_byte_buffer_create(
+            decompressed_slices_buffer.slices,
+            decompressed_slices_buffer.count);
+        gpr_slice_buffer_destroy(&decompressed_slices_buffer);
+      } else {  /* not compressed, use the input buffer as output */
+        reader->buffer_out = reader->buffer_in;
+      }
       reader->current.index = 0;
+      break;
   }
 }
 
 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_RAW:
+      /* keeping the same if-else structure as in the init function */
+      if (is_compressed(reader->buffer_in)) {
+        grpc_byte_buffer_destroy(reader->buffer_out);
+      }
+      break;
+  }
 }
 
 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) {
-    case GRPC_BB_SLICE_BUFFER:
-      slice_buffer = &buffer->data.slice_buffer;
+  switch (reader->buffer_in->type) {
+    case GRPC_BB_RAW: {
+      gpr_slice_buffer *slice_buffer;
+      slice_buffer = &reader->buffer_out->data.raw.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;
       }
       break;
+    }
   }
   return 0;
 }

src/core/surface/call.c

@@ -636,7 +636,7 @@ static void call_on_done_send(void *pc, int success) {
 
 static void finish_message(grpc_call *call) {
   /* TODO(ctiller): this could be a lot faster if coded directly */
-  grpc_byte_buffer *byte_buffer = grpc_byte_buffer_create(
+  grpc_byte_buffer *byte_buffer = grpc_raw_byte_buffer_create(
       call->incoming_message.slices, call->incoming_message.count);
   gpr_slice_buffer_reset_and_unref(&call->incoming_message);
 
@@ -806,9 +806,9 @@ static void copy_byte_buffer_to_stream_ops(grpc_byte_buffer *byte_buffer,
   size_t i;
 
   switch (byte_buffer->type) {
-    case GRPC_BB_SLICE_BUFFER:
-      for (i = 0; i < byte_buffer->data.slice_buffer.count; i++) {
-        gpr_slice slice = byte_buffer->data.slice_buffer.slices[i];
+    case GRPC_BB_RAW:
+      for (i = 0; i < byte_buffer->data.raw.slice_buffer.count; i++) {
+        gpr_slice slice = byte_buffer->data.raw.slice_buffer.slices[i];
         gpr_slice_ref(slice);
         grpc_sopb_add_slice(sopb, slice);
       }

src/cpp/proto/proto_utils.cc

@@ -49,8 +49,8 @@ class GrpcBufferWriter GRPC_FINAL
   explicit GrpcBufferWriter(grpc_byte_buffer** bp,
                             int block_size = kMaxBufferLength)
       : block_size_(block_size), byte_count_(0), have_backup_(false) {
-    *bp = grpc_byte_buffer_create(NULL, 0);
-    slice_buffer_ = &(*bp)->data.slice_buffer;
+    *bp = grpc_raw_byte_buffer_create(NULL, 0);
+    slice_buffer_ = &(*bp)->data.raw.slice_buffer;
   }
 
   ~GrpcBufferWriter() GRPC_OVERRIDE {

src/cpp/util/byte_buffer.cc

@@ -42,7 +42,7 @@ ByteBuffer::ByteBuffer(Slice* slices, size_t nslices) {
   for (size_t i = 0; i < nslices; i++) {
     c_slices[i] = slices[i].slice_;
   }
-  buffer_ = grpc_byte_buffer_create(c_slices.data(), nslices);
+  buffer_ = grpc_raw_byte_buffer_create(c_slices.data(), nslices);
 }
 
 void ByteBuffer::Clear() {

src/csharp/ext/grpc_csharp_ext.c

@@ -60,7 +60,7 @@
 
 grpc_byte_buffer *string_to_byte_buffer(const char *buffer, size_t len) {
   gpr_slice slice = gpr_slice_from_copied_buffer(buffer, len);
-  grpc_byte_buffer *bb = grpc_byte_buffer_create(&slice, 1);
+  grpc_byte_buffer *bb = grpc_raw_byte_buffer_create(&slice, 1);
   gpr_slice_unref(slice);
   return bb;
 }

src/node/ext/byte_buffer.cc

@@ -57,7 +57,7 @@ grpc_byte_buffer *BufferToByteBuffer(Handle<Value> buffer) {
   char *data = ::node::Buffer::Data(buffer);
   gpr_slice slice = gpr_slice_malloc(length);
   memcpy(GPR_SLICE_START_PTR(slice), data, length);
-  grpc_byte_buffer *byte_buffer(grpc_byte_buffer_create(&slice, 1));
+  grpc_byte_buffer *byte_buffer(grpc_raw_byte_buffer_create(&slice, 1));
   gpr_slice_unref(slice);
   return byte_buffer;
 }

src/objective-c/GRPCClient/private/NSData+GRPC.m

@@ -55,7 +55,7 @@ static void CopyByteBufferToCharArray(grpc_byte_buffer *buffer, char *array) {
 static grpc_byte_buffer *CopyCharArrayToNewByteBuffer(const char *array,
                                                       size_t length) {
   gpr_slice slice = gpr_slice_from_copied_buffer(array, length);
-  grpc_byte_buffer *buffer = grpc_byte_buffer_create(&slice, 1);
+  grpc_byte_buffer *buffer = grpc_raw_byte_buffer_create(&slice, 1);
   gpr_slice_unref(slice);
   return buffer;
 }
@@ -85,7 +85,7 @@ static grpc_byte_buffer *CopyCharArrayToNewByteBuffer(const char *array,
   // The following implementation is thus not optimal, sometimes requiring two
   // copies (one by self.bytes and another by gpr_slice_from_copied_buffer).
   // If it turns out to be an issue, we can use enumerateByteRangesUsingblock:
-  // to create an array of gpr_slice objects to pass to grpc_byte_buffer_create.
+  // to create an array of gpr_slice objects to pass to grpc_raw_byte_buffer_create.
   // That would make it do exactly one copy, always.
   return CopyCharArrayToNewByteBuffer((const char *)self.bytes, (size_t)self.length);
 }

src/php/ext/grpc/byte_buffer.c

@@ -51,7 +51,7 @@
 
 grpc_byte_buffer *string_to_byte_buffer(char *string, size_t length) {
   gpr_slice slice = gpr_slice_from_copied_buffer(string, length);
-  grpc_byte_buffer *buffer = grpc_byte_buffer_create(&slice, 1);
+  grpc_byte_buffer *buffer = grpc_raw_byte_buffer_create(&slice, 1);
   gpr_slice_unref(slice);
   return buffer;
 }

src/python/src/grpc/_adapter/_c/utility.c

@@ -179,7 +179,7 @@ int pygrpc_produce_op(PyObject *op, grpc_op *result) {
     PyString_AsStringAndSize(
         PyTuple_GET_ITEM(op, MESSAGE_INDEX), &message, &message_size);
     message_slice = gpr_slice_from_copied_buffer(message, message_size);
-    c_op.data.send_message = grpc_byte_buffer_create(&message_slice, 1);
+    c_op.data.send_message = grpc_raw_byte_buffer_create(&message_slice, 1);
     gpr_slice_unref(message_slice);
     break;
   case GRPC_OP_SEND_CLOSE_FROM_CLIENT:

src/ruby/ext/grpc/rb_byte_buffer.c

@@ -42,7 +42,7 @@
 
 grpc_byte_buffer* grpc_rb_s_to_byte_buffer(char *string, size_t length) {
   gpr_slice slice = gpr_slice_from_copied_buffer(string, length);
-  grpc_byte_buffer *buffer = grpc_byte_buffer_create(&slice, 1);
+  grpc_byte_buffer *buffer = grpc_raw_byte_buffer_create(&slice, 1);
   gpr_slice_unref(slice);
   return buffer;
 }

test/core/end2end/cq_verifier.c

@@ -133,7 +133,8 @@ static int byte_buffer_eq_slice(grpc_byte_buffer *bb, gpr_slice b) {
 
   if (!bb) return 0;
 
-  a = merge_slices(bb->data.slice_buffer.slices, bb->data.slice_buffer.count);
+  a = merge_slices(bb->data.raw.slice_buffer.slices,
+                   bb->data.raw.slice_buffer.count);
   ok = GPR_SLICE_LENGTH(a) == GPR_SLICE_LENGTH(b) &&
        0 == memcmp(GPR_SLICE_START_PTR(a), GPR_SLICE_START_PTR(b),
                    GPR_SLICE_LENGTH(a));

test/core/end2end/tests/cancel_after_accept.c

@@ -121,9 +121,9 @@ static void test_cancel_after_accept(grpc_end2end_test_config config,
   gpr_slice request_payload_slice = gpr_slice_from_copied_string("hello world");
   gpr_slice response_payload_slice = gpr_slice_from_copied_string("hello you");
   grpc_byte_buffer *request_payload =
-      grpc_byte_buffer_create(&request_payload_slice, 1);
+      grpc_raw_byte_buffer_create(&request_payload_slice, 1);
   grpc_byte_buffer *response_payload =
-      grpc_byte_buffer_create(&response_payload_slice, 1);
+      grpc_raw_byte_buffer_create(&response_payload_slice, 1);
   int was_cancelled = 2;
 
   c = grpc_channel_create_call(f.client, f.client_cq, "/foo",

test/core/end2end/tests/cancel_after_accept_and_writes_closed.c

@@ -121,9 +121,9 @@ static void test_cancel_after_accept_and_writes_closed(
   gpr_slice request_payload_slice = gpr_slice_from_copied_string("hello world");
   gpr_slice response_payload_slice = gpr_slice_from_copied_string("hello you");
   grpc_byte_buffer *request_payload =
-      grpc_byte_buffer_create(&request_payload_slice, 1);
+      grpc_raw_byte_buffer_create(&request_payload_slice, 1);
   grpc_byte_buffer *response_payload =
-      grpc_byte_buffer_create(&response_payload_slice, 1);
+      grpc_raw_byte_buffer_create(&response_payload_slice, 1);
   int was_cancelled = 2;
 
   c = grpc_channel_create_call(f.client, f.client_cq, "/foo",

test/core/end2end/tests/cancel_after_invoke.c

@@ -119,7 +119,7 @@ static void test_cancel_after_invoke(grpc_end2end_test_config config,
   grpc_byte_buffer *response_payload_recv = NULL;
   gpr_slice request_payload_slice = gpr_slice_from_copied_string("hello world");
   grpc_byte_buffer *request_payload =
-      grpc_byte_buffer_create(&request_payload_slice, 1);
+      grpc_raw_byte_buffer_create(&request_payload_slice, 1);
 
   c = grpc_channel_create_call(f.client, f.client_cq, "/foo",
                                "foo.test.google.fr", deadline);

test/core/end2end/tests/cancel_before_invoke.c

@@ -116,7 +116,7 @@ static void test_cancel_before_invoke(grpc_end2end_test_config config,
   grpc_byte_buffer *response_payload_recv = NULL;
   gpr_slice request_payload_slice = gpr_slice_from_copied_string("hello world");
   grpc_byte_buffer *request_payload =
-      grpc_byte_buffer_create(&request_payload_slice, 1);
+      grpc_raw_byte_buffer_create(&request_payload_slice, 1);
 
   c = grpc_channel_create_call(f.client, f.client_cq, "/foo",
                                "foo.test.google.fr", deadline);

test/core/end2end/tests/invoke_large_request.c

@@ -109,9 +109,9 @@ static void test_invoke_large_request(grpc_end2end_test_config config) {
   grpc_call *c;
   grpc_call *s;
   grpc_byte_buffer *request_payload =
-      grpc_byte_buffer_create(&request_payload_slice, 1);
+      grpc_raw_byte_buffer_create(&request_payload_slice, 1);
   grpc_byte_buffer *response_payload =
-      grpc_byte_buffer_create(&response_payload_slice, 1);
+      grpc_raw_byte_buffer_create(&response_payload_slice, 1);
   gpr_timespec deadline = n_seconds_time(30);
   cq_verifier *v_client = cq_verifier_create(f.client_cq);
   cq_verifier *v_server = cq_verifier_create(f.server_cq);

test/core/end2end/tests/max_message_length.c

@@ -109,7 +109,7 @@ static void test_max_message_length(grpc_end2end_test_config config) {
   grpc_op *op;
   gpr_slice request_payload_slice = gpr_slice_from_copied_string("hello world");
   grpc_byte_buffer *request_payload =
-      grpc_byte_buffer_create(&request_payload_slice, 1);
+      grpc_raw_byte_buffer_create(&request_payload_slice, 1);
   grpc_metadata_array initial_metadata_recv;
   grpc_metadata_array trailing_metadata_recv;
   grpc_metadata_array request_metadata_recv;

test/core/end2end/tests/ping_pong_streaming.c

@@ -165,8 +165,8 @@ static void test_pingpong_streaming(grpc_end2end_test_config config,
   GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(101)));
 
   for (i = 0; i < messages; i++) {
-    request_payload = grpc_byte_buffer_create(&request_payload_slice, 1);
-    response_payload = grpc_byte_buffer_create(&response_payload_slice, 1);
+    request_payload = grpc_raw_byte_buffer_create(&request_payload_slice, 1);
+    response_payload = grpc_raw_byte_buffer_create(&response_payload_slice, 1);
 
     op = ops;
     op->op = GRPC_OP_SEND_MESSAGE;

test/core/end2end/tests/request_response_with_binary_metadata_and_payload.c

@@ -105,9 +105,9 @@ static void test_request_response_with_metadata_and_payload(
   gpr_slice request_payload_slice = gpr_slice_from_copied_string("hello world");
   gpr_slice response_payload_slice = gpr_slice_from_copied_string("hello you");
   grpc_byte_buffer *request_payload =
-      grpc_byte_buffer_create(&request_payload_slice, 1);
+      grpc_raw_byte_buffer_create(&request_payload_slice, 1);
   grpc_byte_buffer *response_payload =
-      grpc_byte_buffer_create(&response_payload_slice, 1);
+      grpc_raw_byte_buffer_create(&response_payload_slice, 1);
   gpr_timespec deadline = five_seconds_time();
   grpc_metadata meta_c[2] = {
       {"key1-bin",

test/core/end2end/tests/request_response_with_metadata_and_payload.c

@@ -105,9 +105,9 @@ static void test_request_response_with_metadata_and_payload(
   gpr_slice request_payload_slice = gpr_slice_from_copied_string("hello world");
   gpr_slice response_payload_slice = gpr_slice_from_copied_string("hello you");
   grpc_byte_buffer *request_payload =
-      grpc_byte_buffer_create(&request_payload_slice, 1);
+      grpc_raw_byte_buffer_create(&request_payload_slice, 1);
   grpc_byte_buffer *response_payload =
-      grpc_byte_buffer_create(&response_payload_slice, 1);
+      grpc_raw_byte_buffer_create(&response_payload_slice, 1);
   gpr_timespec deadline = five_seconds_time();
   grpc_metadata meta_c[2] = {{"key1", "val1", 4, {{NULL, NULL, NULL}}},
                              {"key2", "val2", 4, {{NULL, NULL, NULL}}}};

test/core/end2end/tests/request_response_with_payload.c

@@ -103,9 +103,9 @@ static void request_response_with_payload(grpc_end2end_test_fixture f) {
   grpc_call *c;
   grpc_call *s;
   grpc_byte_buffer *request_payload =
-      grpc_byte_buffer_create(&request_payload_slice, 1);
+      grpc_raw_byte_buffer_create(&request_payload_slice, 1);
   grpc_byte_buffer *response_payload =
-      grpc_byte_buffer_create(&response_payload_slice, 1);
+      grpc_raw_byte_buffer_create(&response_payload_slice, 1);
   gpr_timespec deadline = five_seconds_time();
   cq_verifier *v_client = cq_verifier_create(f.client_cq);
   cq_verifier *v_server = cq_verifier_create(f.server_cq);

test/core/end2end/tests/request_response_with_payload_and_call_creds.c

@@ -156,9 +156,9 @@ static void request_response_with_payload_and_call_creds(
   gpr_slice request_payload_slice = gpr_slice_from_copied_string("hello world");
   gpr_slice response_payload_slice = gpr_slice_from_copied_string("hello you");
   grpc_byte_buffer *request_payload =
-      grpc_byte_buffer_create(&request_payload_slice, 1);
+      grpc_raw_byte_buffer_create(&request_payload_slice, 1);
   grpc_byte_buffer *response_payload =
-      grpc_byte_buffer_create(&response_payload_slice, 1);
+      grpc_raw_byte_buffer_create(&response_payload_slice, 1);
   gpr_timespec deadline = five_seconds_time();
 
   grpc_end2end_test_fixture f = begin_test(config, test_name, NULL, NULL);

test/core/end2end/tests/request_response_with_trailing_metadata_and_payload.c

@@ -105,9 +105,9 @@ static void test_request_response_with_metadata_and_payload(
   gpr_slice request_payload_slice = gpr_slice_from_copied_string("hello world");
   gpr_slice response_payload_slice = gpr_slice_from_copied_string("hello you");
   grpc_byte_buffer *request_payload =
-      grpc_byte_buffer_create(&request_payload_slice, 1);
+      grpc_raw_byte_buffer_create(&request_payload_slice, 1);
   grpc_byte_buffer *response_payload =
-      grpc_byte_buffer_create(&response_payload_slice, 1);
+      grpc_raw_byte_buffer_create(&response_payload_slice, 1);
   gpr_timespec deadline = five_seconds_time();
   grpc_metadata meta_c[2] = {{"key1", "val1", 4, {{NULL, NULL, NULL}}}, {"key2", "val2", 4, {{NULL, NULL, NULL}}}};
   grpc_metadata meta_s[2] = {{"key3", "val3", 4, {{NULL, NULL, NULL}}}, {"key4", "val4", 4, {{NULL, NULL, NULL}}}};

test/core/end2end/tests/request_with_large_metadata.c

@@ -103,7 +103,7 @@ static void test_request_with_large_metadata(grpc_end2end_test_config config) {
   grpc_call *s;
   gpr_slice request_payload_slice = gpr_slice_from_copied_string("hello world");
   grpc_byte_buffer *request_payload =
-      grpc_byte_buffer_create(&request_payload_slice, 1);
+      grpc_raw_byte_buffer_create(&request_payload_slice, 1);
   gpr_timespec deadline = five_seconds_time();
   grpc_metadata meta;
   grpc_end2end_test_fixture f = begin_test(config, "test_request_with_large_metadata", NULL, NULL);

test/core/end2end/tests/request_with_payload.c

@@ -103,7 +103,7 @@ static void test_invoke_request_with_payload(grpc_end2end_test_config config) {
   grpc_call *s;
   gpr_slice request_payload_slice = gpr_slice_from_copied_string("hello world");
   grpc_byte_buffer *request_payload =
-      grpc_byte_buffer_create(&request_payload_slice, 1);
+      grpc_raw_byte_buffer_create(&request_payload_slice, 1);
   gpr_timespec deadline = five_seconds_time();
   grpc_end2end_test_fixture f = begin_test(config, "test_invoke_request_with_payload", NULL, NULL);
   cq_verifier *v_client = cq_verifier_create(f.client_cq);

test/core/fling/client.c

@@ -183,7 +183,7 @@ int main(int argc, char **argv) {
 
   channel = grpc_channel_create(target, NULL);
   cq = grpc_completion_queue_create();
-  the_buffer = grpc_byte_buffer_create(&slice, payload_size);
+  the_buffer = grpc_raw_byte_buffer_create(&slice, payload_size);
   histogram = gpr_histogram_create(0.01, 60e9);
 
   sc.init();

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)
@@ -55,7 +57,7 @@ static void test_read_one_slice(void) {
 
   LOG_TEST("test_read_one_slice");
   slice = gpr_slice_from_copied_string("test");
-  buffer = grpc_byte_buffer_create(&slice, 1);
+  buffer = grpc_raw_byte_buffer_create(&slice, 1);
   gpr_slice_unref(slice);
   grpc_byte_buffer_reader_init(&reader, buffer);
   first_code = grpc_byte_buffer_reader_next(&reader, &first_slice);
@@ -77,7 +79,28 @@ static void test_read_one_slice_malloc(void) {
   LOG_TEST("test_read_one_slice_malloc");
   slice = gpr_slice_malloc(4);
   memcpy(GPR_SLICE_START_PTR(slice), "test", 4);
-  buffer = grpc_byte_buffer_create(&slice, 1);
+  buffer = grpc_raw_byte_buffer_create(&slice, 1);
+  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 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_raw_byte_buffer_create(&slice, 1);
   gpr_slice_unref(slice);
   grpc_byte_buffer_reader_init(&reader, buffer);
   first_code = grpc_byte_buffer_reader_next(&reader, &first_slice);
@@ -89,9 +112,61 @@ static void test_read_one_slice_malloc(void) {
   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_raw_compressed_byte_buffer_create(
+      sliceb_out.slices, sliceb_out.count, 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;
 }

tools/doxygen/Doxyfile.core

@@ -760,7 +760,7 @@ WARN_LOGFILE           =
 # spaces.
 # Note: If this tag is empty the current directory is searched.
 
-INPUT                  = include/grpc/grpc_security.h include/grpc/byte_buffer.h include/grpc/byte_buffer_reader.h include/grpc/grpc.h include/grpc/status.h include/grpc/census.h include/grpc/support/alloc.h include/grpc/support/atm.h include/grpc/support/atm_gcc_atomic.h include/grpc/support/atm_gcc_sync.h include/grpc/support/atm_win32.h include/grpc/support/cancellable_platform.h include/grpc/support/cmdline.h include/grpc/support/cpu.h include/grpc/support/histogram.h include/grpc/support/host_port.h include/grpc/support/log.h include/grpc/support/log_win32.h include/grpc/support/port_platform.h include/grpc/support/slice.h include/grpc/support/slice_buffer.h include/grpc/support/string_util.h include/grpc/support/subprocess.h include/grpc/support/sync.h include/grpc/support/sync_generic.h include/grpc/support/sync_posix.h include/grpc/support/sync_win32.h include/grpc/support/thd.h include/grpc/support/time.h include/grpc/support/tls.h include/grpc/support/tls_gcc.h include/grpc/support/tls_msvc.h include/grpc/support/tls_pthread.h include/grpc/support/useful.h
+INPUT                  = include/grpc/grpc_security.h include/grpc/byte_buffer.h include/grpc/byte_buffer_reader.h include/grpc/compression.h include/grpc/grpc.h include/grpc/status.h include/grpc/census.h include/grpc/support/alloc.h include/grpc/support/atm.h include/grpc/support/atm_gcc_atomic.h include/grpc/support/atm_gcc_sync.h include/grpc/support/atm_win32.h include/grpc/support/cancellable_platform.h include/grpc/support/cmdline.h include/grpc/support/cpu.h include/grpc/support/histogram.h include/grpc/support/host_port.h include/grpc/support/log.h include/grpc/support/log_win32.h include/grpc/support/port_platform.h include/grpc/support/slice.h include/grpc/support/slice_buffer.h include/grpc/support/string_util.h include/grpc/support/subprocess.h include/grpc/support/sync.h include/grpc/support/sync_generic.h include/grpc/support/sync_posix.h include/grpc/support/sync_win32.h include/grpc/support/thd.h include/grpc/support/time.h include/grpc/support/tls.h include/grpc/support/tls_gcc.h include/grpc/support/tls_msvc.h include/grpc/support/tls_pthread.h include/grpc/support/useful.h
 
 # This tag can be used to specify the character encoding of the source files
 # that doxygen parses. Internally doxygen uses the UTF-8 encoding. Doxygen uses

vsprojects/grpc/grpc.vcxproj

@@ -150,6 +150,7 @@
     <ClInclude Include="..\..\include\grpc\grpc_security.h" />
     <ClInclude Include="..\..\include\grpc\byte_buffer.h" />
     <ClInclude Include="..\..\include\grpc\byte_buffer_reader.h" />
+    <ClInclude Include="..\..\include\grpc\compression.h" />
     <ClInclude Include="..\..\include\grpc\grpc.h" />
     <ClInclude Include="..\..\include\grpc\status.h" />
     <ClInclude Include="..\..\include\grpc\census.h" />

vsprojects/grpc/grpc.vcxproj.filters

@@ -360,6 +360,9 @@
     <ClInclude Include="..\..\include\grpc\byte_buffer_reader.h">
       <Filter>include\grpc</Filter>
     </ClInclude>
+    <ClInclude Include="..\..\include\grpc\compression.h">
+      <Filter>include\grpc</Filter>
+    </ClInclude>
     <ClInclude Include="..\..\include\grpc\grpc.h">
       <Filter>include\grpc</Filter>
     </ClInclude>

vsprojects/grpc_unsecure/grpc_unsecure.vcxproj

@@ -148,6 +148,7 @@
   <ItemGroup>
     <ClInclude Include="..\..\include\grpc\byte_buffer.h" />
     <ClInclude Include="..\..\include\grpc\byte_buffer_reader.h" />
+    <ClInclude Include="..\..\include\grpc\compression.h" />
     <ClInclude Include="..\..\include\grpc\grpc.h" />
     <ClInclude Include="..\..\include\grpc\status.h" />
     <ClInclude Include="..\..\include\grpc\census.h" />

vsprojects/grpc_unsecure/grpc_unsecure.vcxproj.filters

@@ -291,6 +291,9 @@
     <ClInclude Include="..\..\include\grpc\byte_buffer_reader.h">
       <Filter>include\grpc</Filter>
     </ClInclude>
+    <ClInclude Include="..\..\include\grpc\compression.h">
+      <Filter>include\grpc</Filter>
+    </ClInclude>
     <ClInclude Include="..\..\include\grpc\grpc.h">
       <Filter>include\grpc</Filter>
     </ClInclude>