Add codec to encode/decode GRPC data frame.

source/common/CMakeLists.txt

@@ -45,6 +45,7 @@ add_library(
   filter/ratelimit.cc
   filter/tcp_proxy.cc
   generated/ratelimit.pb.cc
+  grpc/codec.cc
   grpc/common.cc
   grpc/http1_bridge_filter.cc
   grpc/rpc_channel_impl.cc

source/common/grpc/codec.cc

@@ -0,0 +1,89 @@
+#include "common/grpc/codec.h"
+
+#include "common/buffer/buffer_impl.h"
+
+namespace Grpc {
+
+Encoder::Encoder() {}
+
+void Encoder::NewFrame(uint8_t flags, uint64_t length, uint8_t* output) {
+  output[0] = flags;
+  output[1] = static_cast<uint8_t>(length >> 24);
+  output[2] = static_cast<uint8_t>(length >> 16);
+  output[3] = static_cast<uint8_t>(length >> 8);
+  output[4] = static_cast<uint8_t>(length);
+}
+
+Decoder::Decoder() : state_(State::FH_FLAG) {}
+
+bool Decoder::Decode(Buffer::Instance& input, std::vector<Frame>* output) {
+  uint64_t count = input.getRawSlices(nullptr, 0);
+  Buffer::RawSlice slices[count];
+  input.getRawSlices(slices, count);
+  for (Buffer::RawSlice& slice : slices) {
+    uint8_t* mem = reinterpret_cast<uint8_t*>(slice.mem_);
+    for (uint64_t j = 0; j < slice.len_;) {
+      uint8_t c = *mem;
+      switch (state_) {
+      case State::FH_FLAG:
+        if (c & ~GRPC_FH_COMPRESSED) {
+          // Unsupported flags.
+          return false;
+        }
+        frame_.flags = c;
+        state_ = State::FH_LEN_0;
+        mem++;
+        j++;
+        break;
+      case State::FH_LEN_0:
+        frame_.length = static_cast<uint32_t>(c) << 24;
+        state_ = State::FH_LEN_1;
+        mem++;
+        j++;
+        break;
+      case State::FH_LEN_1:
+        frame_.length |= static_cast<uint32_t>(c) << 16;
+        state_ = State::FH_LEN_2;
+        mem++;
+        j++;
+        break;
+      case State::FH_LEN_2:
+        frame_.length |= static_cast<uint32_t>(c) << 8;
+        state_ = State::FH_LEN_3;
+        mem++;
+        j++;
+        break;
+      case State::FH_LEN_3:
+        frame_.length |= static_cast<uint32_t>(c);
+        frame_.data = new Buffer::OwnedImpl();
+        state_ = State::DATA;
+        mem++;
+        j++;
+        break;
+      case State::DATA:
+        uint64_t remain_in_buffer = slice.len_ - j;
+        uint64_t remain_in_frame = frame_.length - frame_.data->length();
+        if (remain_in_buffer <= remain_in_frame) {
+          frame_.data->add(mem, remain_in_buffer);
+          mem += remain_in_buffer;
+          j += remain_in_buffer;
+        } else {
+          frame_.data->add(mem, remain_in_frame);
+          mem += remain_in_frame;
+          j += remain_in_frame;
+        }
+        if (frame_.length == frame_.data->length()) {
+          output->push_back(frame_); // make a copy.
+          frame_.flags = 0;
+          frame_.length = 0;
+          frame_.data = nullptr;
+          state_ = State::FH_FLAG;
+        }
+        break;
+      }
+    }
+  }
+  return true;
+}
+
+} // namespace Grpc

source/common/grpc/codec.h

@@ -0,0 +1,74 @@
+#pragma once
+
+#include "envoy/buffer/buffer.h"
+
+namespace Grpc {
+// Last bit for an expanded message without compression.
+const uint8_t GRPC_FH_DEFAULT = 0b0u;
+// Last bit for a compressed message.
+const uint8_t GRPC_FH_COMPRESSED = 0b1u;
+
+enum class CompressionAlgorithm { None, Gzip };
+
+struct Frame {
+  uint8_t flags;
+  uint32_t length;
+  Buffer::Instance* data;
+};
+
+class Encoder {
+public:
+  Encoder();
+
+  // Creates a new GRPC data frame with the given flags and length.
+  // @param flags supplies the GRPC data frame flags.
+  // @param length supplies the GRPC data frame length.
+  // @param output the buffer to store the encoded data, it's size must be 5.
+  void NewFrame(uint8_t flags, uint64_t length, uint8_t* output);
+};
+
+class Decoder {
+public:
+  Decoder();
+
+  // Decodes the given buffer with GRPC data frame.
+  // @param input supplies the binary octets wrapped in a GRPC data frame.
+  // @param input supplies the buffer to store the decoded data.
+  // @return bool whether the decoding success.
+  bool Decode(Buffer::Instance& input, std::vector<Frame>* output);
+
+private:
+  // Wiring format of GRPC data frame header:
+  // -----------------------------------------------------------------------
+  // |R|R|R|R|R|R|R|R|C|      L     |      L     |      L     |      L     |
+  // -----------------------------------------------------------------------
+  //    Flag (1 byte)                Message Length (4 bytes)
+  //
+  // A fixed header consists of five bytes.
+  // The first byte is the Flag. The last one "C" bit indicates if the message
+  // is compressed or not (0 is uncompressed, 1 is compressed). The rest seven
+  // "R" bits is reserved for future use.
+  // The next four "L" bytes represent the message length in BigEndian format.
+  enum class State {
+    // Waiting for decoding the flags (1 byte) of the GRPC data frame.
+    FH_FLAG,
+    // Waiting for decoding the 1st byte of the length (4 bytes in total) of the
+    // GRPC data frame.
+    FH_LEN_0,
+    // Waiting for decoding the 2nd byte of the length (4 bytes in total) of the
+    // GRPC data frame.
+    FH_LEN_1,
+    // Waiting for decoding the 3rd byte of the length (4 bytes in total) of the
+    // GRPC data frame.
+    FH_LEN_2,
+    // Waiting for decoding the 4th byte of the length (4 bytes in total) of the
+    // GRPC data frame.
+    FH_LEN_3,
+    // Waiting for decoding the data.
+    DATA,
+  };
+
+  State state_;
+  Frame frame_;
+};
+} // Grpc

test/CMakeLists.txt

@@ -36,6 +36,7 @@ add_executable(envoy-test
   common/event/file_event_impl_test.cc
   common/filesystem/filesystem_impl_test.cc
   common/filesystem/watcher_impl_test.cc
+  common/grpc/codec_test.cc
   common/grpc/common_test.cc
   common/grpc/http1_bridge_filter_test.cc
   common/grpc/rpc_channel_impl_test.cc

test/common/grpc/codec_test.cc

@@ -0,0 +1,122 @@
+#include "common/buffer/buffer_impl.h"
+#include "common/grpc/codec.h"
+#include "test/generated/helloworld.pb.h"
+
+namespace Grpc {
+
+TEST(CodecTest, encodeHeader) {
+  Encoder encoder;
+  uint8_t buffer[5];
+
+  encoder.NewFrame(GRPC_FH_DEFAULT, 1, buffer);
+  EXPECT_EQ(buffer[0], GRPC_FH_DEFAULT);
+  EXPECT_EQ(buffer[1], 0);
+  EXPECT_EQ(buffer[2], 0);
+  EXPECT_EQ(buffer[3], 0);
+  EXPECT_EQ(buffer[4], 1);
+
+  encoder.NewFrame(GRPC_FH_COMPRESSED, 1, buffer);
+  EXPECT_EQ(buffer[0], GRPC_FH_COMPRESSED);
+  EXPECT_EQ(buffer[1], 0);
+  EXPECT_EQ(buffer[2], 0);
+  EXPECT_EQ(buffer[3], 0);
+  EXPECT_EQ(buffer[4], 1);
+
+  encoder.NewFrame(GRPC_FH_DEFAULT, 0x100, buffer);
+  EXPECT_EQ(buffer[0], GRPC_FH_DEFAULT);
+  EXPECT_EQ(buffer[1], 0);
+  EXPECT_EQ(buffer[2], 0);
+  EXPECT_EQ(buffer[3], 1);
+  EXPECT_EQ(buffer[4], 0);
+
+  encoder.NewFrame(GRPC_FH_DEFAULT, 0x10000, buffer);
+  EXPECT_EQ(buffer[0], GRPC_FH_DEFAULT);
+  EXPECT_EQ(buffer[1], 0);
+  EXPECT_EQ(buffer[2], 1);
+  EXPECT_EQ(buffer[3], 0);
+  EXPECT_EQ(buffer[4], 0);
+
+  encoder.NewFrame(GRPC_FH_DEFAULT, 0x1000000, buffer);
+  EXPECT_EQ(buffer[0], GRPC_FH_DEFAULT);
+  EXPECT_EQ(buffer[1], 1);
+  EXPECT_EQ(buffer[2], 0);
+  EXPECT_EQ(buffer[3], 0);
+  EXPECT_EQ(buffer[4], 0);
+}
+
+TEST(CodecTest, decodeInvalidFrame) {
+  helloworld::HelloRequest request;
+  request.set_name("hello");
+
+  Buffer::OwnedImpl buffer;
+  uint8_t header[5];
+  Encoder encoder;
+  encoder.NewFrame(0b10u, request.ByteSize(), header);
+  buffer.add(header, 5);
+  buffer.add(request.SerializeAsString());
+
+  std::vector<Frame> frames;
+  Decoder decoder;
+  EXPECT_FALSE(decoder.Decode(buffer, &frames));
+  for (Frame& frame : frames) {
+    delete frame.data;
+  }
+}
+
+TEST(CodecTest, decodeSingleFrame) {
+  helloworld::HelloRequest request;
+  request.set_name("hello");
+
+  Buffer::OwnedImpl buffer;
+  uint8_t header[5];
+  Encoder encoder;
+  encoder.NewFrame(GRPC_FH_DEFAULT, request.ByteSize(), header);
+  buffer.add(header, 5);
+  buffer.add(request.SerializeAsString());
+
+  std::vector<Frame> frames;
+  Decoder decoder;
+  decoder.Decode(buffer, &frames);
+  EXPECT_EQ(frames.size(), static_cast<uint64_t>(1));
+  EXPECT_EQ(GRPC_FH_DEFAULT, frames[0].flags);
+  EXPECT_EQ(static_cast<uint64_t>(request.ByteSize()), frames[0].length);
+
+  helloworld::HelloRequest result;
+  result.ParseFromArray(frames[0].data->linearize(frames[0].data->length()),
+                        frames[0].data->length());
+  EXPECT_EQ("hello", result.name());
+  for (Frame& frame : frames) {
+    delete frame.data;
+  }
+}
+
+TEST(CodecTest, decodeMultipleFrame) {
+  helloworld::HelloRequest request;
+  request.set_name("hello");
+
+  Buffer::OwnedImpl buffer;
+  uint8_t header[5];
+  Encoder encoder;
+  encoder.NewFrame(GRPC_FH_DEFAULT, request.ByteSize(), header);
+  for (int i = 0; i < 1009; i++) {
+    buffer.add(header, 5);
+    buffer.add(request.SerializeAsString());
+  }
+
+  std::vector<Frame> frames;
+  Decoder decoder;
+  decoder.Decode(buffer, &frames);
+  EXPECT_EQ(frames.size(), static_cast<uint64_t>(1009));
+  for (Frame& frame : frames) {
+    EXPECT_EQ(GRPC_FH_DEFAULT, frame.flags);
+    EXPECT_EQ(static_cast<uint64_t>(request.ByteSize()), frame.length);
+
+    helloworld::HelloRequest result;
+    result.ParseFromArray(frame.data->linearize(frame.data->length()), frame.data->length());
+    EXPECT_EQ("hello", result.name());
+  }
+  for (Frame& frame : frames) {
+    delete frame.data;
+  }
+}
+} // Grpc