Add envoy Buffer based TranscoderInputStream

src/envoy/transcoding/BUILD

@@ -0,0 +1,40 @@
+# Copyright 2016 Google Inc. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+################################################################################
+#
+cc_library(
+    name = "envoy_input_stream",
+    srcs = [
+        "envoy_input_stream.cc",
+    ],
+    hdrs = [
+        "envoy_input_stream.h",
+    ],
+    deps = [
+        "//contrib/endpoints/src/grpc/transcoding:transcoder_input_stream",
+        "@envoy//source/exe:envoy_common_lib",
+    ],
+)
+
+cc_test(
+    name = "envoy_input_stream_test",
+    srcs = [
+        "envoy_input_stream_test.cc",
+    ],
+    deps = [
+        ":envoy_input_stream",
+        "@googletest_git//:googletest_main",
+    ],
+)

src/envoy/transcoding/envoy_input_stream.cc

@@ -0,0 +1,63 @@
+/* Copyright 2017 Istio Authors. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "src/envoy/transcoding/envoy_input_stream.h"
+
+namespace Grpc {
+
+void EnvoyInputStream::Move(Buffer::Instance &instance) {
+  if (!finished_) {
+    buffer_.move(instance);
+  }
+}
+
+bool EnvoyInputStream::Next(const void **data, int *size) {
+  if (position_ != 0) {
+    buffer_.drain(position_);
+    position_ = 0;
+  }
+
+  Buffer::RawSlice slice;
+  uint64_t num_slices = buffer_.getRawSlices(&slice, 1);
+
+  if (num_slices) {
+    *data = slice.mem_;
+    *size = slice.len_;
+    position_ = slice.len_;
+    byte_count_ += slice.len_;
+    return true;
+  }
+
+  if (!finished_) {
+    *data = nullptr;
+    *size = 0;
+    return true;
+  }
+  return false;
+}
+
+void EnvoyInputStream::BackUp(int count) {
+  GOOGLE_CHECK_GE(count, 0);
+  GOOGLE_CHECK_LE(count, position_);
+
+  position_ -= count;
+  byte_count_ -= count;
+}
+
+int64_t EnvoyInputStream::BytesAvailable() const {
+  return buffer_.length() - position_;
+}
+
+}  // namespace Grpc

src/envoy/transcoding/envoy_input_stream.h

@@ -0,0 +1,51 @@
+/* Copyright 2017 Istio Authors. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#pragma once
+
+#include "precompiled/precompiled.h"
+
+#include "common/buffer/buffer_impl.h"
+#include "contrib/endpoints/src/grpc/transcoding/transcoder_input_stream.h"
+
+namespace Grpc {
+
+class EnvoyInputStream
+    : public google::api_manager::transcoding::TranscoderInputStream {
+ public:
+  // Add a buffer to input stream, will consume all buffer from parameter
+  // if the stream is not finished
+  void Move(Buffer::Instance &instance);
+
+  // Mark the buffer is finished
+  void Finish() { finished_ = true; }
+
+  // TranscoderInputStream
+  virtual bool Next(const void **data, int *size) override;
+  virtual void BackUp(int count) override;
+  virtual bool Skip(int count) override { return false; }  // Not implemented
+  virtual google::protobuf::int64 ByteCount() const override {
+    return byte_count_;
+  }
+  virtual int64_t BytesAvailable() const override;
+
+ private:
+  Buffer::OwnedImpl buffer_;
+  int position_{0};
+  int64_t byte_count_{0};
+  bool finished_{false};
+};
+
+}  // namespace Grpc

src/envoy/transcoding/envoy_input_stream_test.cc

@@ -0,0 +1,97 @@
+/* Copyright 2017 Istio Authors. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "src/envoy/transcoding/envoy_input_stream.h"
+
+#include "gtest/gtest.h"
+
+namespace Grpc {
+namespace {
+
+class EnvoyInputStreamTest : public testing::Test {
+ public:
+  EnvoyInputStreamTest() {
+    Buffer::OwnedImpl buffer{"abcd"};
+    stream_.Move(buffer);
+  }
+
+  std::string slice_data_{"abcd"};
+  EnvoyInputStream stream_;
+
+  const void *data_;
+  int size_;
+};
+
+TEST_F(EnvoyInputStreamTest, Move) {
+  Buffer::OwnedImpl buffer{"abcd"};
+  stream_.Move(buffer);
+
+  EXPECT_EQ(0, buffer.length());
+  EXPECT_EQ(8, stream_.BytesAvailable());
+}
+
+TEST_F(EnvoyInputStreamTest, Next) {
+  EXPECT_TRUE(stream_.Next(&data_, &size_));
+  EXPECT_EQ(4, size_);
+  EXPECT_EQ(0, memcmp(slice_data_.data(), data_, size_));
+}
+
+TEST_F(EnvoyInputStreamTest, TwoSlices) {
+  Buffer::OwnedImpl buffer("efgh");
+
+  stream_.Move(buffer);
+
+  EXPECT_TRUE(stream_.Next(&data_, &size_));
+  EXPECT_EQ(4, size_);
+  EXPECT_EQ(0, memcmp(slice_data_.data(), data_, size_));
+  EXPECT_TRUE(stream_.Next(&data_, &size_));
+  EXPECT_EQ(4, size_);
+  EXPECT_EQ(0, memcmp("efgh", data_, size_));
+}
+
+TEST_F(EnvoyInputStreamTest, BackUp) {
+  EXPECT_TRUE(stream_.Next(&data_, &size_));
+  EXPECT_EQ(4, size_);
+
+  stream_.BackUp(3);
+  EXPECT_EQ(3, stream_.BytesAvailable());
+  EXPECT_EQ(1, stream_.ByteCount());
+
+  EXPECT_TRUE(stream_.Next(&data_, &size_));
+  EXPECT_EQ(3, size_);
+  EXPECT_EQ(0, memcmp("bcd", data_, size_));
+  EXPECT_EQ(4, stream_.ByteCount());
+}
+
+TEST_F(EnvoyInputStreamTest, ByteCount) {
+  EXPECT_EQ(0, stream_.ByteCount());
+  EXPECT_TRUE(stream_.Next(&data_, &size_));
+  EXPECT_EQ(4, stream_.ByteCount());
+}
+
+TEST_F(EnvoyInputStreamTest, Finish) {
+  EXPECT_TRUE(stream_.Next(&data_, &size_));
+  EXPECT_TRUE(stream_.Next(&data_, &size_));
+  EXPECT_EQ(0, size_);
+  stream_.Finish();
+  EXPECT_FALSE(stream_.Next(&data_, &size_));
+
+  Buffer::OwnedImpl buffer("efgh");
+  stream_.Move(buffer);
+
+  EXPECT_EQ(4, buffer.length());
+}
+}
+}