Built motion from commit 6a09e18b.|2.6.11

[motion2.git] / legacy-libs / grpc-cloned / deps / grpc / src / core / lib / transport / byte_stream.cc

diff --git a/legacy-libs/grpc-cloned/deps/grpc/src/core/lib/transport/byte_stream.cc b/legacy-libs/grpc-cloned/deps/grpc/src/core/lib/transport/byte_stream.cc
new file mode 100644 (file)
index 0000000..1dd234c
--- /dev/null
+++ b/legacy-libs/grpc-cloned/deps/grpc/src/core/lib/transport/byte_stream.cc
@@ -0,0 +1,158 @@
+/*
+ *
+ * Copyright 2015 gRPC authors.
+ *
+ * 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 <grpc/support/port_platform.h>
+
+#include "src/core/lib/transport/byte_stream.h"
+
+#include <stdlib.h>
+#include <string.h>
+
+#include <grpc/support/log.h>
+
+#include "src/core/lib/gprpp/memory.h"
+#include "src/core/lib/slice/slice_internal.h"
+
+namespace grpc_core {
+
+//
+// SliceBufferByteStream
+//
+
+SliceBufferByteStream::SliceBufferByteStream(grpc_slice_buffer* slice_buffer,
+                                             uint32_t flags)
+    : ByteStream(static_cast<uint32_t>(slice_buffer->length), flags) {
+  GPR_ASSERT(slice_buffer->length <= UINT32_MAX);
+  grpc_slice_buffer_init(&backing_buffer_);
+  grpc_slice_buffer_swap(slice_buffer, &backing_buffer_);
+}
+
+SliceBufferByteStream::~SliceBufferByteStream() {}
+
+void SliceBufferByteStream::Orphan() {
+  grpc_slice_buffer_destroy_internal(&backing_buffer_);
+  GRPC_ERROR_UNREF(shutdown_error_);
+  // Note: We do not actually delete the object here, since
+  // SliceBufferByteStream is usually allocated as part of a larger
+  // object and has an OrphanablePtr of itself passed down through the
+  // filter stack.
+}
+
+bool SliceBufferByteStream::Next(size_t max_size_hint,
+                                 grpc_closure* on_complete) {
+  GPR_DEBUG_ASSERT(backing_buffer_.count > 0);
+  return true;
+}
+
+grpc_error* SliceBufferByteStream::Pull(grpc_slice* slice) {
+  if (GPR_UNLIKELY(shutdown_error_ != GRPC_ERROR_NONE)) {
+    return GRPC_ERROR_REF(shutdown_error_);
+  }
+  *slice = grpc_slice_buffer_take_first(&backing_buffer_);
+  return GRPC_ERROR_NONE;
+}
+
+void SliceBufferByteStream::Shutdown(grpc_error* error) {
+  GRPC_ERROR_UNREF(shutdown_error_);
+  shutdown_error_ = error;
+}
+
+//
+// ByteStreamCache
+//
+
+ByteStreamCache::ByteStreamCache(OrphanablePtr<ByteStream> underlying_stream)
+    : underlying_stream_(std::move(underlying_stream)),
+      length_(underlying_stream_->length()),
+      flags_(underlying_stream_->flags()) {
+  grpc_slice_buffer_init(&cache_buffer_);
+}
+
+ByteStreamCache::~ByteStreamCache() { Destroy(); }
+
+void ByteStreamCache::Destroy() {
+  underlying_stream_.reset();
+  if (cache_buffer_.length > 0) {
+    grpc_slice_buffer_destroy_internal(&cache_buffer_);
+  }
+}
+
+//
+// ByteStreamCache::CachingByteStream
+//
+
+ByteStreamCache::CachingByteStream::CachingByteStream(ByteStreamCache* cache)
+    : ByteStream(cache->length_, cache->flags_), cache_(cache) {}
+
+ByteStreamCache::CachingByteStream::~CachingByteStream() {}
+
+void ByteStreamCache::CachingByteStream::Orphan() {
+  GRPC_ERROR_UNREF(shutdown_error_);
+  // Note: We do not actually delete the object here, since
+  // CachingByteStream is usually allocated as part of a larger
+  // object and has an OrphanablePtr of itself passed down through the
+  // filter stack.
+}
+
+bool ByteStreamCache::CachingByteStream::Next(size_t max_size_hint,
+                                              grpc_closure* on_complete) {
+  if (shutdown_error_ != GRPC_ERROR_NONE) return true;
+  if (cursor_ < cache_->cache_buffer_.count) return true;
+  GPR_ASSERT(cache_->underlying_stream_ != nullptr);
+  return cache_->underlying_stream_->Next(max_size_hint, on_complete);
+}
+
+grpc_error* ByteStreamCache::CachingByteStream::Pull(grpc_slice* slice) {
+  if (shutdown_error_ != GRPC_ERROR_NONE) {
+    return GRPC_ERROR_REF(shutdown_error_);
+  }
+  if (cursor_ < cache_->cache_buffer_.count) {
+    *slice = grpc_slice_ref_internal(cache_->cache_buffer_.slices[cursor_]);
+    ++cursor_;
+    offset_ += GRPC_SLICE_LENGTH(*slice);
+    return GRPC_ERROR_NONE;
+  }
+  GPR_ASSERT(cache_->underlying_stream_ != nullptr);
+  grpc_error* error = cache_->underlying_stream_->Pull(slice);
+  if (error == GRPC_ERROR_NONE) {
+    grpc_slice_buffer_add(&cache_->cache_buffer_,
+                          grpc_slice_ref_internal(*slice));
+    ++cursor_;
+    offset_ += GRPC_SLICE_LENGTH(*slice);
+    // Orphan the underlying stream if it's been drained.
+    if (offset_ == cache_->underlying_stream_->length()) {
+      cache_->underlying_stream_.reset();
+    }
+  }
+  return error;
+}
+
+void ByteStreamCache::CachingByteStream::Shutdown(grpc_error* error) {
+  GRPC_ERROR_UNREF(shutdown_error_);
+  shutdown_error_ = GRPC_ERROR_REF(error);
+  if (cache_->underlying_stream_ != nullptr) {
+    cache_->underlying_stream_->Shutdown(error);
+  }
+}
+
+void ByteStreamCache::CachingByteStream::Reset() {
+  cursor_ = 0;
+  offset_ = 0;
+}
+
+}  // namespace grpc_core