include/grpcpp/impl/codegen/async_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 GRPCPP_IMPL_CODEGEN_ASYNC_STREAM_H
  20 #define GRPCPP_IMPL_CODEGEN_ASYNC_STREAM_H
  21
  22 #include <grpcpp/impl/codegen/call.h>
  23 #include <grpcpp/impl/codegen/channel_interface.h>
  24 #include <grpcpp/impl/codegen/core_codegen_interface.h>
  25 #include <grpcpp/impl/codegen/server_context.h>
  26 #include <grpcpp/impl/codegen/service_type.h>
  27 #include <grpcpp/impl/codegen/status.h>
  28
  29 namespace grpc {
  30
  31 namespace internal {
  32 /// Common interface for all client side asynchronous streaming.
  33 class ClientAsyncStreamingInterface {
  34  public:
  35   virtual ~ClientAsyncStreamingInterface() {}
  36
  37   /// Start the call that was set up by the constructor, but only if the
  38   /// constructor was invoked through the "Prepare" API which doesn't actually
  39   /// start the call
  40   virtual void StartCall(void* tag) = 0;
  41
  42   /// Request notification of the reading of the initial metadata. Completion
  43   /// will be notified by \a tag on the associated completion queue.
  44   /// This call is optional, but if it is used, it cannot be used concurrently
  45   /// with or after the \a AsyncReaderInterface::Read method.
  46   ///
  47   /// \param[in] tag Tag identifying this request.
  48   virtual void ReadInitialMetadata(void* tag) = 0;
  49
  50   /// Indicate that the stream is to be finished and request notification for
  51   /// when the call has been ended.
  52   /// Should not be used concurrently with other operations.
  53   ///
  54   /// It is appropriate to call this method when both:
  55   ///   * the client side has no more message to send
  56   ///     (this can be declared implicitly by calling this method, or
  57   ///     explicitly through an earlier call to the <i>WritesDone</i> method
  58   ///     of the class in use, e.g. \a ClientAsyncWriterInterface::WritesDone or
  59   ///     \a ClientAsyncReaderWriterInterface::WritesDone).
  60   ///   * there are no more messages to be received from the server (this can
  61   ///     be known implicitly by the calling code, or explicitly from an
  62   ///     earlier call to \a AsyncReaderInterface::Read that yielded a failed
  63   ///     result, e.g. cq->Next(&read_tag, &ok) filled in 'ok' with 'false').
  64   ///
  65   /// The tag will be returned when either:
  66   /// - all incoming messages have been read and the server has returned
  67   ///   a status.
  68   /// - the server has returned a non-OK status.
  69   /// - the call failed for some reason and the library generated a
  70   ///   status.
  71   ///
  72   /// Note that implementations of this method attempt to receive initial
  73   /// metadata from the server if initial metadata hasn't yet been received.
  74   ///
  75   /// \param[in] tag Tag identifying this request.
  76   /// \param[out] status To be updated with the operation status.
  77   virtual void Finish(Status* status, void* tag) = 0;
  78 };
  79
  80 /// An interface that yields a sequence of messages of type \a R.
  81 template <class R>
  82 class AsyncReaderInterface {
  83  public:
  84   virtual ~AsyncReaderInterface() {}
  85
  86   /// Read a message of type \a R into \a msg. Completion will be notified by \a
  87   /// tag on the associated completion queue.
  88   /// This is thread-safe with respect to \a Write or \a WritesDone methods. It
  89   /// should not be called concurrently with other streaming APIs
  90   /// on the same stream. It is not meaningful to call it concurrently
  91   /// with another \a AsyncReaderInterface::Read on the same stream since reads
  92   /// on the same stream are delivered in order.
  93   ///
  94   /// \param[out] msg Where to eventually store the read message.
  95   /// \param[in] tag The tag identifying the operation.
  96   ///
  97   /// Side effect: note that this method attempt to receive initial metadata for
  98   /// a stream if it hasn't yet been received.
  99   virtual void Read(R* msg, void* tag) = 0;
 100 };
 101
 102 /// An interface that can be fed a sequence of messages of type \a W.
 103 template <class W>
 104 class AsyncWriterInterface {
 105  public:
 106   virtual ~AsyncWriterInterface() {}
 107
 108   /// Request the writing of \a msg with identifying tag \a tag.
 109   ///
 110   /// Only one write may be outstanding at any given time. This means that
 111   /// after calling Write, one must wait to receive \a tag from the completion
 112   /// queue BEFORE calling Write again.
 113   /// This is thread-safe with respect to \a AsyncReaderInterface::Read
 114   ///
 115   /// gRPC doesn't take ownership or a reference to \a msg, so it is safe to
 116   /// to deallocate once Write returns.
 117   ///
 118   /// \param[in] msg The message to be written.
 119   /// \param[in] tag The tag identifying the operation.
 120   virtual void Write(const W& msg, void* tag) = 0;
 121
 122   /// Request the writing of \a msg using WriteOptions \a options with
 123   /// identifying tag \a tag.
 124   ///
 125   /// Only one write may be outstanding at any given time. This means that
 126   /// after calling Write, one must wait to receive \a tag from the completion
 127   /// queue BEFORE calling Write again.
 128   /// WriteOptions \a options is used to set the write options of this message.
 129   /// This is thread-safe with respect to \a AsyncReaderInterface::Read
 130   ///
 131   /// gRPC doesn't take ownership or a reference to \a msg, so it is safe to
 132   /// to deallocate once Write returns.
 133   ///
 134   /// \param[in] msg The message to be written.
 135   /// \param[in] options The WriteOptions to be used to write this message.
 136   /// \param[in] tag The tag identifying the operation.
 137   virtual void Write(const W& msg, WriteOptions options, void* tag) = 0;
 138
 139   /// Request the writing of \a msg and coalesce it with the writing
 140   /// of trailing metadata, using WriteOptions \a options with
 141   /// identifying tag \a tag.
 142   ///
 143   /// For client, WriteLast is equivalent of performing Write and
 144   /// WritesDone in a single step.
 145   /// For server, WriteLast buffers the \a msg. The writing of \a msg is held
 146   /// until Finish is called, where \a msg and trailing metadata are coalesced
 147   /// and write is initiated. Note that WriteLast can only buffer \a msg up to
 148   /// the flow control window size. If \a msg size is larger than the window
 149   /// size, it will be sent on wire without buffering.
 150   ///
 151   /// gRPC doesn't take ownership or a reference to \a msg, so it is safe to
 152   /// to deallocate once Write returns.
 153   ///
 154   /// \param[in] msg The message to be written.
 155   /// \param[in] options The WriteOptions to be used to write this message.
 156   /// \param[in] tag The tag identifying the operation.
 157   void WriteLast(const W& msg, WriteOptions options, void* tag) {
 158     Write(msg, options.set_last_message(), tag);
 159   }
 160 };
 161
 162 }  // namespace internal
 163
 164 template <class R>
 165 class ClientAsyncReaderInterface
 166     : public internal::ClientAsyncStreamingInterface,
 167       public internal::AsyncReaderInterface<R> {};
 168
 169 namespace internal {
 170 template <class R>
 171 class ClientAsyncReaderFactory {
 172  public:
 173   /// Create a stream object.
 174   /// Write the first request out if \a start is set.
 175   /// \a tag will be notified on \a cq when the call has been started and
 176   /// \a request has been written out. If \a start is not set, \a tag must be
 177   /// nullptr and the actual call must be initiated by StartCall
 178   /// Note that \a context will be used to fill in custom initial metadata
 179   /// used to send to the server when starting the call.
 180   template <class W>
 181   static ClientAsyncReader<R>* Create(ChannelInterface* channel,
 182                                       CompletionQueue* cq,
 183                                       const ::grpc::internal::RpcMethod& method,
 184                                       ClientContext* context, const W& request,
 185                                       bool start, void* tag) {
 186     ::grpc::internal::Call call = channel->CreateCall(method, context, cq);
 187     return new (g_core_codegen_interface->grpc_call_arena_alloc(
 188         call.call(), sizeof(ClientAsyncReader<R>)))
 189         ClientAsyncReader<R>(call, context, request, start, tag);
 190   }
 191 };
 192 }  // namespace internal
 193
 194 /// Async client-side API for doing server-streaming RPCs,
 195 /// where the incoming message stream coming from the server has
 196 /// messages of type \a R.
 197 template <class R>
 198 class ClientAsyncReader final : public ClientAsyncReaderInterface<R> {
 199  public:
 200   // always allocated against a call arena, no memory free required
 201   static void operator delete(void* ptr, std::size_t size) {
 202     assert(size == sizeof(ClientAsyncReader));
 203   }
 204
 205   // This operator should never be called as the memory should be freed as part
 206   // of the arena destruction. It only exists to provide a matching operator
 207   // delete to the operator new so that some compilers will not complain (see
 208   // https://github.com/grpc/grpc/issues/11301) Note at the time of adding this
 209   // there are no tests catching the compiler warning.
 210   static void operator delete(void*, void*) { assert(0); }
 211
 212   void StartCall(void* tag) override {
 213     assert(!started_);
 214     started_ = true;
 215     StartCallInternal(tag);
 216   }
 217
 218   /// See the \a ClientAsyncStreamingInterface.ReadInitialMetadata
 219   /// method for semantics.
 220   ///
 221   /// Side effect:
 222   ///   - upon receiving initial metadata from the server,
 223   ///     the \a ClientContext associated with this call is updated, and the
 224   ///     calling code can access the received metadata through the
 225   ///     \a ClientContext.
 226   void ReadInitialMetadata(void* tag) override {
 227     assert(started_);
 228     GPR_CODEGEN_ASSERT(!context_->initial_metadata_received_);
 229
 230     meta_ops_.set_output_tag(tag);
 231     meta_ops_.RecvInitialMetadata(context_);
 232     call_.PerformOps(&meta_ops_);
 233   }
 234
 235   void Read(R* msg, void* tag) override {
 236     assert(started_);
 237     read_ops_.set_output_tag(tag);
 238     if (!context_->initial_metadata_received_) {
 239       read_ops_.RecvInitialMetadata(context_);
 240     }
 241     read_ops_.RecvMessage(msg);
 242     call_.PerformOps(&read_ops_);
 243   }
 244
 245   /// See the \a ClientAsyncStreamingInterface.Finish method for semantics.
 246   ///
 247   /// Side effect:
 248   ///   - the \a ClientContext associated with this call is updated with
 249   ///     possible initial and trailing metadata received from the server.
 250   void Finish(Status* status, void* tag) override {
 251     assert(started_);
 252     finish_ops_.set_output_tag(tag);
 253     if (!context_->initial_metadata_received_) {
 254       finish_ops_.RecvInitialMetadata(context_);
 255     }
 256     finish_ops_.ClientRecvStatus(context_, status);
 257     call_.PerformOps(&finish_ops_);
 258   }
 259
 260  private:
 261   friend class internal::ClientAsyncReaderFactory<R>;
 262   template <class W>
 263   ClientAsyncReader(::grpc::internal::Call call, ClientContext* context,
 264                     const W& request, bool start, void* tag)
 265       : context_(context), call_(call), started_(start) {
 266     // TODO(ctiller): don't assert
 267     GPR_CODEGEN_ASSERT(init_ops_.SendMessage(request).ok());
 268     init_ops_.ClientSendClose();
 269     if (start) {
 270       StartCallInternal(tag);
 271     } else {
 272       assert(tag == nullptr);
 273     }
 274   }
 275
 276   void StartCallInternal(void* tag) {
 277     init_ops_.SendInitialMetadata(&context_->send_initial_metadata_,
 278                                   context_->initial_metadata_flags());
 279     init_ops_.set_output_tag(tag);
 280     call_.PerformOps(&init_ops_);
 281   }
 282
 283   ClientContext* context_;
 284   ::grpc::internal::Call call_;
 285   bool started_;
 286   ::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata,
 287                               ::grpc::internal::CallOpSendMessage,
 288                               ::grpc::internal::CallOpClientSendClose>
 289       init_ops_;
 290   ::grpc::internal::CallOpSet<::grpc::internal::CallOpRecvInitialMetadata>
 291       meta_ops_;
 292   ::grpc::internal::CallOpSet<::grpc::internal::CallOpRecvInitialMetadata,
 293                               ::grpc::internal::CallOpRecvMessage<R>>
 294       read_ops_;
 295   ::grpc::internal::CallOpSet<::grpc::internal::CallOpRecvInitialMetadata,
 296                               ::grpc::internal::CallOpClientRecvStatus>
 297       finish_ops_;
 298 };
 299
 300 /// Common interface for client side asynchronous writing.
 301 template <class W>
 302 class ClientAsyncWriterInterface
 303     : public internal::ClientAsyncStreamingInterface,
 304       public internal::AsyncWriterInterface<W> {
 305  public:
 306   /// Signal the client is done with the writes (half-close the client stream).
 307   /// Thread-safe with respect to \a AsyncReaderInterface::Read
 308   ///
 309   /// \param[in] tag The tag identifying the operation.
 310   virtual void WritesDone(void* tag) = 0;
 311 };
 312
 313 namespace internal {
 314 template <class W>
 315 class ClientAsyncWriterFactory {
 316  public:
 317   /// Create a stream object.
 318   /// Start the RPC if \a start is set
 319   /// \a tag will be notified on \a cq when the call has been started (i.e.
 320   /// intitial metadata sent) and \a request has been written out.
 321   /// If \a start is not set, \a tag must be nullptr and the actual call
 322   /// must be initiated by StartCall
 323   /// Note that \a context will be used to fill in custom initial metadata
 324   /// used to send to the server when starting the call.
 325   /// \a response will be filled in with the single expected response
 326   /// message from the server upon a successful call to the \a Finish
 327   /// method of this instance.
 328   template <class R>
 329   static ClientAsyncWriter<W>* Create(ChannelInterface* channel,
 330                                       CompletionQueue* cq,
 331                                       const ::grpc::internal::RpcMethod& method,
 332                                       ClientContext* context, R* response,
 333                                       bool start, void* tag) {
 334     ::grpc::internal::Call call = channel->CreateCall(method, context, cq);
 335     return new (g_core_codegen_interface->grpc_call_arena_alloc(
 336         call.call(), sizeof(ClientAsyncWriter<W>)))
 337         ClientAsyncWriter<W>(call, context, response, start, tag);
 338   }
 339 };
 340 }  // namespace internal
 341
 342 /// Async API on the client side for doing client-streaming RPCs,
 343 /// where the outgoing message stream going to the server contains
 344 /// messages of type \a W.
 345 template <class W>
 346 class ClientAsyncWriter final : public ClientAsyncWriterInterface<W> {
 347  public:
 348   // always allocated against a call arena, no memory free required
 349   static void operator delete(void* ptr, std::size_t size) {
 350     assert(size == sizeof(ClientAsyncWriter));
 351   }
 352
 353   // This operator should never be called as the memory should be freed as part
 354   // of the arena destruction. It only exists to provide a matching operator
 355   // delete to the operator new so that some compilers will not complain (see
 356   // https://github.com/grpc/grpc/issues/11301) Note at the time of adding this
 357   // there are no tests catching the compiler warning.
 358   static void operator delete(void*, void*) { assert(0); }
 359
 360   void StartCall(void* tag) override {
 361     assert(!started_);
 362     started_ = true;
 363     StartCallInternal(tag);
 364   }
 365
 366   /// See the \a ClientAsyncStreamingInterface.ReadInitialMetadata method for
 367   /// semantics.
 368   ///
 369   /// Side effect:
 370   ///   - upon receiving initial metadata from the server, the \a ClientContext
 371   ///     associated with this call is updated, and the calling code can access
 372   ///     the received metadata through the \a ClientContext.
 373   void ReadInitialMetadata(void* tag) override {
 374     assert(started_);
 375     GPR_CODEGEN_ASSERT(!context_->initial_metadata_received_);
 376
 377     meta_ops_.set_output_tag(tag);
 378     meta_ops_.RecvInitialMetadata(context_);
 379     call_.PerformOps(&meta_ops_);
 380   }
 381
 382   void Write(const W& msg, void* tag) override {
 383     assert(started_);
 384     write_ops_.set_output_tag(tag);
 385     // TODO(ctiller): don't assert
 386     GPR_CODEGEN_ASSERT(write_ops_.SendMessage(msg).ok());
 387     call_.PerformOps(&write_ops_);
 388   }
 389
 390   void Write(const W& msg, WriteOptions options, void* tag) override {
 391     assert(started_);
 392     write_ops_.set_output_tag(tag);
 393     if (options.is_last_message()) {
 394       options.set_buffer_hint();
 395       write_ops_.ClientSendClose();
 396     }
 397     // TODO(ctiller): don't assert
 398     GPR_CODEGEN_ASSERT(write_ops_.SendMessage(msg, options).ok());
 399     call_.PerformOps(&write_ops_);
 400   }
 401
 402   void WritesDone(void* tag) override {
 403     assert(started_);
 404     write_ops_.set_output_tag(tag);
 405     write_ops_.ClientSendClose();
 406     call_.PerformOps(&write_ops_);
 407   }
 408
 409   /// See the \a ClientAsyncStreamingInterface.Finish method for semantics.
 410   ///
 411   /// Side effect:
 412   ///   - the \a ClientContext associated with this call is updated with
 413   ///     possible initial and trailing metadata received from the server.
 414   ///   - attempts to fill in the \a response parameter passed to this class's
 415   ///     constructor with the server's response message.
 416   void Finish(Status* status, void* tag) override {
 417     assert(started_);
 418     finish_ops_.set_output_tag(tag);
 419     if (!context_->initial_metadata_received_) {
 420       finish_ops_.RecvInitialMetadata(context_);
 421     }
 422     finish_ops_.ClientRecvStatus(context_, status);
 423     call_.PerformOps(&finish_ops_);
 424   }
 425
 426  private:
 427   friend class internal::ClientAsyncWriterFactory<W>;
 428   template <class R>
 429   ClientAsyncWriter(::grpc::internal::Call call, ClientContext* context,
 430                     R* response, bool start, void* tag)
 431       : context_(context), call_(call), started_(start) {
 432     finish_ops_.RecvMessage(response);
 433     finish_ops_.AllowNoMessage();
 434     if (start) {
 435       StartCallInternal(tag);
 436     } else {
 437       assert(tag == nullptr);
 438     }
 439   }
 440
 441   void StartCallInternal(void* tag) {
 442     write_ops_.SendInitialMetadata(&context_->send_initial_metadata_,
 443                                    context_->initial_metadata_flags());
 444     // if corked bit is set in context, we just keep the initial metadata
 445     // buffered up to coalesce with later message send. No op is performed.
 446     if (!context_->initial_metadata_corked_) {
 447       write_ops_.set_output_tag(tag);
 448       call_.PerformOps(&write_ops_);
 449     }
 450   }
 451
 452   ClientContext* context_;
 453   ::grpc::internal::Call call_;
 454   bool started_;
 455   ::grpc::internal::CallOpSet<::grpc::internal::CallOpRecvInitialMetadata>
 456       meta_ops_;
 457   ::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata,
 458                               ::grpc::internal::CallOpSendMessage,
 459                               ::grpc::internal::CallOpClientSendClose>
 460       write_ops_;
 461   ::grpc::internal::CallOpSet<::grpc::internal::CallOpRecvInitialMetadata,
 462                               ::grpc::internal::CallOpGenericRecvMessage,
 463                               ::grpc::internal::CallOpClientRecvStatus>
 464       finish_ops_;
 465 };
 466
 467 /// Async client-side interface for bi-directional streaming,
 468 /// where the client-to-server message stream has messages of type \a W,
 469 /// and the server-to-client message stream has messages of type \a R.
 470 template <class W, class R>
 471 class ClientAsyncReaderWriterInterface
 472     : public internal::ClientAsyncStreamingInterface,
 473       public internal::AsyncWriterInterface<W>,
 474       public internal::AsyncReaderInterface<R> {
 475  public:
 476   /// Signal the client is done with the writes (half-close the client stream).
 477   /// Thread-safe with respect to \a AsyncReaderInterface::Read
 478   ///
 479   /// \param[in] tag The tag identifying the operation.
 480   virtual void WritesDone(void* tag) = 0;
 481 };
 482
 483 namespace internal {
 484 template <class W, class R>
 485 class ClientAsyncReaderWriterFactory {
 486  public:
 487   /// Create a stream object.
 488   /// Start the RPC request if \a start is set.
 489   /// \a tag will be notified on \a cq when the call has been started (i.e.
 490   /// intitial metadata sent). If \a start is not set, \a tag must be
 491   /// nullptr and the actual call must be initiated by StartCall
 492   /// Note that \a context will be used to fill in custom initial metadata
 493   /// used to send to the server when starting the call.
 494   static ClientAsyncReaderWriter<W, R>* Create(
 495       ChannelInterface* channel, CompletionQueue* cq,
 496       const ::grpc::internal::RpcMethod& method, ClientContext* context,
 497       bool start, void* tag) {
 498     ::grpc::internal::Call call = channel->CreateCall(method, context, cq);
 499
 500     return new (g_core_codegen_interface->grpc_call_arena_alloc(
 501         call.call(), sizeof(ClientAsyncReaderWriter<W, R>)))
 502         ClientAsyncReaderWriter<W, R>(call, context, start, tag);
 503   }
 504 };
 505 }  // namespace internal
 506
 507 /// Async client-side interface for bi-directional streaming,
 508 /// where the outgoing message stream going to the server
 509 /// has messages of type \a W,  and the incoming message stream coming
 510 /// from the server has messages of type \a R.
 511 template <class W, class R>
 512 class ClientAsyncReaderWriter final
 513     : public ClientAsyncReaderWriterInterface<W, R> {
 514  public:
 515   // always allocated against a call arena, no memory free required
 516   static void operator delete(void* ptr, std::size_t size) {
 517     assert(size == sizeof(ClientAsyncReaderWriter));
 518   }
 519
 520   // This operator should never be called as the memory should be freed as part
 521   // of the arena destruction. It only exists to provide a matching operator
 522   // delete to the operator new so that some compilers will not complain (see
 523   // https://github.com/grpc/grpc/issues/11301) Note at the time of adding this
 524   // there are no tests catching the compiler warning.
 525   static void operator delete(void*, void*) { assert(0); }
 526
 527   void StartCall(void* tag) override {
 528     assert(!started_);
 529     started_ = true;
 530     StartCallInternal(tag);
 531   }
 532
 533   /// See the \a ClientAsyncStreamingInterface.ReadInitialMetadata method
 534   /// for semantics of this method.
 535   ///
 536   /// Side effect:
 537   ///   - upon receiving initial metadata from the server, the \a ClientContext
 538   ///     is updated with it, and then the receiving initial metadata can
 539   ///     be accessed through this \a ClientContext.
 540   void ReadInitialMetadata(void* tag) override {
 541     assert(started_);
 542     GPR_CODEGEN_ASSERT(!context_->initial_metadata_received_);
 543
 544     meta_ops_.set_output_tag(tag);
 545     meta_ops_.RecvInitialMetadata(context_);
 546     call_.PerformOps(&meta_ops_);
 547   }
 548
 549   void Read(R* msg, void* tag) override {
 550     assert(started_);
 551     read_ops_.set_output_tag(tag);
 552     if (!context_->initial_metadata_received_) {
 553       read_ops_.RecvInitialMetadata(context_);
 554     }
 555     read_ops_.RecvMessage(msg);
 556     call_.PerformOps(&read_ops_);
 557   }
 558
 559   void Write(const W& msg, void* tag) override {
 560     assert(started_);
 561     write_ops_.set_output_tag(tag);
 562     // TODO(ctiller): don't assert
 563     GPR_CODEGEN_ASSERT(write_ops_.SendMessage(msg).ok());
 564     call_.PerformOps(&write_ops_);
 565   }
 566
 567   void Write(const W& msg, WriteOptions options, void* tag) override {
 568     assert(started_);
 569     write_ops_.set_output_tag(tag);
 570     if (options.is_last_message()) {
 571       options.set_buffer_hint();
 572       write_ops_.ClientSendClose();
 573     }
 574     // TODO(ctiller): don't assert
 575     GPR_CODEGEN_ASSERT(write_ops_.SendMessage(msg, options).ok());
 576     call_.PerformOps(&write_ops_);
 577   }
 578
 579   void WritesDone(void* tag) override {
 580     assert(started_);
 581     write_ops_.set_output_tag(tag);
 582     write_ops_.ClientSendClose();
 583     call_.PerformOps(&write_ops_);
 584   }
 585
 586   /// See the \a ClientAsyncStreamingInterface.Finish method for semantics.
 587   /// Side effect
 588   ///   - the \a ClientContext associated with this call is updated with
 589   ///     possible initial and trailing metadata sent from the server.
 590   void Finish(Status* status, void* tag) override {
 591     assert(started_);
 592     finish_ops_.set_output_tag(tag);
 593     if (!context_->initial_metadata_received_) {
 594       finish_ops_.RecvInitialMetadata(context_);
 595     }
 596     finish_ops_.ClientRecvStatus(context_, status);
 597     call_.PerformOps(&finish_ops_);
 598   }
 599
 600  private:
 601   friend class internal::ClientAsyncReaderWriterFactory<W, R>;
 602   ClientAsyncReaderWriter(::grpc::internal::Call call, ClientContext* context,
 603                           bool start, void* tag)
 604       : context_(context), call_(call), started_(start) {
 605     if (start) {
 606       StartCallInternal(tag);
 607     } else {
 608       assert(tag == nullptr);
 609     }
 610   }
 611
 612   void StartCallInternal(void* tag) {
 613     write_ops_.SendInitialMetadata(&context_->send_initial_metadata_,
 614                                    context_->initial_metadata_flags());
 615     // if corked bit is set in context, we just keep the initial metadata
 616     // buffered up to coalesce with later message send. No op is performed.
 617     if (!context_->initial_metadata_corked_) {
 618       write_ops_.set_output_tag(tag);
 619       call_.PerformOps(&write_ops_);
 620     }
 621   }
 622
 623   ClientContext* context_;
 624   ::grpc::internal::Call call_;
 625   bool started_;
 626   ::grpc::internal::CallOpSet<::grpc::internal::CallOpRecvInitialMetadata>
 627       meta_ops_;
 628   ::grpc::internal::CallOpSet<::grpc::internal::CallOpRecvInitialMetadata,
 629                               ::grpc::internal::CallOpRecvMessage<R>>
 630       read_ops_;
 631   ::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata,
 632                               ::grpc::internal::CallOpSendMessage,
 633                               ::grpc::internal::CallOpClientSendClose>
 634       write_ops_;
 635   ::grpc::internal::CallOpSet<::grpc::internal::CallOpRecvInitialMetadata,
 636                               ::grpc::internal::CallOpClientRecvStatus>
 637       finish_ops_;
 638 };
 639
 640 template <class W, class R>
 641 class ServerAsyncReaderInterface
 642     : public internal::ServerAsyncStreamingInterface,
 643       public internal::AsyncReaderInterface<R> {
 644  public:
 645   /// Indicate that the stream is to be finished with a certain status code
 646   /// and also send out \a msg response to the client.
 647   /// Request notification for when the server has sent the response and the
 648   /// appropriate signals to the client to end the call.
 649   /// Should not be used concurrently with other operations.
 650   ///
 651   /// It is appropriate to call this method when:
 652   ///   * all messages from the client have been received (either known
 653   ///     implictly, or explicitly because a previous
 654   ///     \a AsyncReaderInterface::Read operation with a non-ok result,
 655   ///     e.g., cq->Next(&read_tag, &ok) filled in 'ok' with 'false').
 656   ///
 657   /// This operation will end when the server has finished sending out initial
 658   /// metadata (if not sent already), response message, and status, or if
 659   /// some failure occurred when trying to do so.
 660   ///
 661   /// gRPC doesn't take ownership or a reference to \a msg or \a status, so it
 662   /// is safe to deallocate once Finish returns.
 663   ///
 664   /// \param[in] tag Tag identifying this request.
 665   /// \param[in] status To be sent to the client as the result of this call.
 666   /// \param[in] msg To be sent to the client as the response for this call.
 667   virtual void Finish(const W& msg, const Status& status, void* tag) = 0;
 668
 669   /// Indicate that the stream is to be finished with a certain
 670   /// non-OK status code.
 671   /// Request notification for when the server has sent the appropriate
 672   /// signals to the client to end the call.
 673   /// Should not be used concurrently with other operations.
 674   ///
 675   /// This call is meant to end the call with some error, and can be called at
 676   /// any point that the server would like to "fail" the call (though note
 677   /// this shouldn't be called concurrently with any other "sending" call, like
 678   /// \a AsyncWriterInterface::Write).
 679   ///
 680   /// This operation will end when the server has finished sending out initial
 681   /// metadata (if not sent already), and status, or if some failure occurred
 682   /// when trying to do so.
 683   ///
 684   /// gRPC doesn't take ownership or a reference to \a status, so it is safe to
 685   /// to deallocate once FinishWithError returns.
 686   ///
 687   /// \param[in] tag Tag identifying this request.
 688   /// \param[in] status To be sent to the client as the result of this call.
 689   ///     - Note: \a status must have a non-OK code.
 690   virtual void FinishWithError(const Status& status, void* tag) = 0;
 691 };
 692
 693 /// Async server-side API for doing client-streaming RPCs,
 694 /// where the incoming message stream from the client has messages of type \a R,
 695 /// and the single response message sent from the server is type \a W.
 696 template <class W, class R>
 697 class ServerAsyncReader final : public ServerAsyncReaderInterface<W, R> {
 698  public:
 699   explicit ServerAsyncReader(ServerContext* ctx)
 700       : call_(nullptr, nullptr, nullptr), ctx_(ctx) {}
 701
 702   /// See \a ServerAsyncStreamingInterface::SendInitialMetadata for semantics.
 703   ///
 704   /// Implicit input parameter:
 705   ///   - The initial metadata that will be sent to the client from this op will
 706   ///     be taken from the \a ServerContext associated with the call.
 707   void SendInitialMetadata(void* tag) override {
 708     GPR_CODEGEN_ASSERT(!ctx_->sent_initial_metadata_);
 709
 710     meta_ops_.set_output_tag(tag);
 711     meta_ops_.SendInitialMetadata(&ctx_->initial_metadata_,
 712                                   ctx_->initial_metadata_flags());
 713     if (ctx_->compression_level_set()) {
 714       meta_ops_.set_compression_level(ctx_->compression_level());
 715     }
 716     ctx_->sent_initial_metadata_ = true;
 717     call_.PerformOps(&meta_ops_);
 718   }
 719
 720   void Read(R* msg, void* tag) override {
 721     read_ops_.set_output_tag(tag);
 722     read_ops_.RecvMessage(msg);
 723     call_.PerformOps(&read_ops_);
 724   }
 725
 726   /// See the \a ServerAsyncReaderInterface.Read method for semantics
 727   ///
 728   /// Side effect:
 729   ///   - also sends initial metadata if not alreay sent.
 730   ///   - uses the \a ServerContext associated with this call to send possible
 731   ///     initial and trailing metadata.
 732   ///
 733   /// Note: \a msg is not sent if \a status has a non-OK code.
 734   ///
 735   /// gRPC doesn't take ownership or a reference to \a msg and \a status, so it
 736   /// is safe to deallocate once Finish returns.
 737   void Finish(const W& msg, const Status& status, void* tag) override {
 738     finish_ops_.set_output_tag(tag);
 739     if (!ctx_->sent_initial_metadata_) {
 740       finish_ops_.SendInitialMetadata(&ctx_->initial_metadata_,
 741                                       ctx_->initial_metadata_flags());
 742       if (ctx_->compression_level_set()) {
 743         finish_ops_.set_compression_level(ctx_->compression_level());
 744       }
 745       ctx_->sent_initial_metadata_ = true;
 746     }
 747     // The response is dropped if the status is not OK.
 748     if (status.ok()) {
 749       finish_ops_.ServerSendStatus(&ctx_->trailing_metadata_,
 750                                    finish_ops_.SendMessage(msg));
 751     } else {
 752       finish_ops_.ServerSendStatus(&ctx_->trailing_metadata_, status);
 753     }
 754     call_.PerformOps(&finish_ops_);
 755   }
 756
 757   /// See the \a ServerAsyncReaderInterface.Read method for semantics
 758   ///
 759   /// Side effect:
 760   ///   - also sends initial metadata if not alreay sent.
 761   ///   - uses the \a ServerContext associated with this call to send possible
 762   ///     initial and trailing metadata.
 763   ///
 764   /// gRPC doesn't take ownership or a reference to \a status, so it is safe to
 765   /// to deallocate once FinishWithError returns.
 766   void FinishWithError(const Status& status, void* tag) override {
 767     GPR_CODEGEN_ASSERT(!status.ok());
 768     finish_ops_.set_output_tag(tag);
 769     if (!ctx_->sent_initial_metadata_) {
 770       finish_ops_.SendInitialMetadata(&ctx_->initial_metadata_,
 771                                       ctx_->initial_metadata_flags());
 772       if (ctx_->compression_level_set()) {
 773         finish_ops_.set_compression_level(ctx_->compression_level());
 774       }
 775       ctx_->sent_initial_metadata_ = true;
 776     }
 777     finish_ops_.ServerSendStatus(&ctx_->trailing_metadata_, status);
 778     call_.PerformOps(&finish_ops_);
 779   }
 780
 781  private:
 782   void BindCall(::grpc::internal::Call* call) override { call_ = *call; }
 783
 784   ::grpc::internal::Call call_;
 785   ServerContext* ctx_;
 786   ::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata>
 787       meta_ops_;
 788   ::grpc::internal::CallOpSet<::grpc::internal::CallOpRecvMessage<R>> read_ops_;
 789   ::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata,
 790                               ::grpc::internal::CallOpSendMessage,
 791                               ::grpc::internal::CallOpServerSendStatus>
 792       finish_ops_;
 793 };
 794
 795 template <class W>
 796 class ServerAsyncWriterInterface
 797     : public internal::ServerAsyncStreamingInterface,
 798       public internal::AsyncWriterInterface<W> {
 799  public:
 800   /// Indicate that the stream is to be finished with a certain status code.
 801   /// Request notification for when the server has sent the appropriate
 802   /// signals to the client to end the call.
 803   /// Should not be used concurrently with other operations.
 804   ///
 805   /// It is appropriate to call this method when either:
 806   ///   * all messages from the client have been received (either known
 807   ///     implictly, or explicitly because a previous \a
 808   ///     AsyncReaderInterface::Read operation with a non-ok
 809   ///     result (e.g., cq->Next(&read_tag, &ok) filled in 'ok' with 'false'.
 810   ///   * it is desired to end the call early with some non-OK status code.
 811   ///
 812   /// This operation will end when the server has finished sending out initial
 813   /// metadata (if not sent already), response message, and status, or if
 814   /// some failure occurred when trying to do so.
 815   ///
 816   /// gRPC doesn't take ownership or a reference to \a status, so it is safe to
 817   /// to deallocate once Finish returns.
 818   ///
 819   /// \param[in] tag Tag identifying this request.
 820   /// \param[in] status To be sent to the client as the result of this call.
 821   virtual void Finish(const Status& status, void* tag) = 0;
 822
 823   /// Request the writing of \a msg and coalesce it with trailing metadata which
 824   /// contains \a status, using WriteOptions options with
 825   /// identifying tag \a tag.
 826   ///
 827   /// WriteAndFinish is equivalent of performing WriteLast and Finish
 828   /// in a single step.
 829   ///
 830   /// gRPC doesn't take ownership or a reference to \a msg and \a status, so it
 831   /// is safe to deallocate once WriteAndFinish returns.
 832   ///
 833   /// \param[in] msg The message to be written.
 834   /// \param[in] options The WriteOptions to be used to write this message.
 835   /// \param[in] status The Status that server returns to client.
 836   /// \param[in] tag The tag identifying the operation.
 837   virtual void WriteAndFinish(const W& msg, WriteOptions options,
 838                               const Status& status, void* tag) = 0;
 839 };
 840
 841 /// Async server-side API for doing server streaming RPCs,
 842 /// where the outgoing message stream from the server has messages of type \a W.
 843 template <class W>
 844 class ServerAsyncWriter final : public ServerAsyncWriterInterface<W> {
 845  public:
 846   explicit ServerAsyncWriter(ServerContext* ctx)
 847       : call_(nullptr, nullptr, nullptr), ctx_(ctx) {}
 848
 849   /// See \a ServerAsyncStreamingInterface::SendInitialMetadata for semantics.
 850   ///
 851   /// Implicit input parameter:
 852   ///   - The initial metadata that will be sent to the client from this op will
 853   ///     be taken from the \a ServerContext associated with the call.
 854   ///
 855   /// \param[in] tag Tag identifying this request.
 856   void SendInitialMetadata(void* tag) override {
 857     GPR_CODEGEN_ASSERT(!ctx_->sent_initial_metadata_);
 858
 859     meta_ops_.set_output_tag(tag);
 860     meta_ops_.SendInitialMetadata(&ctx_->initial_metadata_,
 861                                   ctx_->initial_metadata_flags());
 862     if (ctx_->compression_level_set()) {
 863       meta_ops_.set_compression_level(ctx_->compression_level());
 864     }
 865     ctx_->sent_initial_metadata_ = true;
 866     call_.PerformOps(&meta_ops_);
 867   }
 868
 869   void Write(const W& msg, void* tag) override {
 870     write_ops_.set_output_tag(tag);
 871     EnsureInitialMetadataSent(&write_ops_);
 872     // TODO(ctiller): don't assert
 873     GPR_CODEGEN_ASSERT(write_ops_.SendMessage(msg).ok());
 874     call_.PerformOps(&write_ops_);
 875   }
 876
 877   void Write(const W& msg, WriteOptions options, void* tag) override {
 878     write_ops_.set_output_tag(tag);
 879     if (options.is_last_message()) {
 880       options.set_buffer_hint();
 881     }
 882
 883     EnsureInitialMetadataSent(&write_ops_);
 884     // TODO(ctiller): don't assert
 885     GPR_CODEGEN_ASSERT(write_ops_.SendMessage(msg, options).ok());
 886     call_.PerformOps(&write_ops_);
 887   }
 888
 889   /// See the \a ServerAsyncWriterInterface.WriteAndFinish method for semantics.
 890   ///
 891   /// Implicit input parameter:
 892   ///   - the \a ServerContext associated with this call is used
 893   ///     for sending trailing (and initial) metadata to the client.
 894   ///
 895   /// Note: \a status must have an OK code.
 896   ///
 897   /// gRPC doesn't take ownership or a reference to \a msg and \a status, so it
 898   /// is safe to deallocate once WriteAndFinish returns.
 899   void WriteAndFinish(const W& msg, WriteOptions options, const Status& status,
 900                       void* tag) override {
 901     write_ops_.set_output_tag(tag);
 902     EnsureInitialMetadataSent(&write_ops_);
 903     options.set_buffer_hint();
 904     GPR_CODEGEN_ASSERT(write_ops_.SendMessage(msg, options).ok());
 905     write_ops_.ServerSendStatus(&ctx_->trailing_metadata_, status);
 906     call_.PerformOps(&write_ops_);
 907   }
 908
 909   /// See the \a ServerAsyncWriterInterface.Finish method for semantics.
 910   ///
 911   /// Implicit input parameter:
 912   ///   - the \a ServerContext associated with this call is used for sending
 913   ///     trailing (and initial if not already sent) metadata to the client.
 914   ///
 915   /// Note: there are no restrictions are the code of
 916   /// \a status,it may be non-OK
 917   ///
 918   /// gRPC doesn't take ownership or a reference to \a status, so it is safe to
 919   /// to deallocate once Finish returns.
 920   void Finish(const Status& status, void* tag) override {
 921     finish_ops_.set_output_tag(tag);
 922     EnsureInitialMetadataSent(&finish_ops_);
 923     finish_ops_.ServerSendStatus(&ctx_->trailing_metadata_, status);
 924     call_.PerformOps(&finish_ops_);
 925   }
 926
 927  private:
 928   void BindCall(::grpc::internal::Call* call) override { call_ = *call; }
 929
 930   template <class T>
 931   void EnsureInitialMetadataSent(T* ops) {
 932     if (!ctx_->sent_initial_metadata_) {
 933       ops->SendInitialMetadata(&ctx_->initial_metadata_,
 934                                ctx_->initial_metadata_flags());
 935       if (ctx_->compression_level_set()) {
 936         ops->set_compression_level(ctx_->compression_level());
 937       }
 938       ctx_->sent_initial_metadata_ = true;
 939     }
 940   }
 941
 942   ::grpc::internal::Call call_;
 943   ServerContext* ctx_;
 944   ::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata>
 945       meta_ops_;
 946   ::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata,
 947                               ::grpc::internal::CallOpSendMessage,
 948                               ::grpc::internal::CallOpServerSendStatus>
 949       write_ops_;
 950   ::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata,
 951                               ::grpc::internal::CallOpServerSendStatus>
 952       finish_ops_;
 953 };
 954
 955 /// Server-side interface for asynchronous bi-directional streaming.
 956 template <class W, class R>
 957 class ServerAsyncReaderWriterInterface
 958     : public internal::ServerAsyncStreamingInterface,
 959       public internal::AsyncWriterInterface<W>,
 960       public internal::AsyncReaderInterface<R> {
 961  public:
 962   /// Indicate that the stream is to be finished with a certain status code.
 963   /// Request notification for when the server has sent the appropriate
 964   /// signals to the client to end the call.
 965   /// Should not be used concurrently with other operations.
 966   ///
 967   /// It is appropriate to call this method when either:
 968   ///   * all messages from the client have been received (either known
 969   ///     implictly, or explicitly because a previous \a
 970   ///     AsyncReaderInterface::Read operation
 971   ///     with a non-ok result (e.g., cq->Next(&read_tag, &ok) filled in 'ok'
 972   ///     with 'false'.
 973   ///   * it is desired to end the call early with some non-OK status code.
 974   ///
 975   /// This operation will end when the server has finished sending out initial
 976   /// metadata (if not sent already), response message, and status, or if some
 977   /// failure occurred when trying to do so.
 978   ///
 979   /// gRPC doesn't take ownership or a reference to \a status, so it is safe to
 980   /// to deallocate once Finish returns.
 981   ///
 982   /// \param[in] tag Tag identifying this request.
 983   /// \param[in] status To be sent to the client as the result of this call.
 984   virtual void Finish(const Status& status, void* tag) = 0;
 985
 986   /// Request the writing of \a msg and coalesce it with trailing metadata which
 987   /// contains \a status, using WriteOptions options with
 988   /// identifying tag \a tag.
 989   ///
 990   /// WriteAndFinish is equivalent of performing WriteLast and Finish in a
 991   /// single step.
 992   ///
 993   /// gRPC doesn't take ownership or a reference to \a msg and \a status, so it
 994   /// is safe to deallocate once WriteAndFinish returns.
 995   ///
 996   /// \param[in] msg The message to be written.
 997   /// \param[in] options The WriteOptions to be used to write this message.
 998   /// \param[in] status The Status that server returns to client.
 999   /// \param[in] tag The tag identifying the operation.
1000   virtual void WriteAndFinish(const W& msg, WriteOptions options,
1001                               const Status& status, void* tag) = 0;
1002 };
1003
1004 /// Async server-side API for doing bidirectional streaming RPCs,
1005 /// where the incoming message stream coming from the client has messages of
1006 /// type \a R, and the outgoing message stream coming from the server has
1007 /// messages of type \a W.
1008 template <class W, class R>
1009 class ServerAsyncReaderWriter final
1010     : public ServerAsyncReaderWriterInterface<W, R> {
1011  public:
1012   explicit ServerAsyncReaderWriter(ServerContext* ctx)
1013       : call_(nullptr, nullptr, nullptr), ctx_(ctx) {}
1014
1015   /// See \a ServerAsyncStreamingInterface::SendInitialMetadata for semantics.
1016   ///
1017   /// Implicit input parameter:
1018   ///   - The initial metadata that will be sent to the client from this op will
1019   ///     be taken from the \a ServerContext associated with the call.
1020   ///
1021   /// \param[in] tag Tag identifying this request.
1022   void SendInitialMetadata(void* tag) override {
1023     GPR_CODEGEN_ASSERT(!ctx_->sent_initial_metadata_);
1024
1025     meta_ops_.set_output_tag(tag);
1026     meta_ops_.SendInitialMetadata(&ctx_->initial_metadata_,
1027                                   ctx_->initial_metadata_flags());
1028     if (ctx_->compression_level_set()) {
1029       meta_ops_.set_compression_level(ctx_->compression_level());
1030     }
1031     ctx_->sent_initial_metadata_ = true;
1032     call_.PerformOps(&meta_ops_);
1033   }
1034
1035   void Read(R* msg, void* tag) override {
1036     read_ops_.set_output_tag(tag);
1037     read_ops_.RecvMessage(msg);
1038     call_.PerformOps(&read_ops_);
1039   }
1040
1041   void Write(const W& msg, void* tag) override {
1042     write_ops_.set_output_tag(tag);
1043     EnsureInitialMetadataSent(&write_ops_);
1044     // TODO(ctiller): don't assert
1045     GPR_CODEGEN_ASSERT(write_ops_.SendMessage(msg).ok());
1046     call_.PerformOps(&write_ops_);
1047   }
1048
1049   void Write(const W& msg, WriteOptions options, void* tag) override {
1050     write_ops_.set_output_tag(tag);
1051     if (options.is_last_message()) {
1052       options.set_buffer_hint();
1053     }
1054     EnsureInitialMetadataSent(&write_ops_);
1055     GPR_CODEGEN_ASSERT(write_ops_.SendMessage(msg, options).ok());
1056     call_.PerformOps(&write_ops_);
1057   }
1058
1059   /// See the \a ServerAsyncReaderWriterInterface.WriteAndFinish
1060   /// method for semantics.
1061   ///
1062   /// Implicit input parameter:
1063   ///   - the \a ServerContext associated with this call is used
1064   ///     for sending trailing (and initial) metadata to the client.
1065   ///
1066   /// Note: \a status must have an OK code.
1067   //
1068   /// gRPC doesn't take ownership or a reference to \a msg and \a status, so it
1069   /// is safe to deallocate once WriteAndFinish returns.
1070   void WriteAndFinish(const W& msg, WriteOptions options, const Status& status,
1071                       void* tag) override {
1072     write_ops_.set_output_tag(tag);
1073     EnsureInitialMetadataSent(&write_ops_);
1074     options.set_buffer_hint();
1075     GPR_CODEGEN_ASSERT(write_ops_.SendMessage(msg, options).ok());
1076     write_ops_.ServerSendStatus(&ctx_->trailing_metadata_, status);
1077     call_.PerformOps(&write_ops_);
1078   }
1079
1080   /// See the \a ServerAsyncReaderWriterInterface.Finish method for semantics.
1081   ///
1082   /// Implicit input parameter:
1083   ///   - the \a ServerContext associated with this call is used for sending
1084   ///     trailing (and initial if not already sent) metadata to the client.
1085   ///
1086   /// Note: there are no restrictions are the code of \a status,
1087   /// it may be non-OK
1088   //
1089   /// gRPC doesn't take ownership or a reference to \a status, so it is safe to
1090   /// to deallocate once Finish returns.
1091   void Finish(const Status& status, void* tag) override {
1092     finish_ops_.set_output_tag(tag);
1093     EnsureInitialMetadataSent(&finish_ops_);
1094
1095     finish_ops_.ServerSendStatus(&ctx_->trailing_metadata_, status);
1096     call_.PerformOps(&finish_ops_);
1097   }
1098
1099  private:
1100   friend class ::grpc_impl::Server;
1101
1102   void BindCall(::grpc::internal::Call* call) override { call_ = *call; }
1103
1104   template <class T>
1105   void EnsureInitialMetadataSent(T* ops) {
1106     if (!ctx_->sent_initial_metadata_) {
1107       ops->SendInitialMetadata(&ctx_->initial_metadata_,
1108                                ctx_->initial_metadata_flags());
1109       if (ctx_->compression_level_set()) {
1110         ops->set_compression_level(ctx_->compression_level());
1111       }
1112       ctx_->sent_initial_metadata_ = true;
1113     }
1114   }
1115
1116   ::grpc::internal::Call call_;
1117   ServerContext* ctx_;
1118   ::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata>
1119       meta_ops_;
1120   ::grpc::internal::CallOpSet<::grpc::internal::CallOpRecvMessage<R>> read_ops_;
1121   ::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata,
1122                               ::grpc::internal::CallOpSendMessage,
1123                               ::grpc::internal::CallOpServerSendStatus>
1124       write_ops_;
1125   ::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata,
1126                               ::grpc::internal::CallOpServerSendStatus>
1127       finish_ops_;
1128 };
1129
1130 }  // namespace grpc
1131
1132 #endif  // GRPCPP_IMPL_CODEGEN_ASYNC_STREAM_H