legacy-libs/grpc/deps/grpc/src/core/lib/transport/byte_stream.h

   1 /*
   2  *
   3  * Copyright 2015 gRPC authors.
   4  *
   5  * Licensed under the Apache License, Version 2.0 (the "License");
   6  * you may not use this file except in compliance with the License.
   7  * You may obtain a copy of the License at
   8  *
   9  *     http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  * Unless required by applicable law or agreed to in writing, software
  12  * distributed under the License is distributed on an "AS IS" BASIS,
  13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14  * See the License for the specific language governing permissions and
  15  * limitations under the License.
  16  *
  17  */
  18
  19 #ifndef GRPC_CORE_LIB_TRANSPORT_BYTE_STREAM_H
  20 #define GRPC_CORE_LIB_TRANSPORT_BYTE_STREAM_H
  21
  22 #include <grpc/support/port_platform.h>
  23
  24 #include <grpc/slice_buffer.h>
  25 #include "src/core/lib/gprpp/abstract.h"
  26 #include "src/core/lib/gprpp/orphanable.h"
  27 #include "src/core/lib/iomgr/closure.h"
  28
  29 /** Internal bit flag for grpc_begin_message's \a flags signaling the use of
  30  * compression for the message */
  31 #define GRPC_WRITE_INTERNAL_COMPRESS (0x80000000u)
  32 /** Mask of all valid internal flags. */
  33 #define GRPC_WRITE_INTERNAL_USED_MASK (GRPC_WRITE_INTERNAL_COMPRESS)
  34
  35 namespace grpc_core {
  36
  37 class ByteStream : public Orphanable {
  38  public:
  39   virtual ~ByteStream() {}
  40
  41   // Returns true if the bytes are available immediately (in which case
  42   // on_complete will not be called), or false if the bytes will be available
  43   // asynchronously (in which case on_complete will be called when they
  44   // are available).
  45   //
  46   // max_size_hint can be set as a hint as to the maximum number
  47   // of bytes that would be acceptable to read.
  48   virtual bool Next(size_t max_size_hint,
  49                     grpc_closure* on_complete) GRPC_ABSTRACT;
  50
  51   // Returns the next slice in the byte stream when it is available, as
  52   // indicated by Next().
  53   //
  54   // Once a slice is returned into *slice, it is owned by the caller.
  55   virtual grpc_error* Pull(grpc_slice* slice) GRPC_ABSTRACT;
  56
  57   // Shuts down the byte stream.
  58   //
  59   // If there is a pending call to on_complete from Next(), it will be
  60   // invoked with the error passed to Shutdown().
  61   //
  62   // The next call to Pull() (if any) will return the error passed to
  63   // Shutdown().
  64   virtual void Shutdown(grpc_error* error) GRPC_ABSTRACT;
  65
  66   uint32_t length() const { return length_; }
  67   uint32_t flags() const { return flags_; }
  68
  69   void set_flags(uint32_t flags) { flags_ = flags; }
  70
  71   GRPC_ABSTRACT_BASE_CLASS
  72
  73  protected:
  74   ByteStream(uint32_t length, uint32_t flags)
  75       : length_(length), flags_(flags) {}
  76
  77  private:
  78   const uint32_t length_;
  79   uint32_t flags_;
  80 };
  81
  82 //
  83 // SliceBufferByteStream
  84 //
  85 // A ByteStream that wraps a slice buffer.
  86 //
  87
  88 class SliceBufferByteStream : public ByteStream {
  89  public:
  90   // Removes all slices in slice_buffer, leaving it empty.
  91   SliceBufferByteStream(grpc_slice_buffer* slice_buffer, uint32_t flags);
  92
  93   ~SliceBufferByteStream();
  94
  95   void Orphan() override;
  96
  97   bool Next(size_t max_size_hint, grpc_closure* on_complete) override;
  98   grpc_error* Pull(grpc_slice* slice) override;
  99   void Shutdown(grpc_error* error) override;
 100
 101  private:
 102   grpc_error* shutdown_error_ = GRPC_ERROR_NONE;
 103   grpc_slice_buffer backing_buffer_;
 104 };
 105
 106 //
 107 // CachingByteStream
 108 //
 109 // A ByteStream that that wraps an underlying byte stream but caches
 110 // the resulting slices in a slice buffer.  If an initial attempt fails
 111 // without fully draining the underlying stream, a new caching stream
 112 // can be created from the same underlying cache, in which case it will
 113 // return whatever is in the backing buffer before continuing to read the
 114 // underlying stream.
 115 //
 116 // NOTE: No synchronization is done, so it is not safe to have multiple
 117 // CachingByteStreams simultaneously drawing from the same underlying
 118 // ByteStreamCache at the same time.
 119 //
 120
 121 class ByteStreamCache {
 122  public:
 123   class CachingByteStream : public ByteStream {
 124    public:
 125     explicit CachingByteStream(ByteStreamCache* cache);
 126
 127     ~CachingByteStream();
 128
 129     void Orphan() override;
 130
 131     bool Next(size_t max_size_hint, grpc_closure* on_complete) override;
 132     grpc_error* Pull(grpc_slice* slice) override;
 133     void Shutdown(grpc_error* error) override;
 134
 135     // Resets the byte stream to the start of the underlying stream.
 136     void Reset();
 137
 138    private:
 139     ByteStreamCache* cache_;
 140     size_t cursor_ = 0;
 141     size_t offset_ = 0;
 142     grpc_error* shutdown_error_ = GRPC_ERROR_NONE;
 143   };
 144
 145   explicit ByteStreamCache(OrphanablePtr<ByteStream> underlying_stream);
 146
 147   ~ByteStreamCache();
 148
 149   // Must not be destroyed while still in use by a CachingByteStream.
 150   void Destroy();
 151
 152   grpc_slice_buffer* cache_buffer() { return &cache_buffer_; }
 153
 154  private:
 155   OrphanablePtr<ByteStream> underlying_stream_;
 156   uint32_t length_;
 157   uint32_t flags_;
 158   grpc_slice_buffer cache_buffer_;
 159 };
 160
 161 }  // namespace grpc_core
 162
 163 #endif /* GRPC_CORE_LIB_TRANSPORT_BYTE_STREAM_H */