|
|
|
@ -79,7 +79,9 @@ void SliceBufferByteStream::Shutdown(grpc_error* error) { |
|
|
|
|
//
|
|
|
|
|
|
|
|
|
|
ByteStreamCache::ByteStreamCache(OrphanablePtr<ByteStream> underlying_stream) |
|
|
|
|
: underlying_stream_(std::move(underlying_stream)) { |
|
|
|
|
: underlying_stream_(std::move(underlying_stream)), |
|
|
|
|
length_(underlying_stream_->length()), |
|
|
|
|
flags_(underlying_stream_->flags()) { |
|
|
|
|
grpc_slice_buffer_init(&cache_buffer_); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
@ -97,8 +99,7 @@ void ByteStreamCache::Destroy() { |
|
|
|
|
//
|
|
|
|
|
|
|
|
|
|
ByteStreamCache::CachingByteStream::CachingByteStream(ByteStreamCache* cache) |
|
|
|
|
: ByteStream(cache->underlying_stream_->length(), |
|
|
|
|
cache->underlying_stream_->flags()), |
|
|
|
|
: ByteStream(cache->length_, cache->flags_), |
|
|
|
|
cache_(cache) {} |
|
|
|
|
|
|
|
|
|
ByteStreamCache::CachingByteStream::~CachingByteStream() {} |
|
|
|
@ -115,6 +116,7 @@ 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); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
@ -128,6 +130,7 @@ grpc_error* ByteStreamCache::CachingByteStream::Pull(grpc_slice* slice) { |
|
|
|
|
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_, |
|
|
|
@ -145,9 +148,14 @@ grpc_error* ByteStreamCache::CachingByteStream::Pull(grpc_slice* slice) { |
|
|
|
|
void ByteStreamCache::CachingByteStream::Shutdown(grpc_error* error) { |
|
|
|
|
GRPC_ERROR_UNREF(shutdown_error_); |
|
|
|
|
shutdown_error_ = GRPC_ERROR_REF(error); |
|
|
|
|
cache_->underlying_stream_->Shutdown(error); |
|
|
|
|
if (cache_->underlying_stream_ != nullptr) { |
|
|
|
|
cache_->underlying_stream_->Shutdown(error); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
void ByteStreamCache::CachingByteStream::Reset() { cursor_ = 0; } |
|
|
|
|
void ByteStreamCache::CachingByteStream::Reset() { |
|
|
|
|
cursor_ = 0; |
|
|
|
|
offset_ = 0; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
} // namespace grpc_core
|
|
|
|
|