3 * Copyright 2015 gRPC authors.
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
9 * http://www.apache.org/licenses/LICENSE-2.0
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.
19 #ifndef GRPCPP_IMPL_CODEGEN_ASYNC_STREAM_H
20 #define GRPCPP_IMPL_CODEGEN_ASYNC_STREAM_H
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>
32 /// Common interface for all client side asynchronous streaming.
33 class ClientAsyncStreamingInterface {
35 virtual ~ClientAsyncStreamingInterface() {}
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
40 virtual void StartCall(void* tag) = 0;
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.
47 /// \param[in] tag Tag identifying this request.
48 virtual void ReadInitialMetadata(void* tag) = 0;
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.
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').
65 /// The tag will be returned when either:
66 /// - all incoming messages have been read and the server has returned
68 /// - the server has returned a non-OK status.
69 /// - the call failed for some reason and the library generated a
72 /// Note that implementations of this method attempt to receive initial
73 /// metadata from the server if initial metadata hasn't yet been received.
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;
80 /// An interface that yields a sequence of messages of type \a R.
82 class AsyncReaderInterface {
84 virtual ~AsyncReaderInterface() {}
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.
94 /// \param[out] msg Where to eventually store the read message.
95 /// \param[in] tag The tag identifying the operation.
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;
102 /// An interface that can be fed a sequence of messages of type \a W.
104 class AsyncWriterInterface {
106 virtual ~AsyncWriterInterface() {}
108 /// Request the writing of \a msg with identifying tag \a tag.
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
115 /// gRPC doesn't take ownership or a reference to \a msg, so it is safe to
116 /// to deallocate once Write returns.
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;
122 /// Request the writing of \a msg using WriteOptions \a options with
123 /// identifying tag \a tag.
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
131 /// gRPC doesn't take ownership or a reference to \a msg, so it is safe to
132 /// to deallocate once Write returns.
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;
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.
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.
151 /// gRPC doesn't take ownership or a reference to \a msg, so it is safe to
152 /// to deallocate once Write returns.
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);
162 } // namespace internal
165 class ClientAsyncReaderInterface
166 : public internal::ClientAsyncStreamingInterface,
167 public internal::AsyncReaderInterface<R> {};
171 class ClientAsyncReaderFactory {
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.
181 static ClientAsyncReader<R>* Create(ChannelInterface* channel,
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);
192 } // namespace internal
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.
198 class ClientAsyncReader final : public ClientAsyncReaderInterface<R> {
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));
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); }
212 void StartCall(void* tag) override {
215 StartCallInternal(tag);
218 /// See the \a ClientAsyncStreamingInterface.ReadInitialMetadata
219 /// method for semantics.
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 {
228 GPR_CODEGEN_ASSERT(!context_->initial_metadata_received_);
230 meta_ops_.set_output_tag(tag);
231 meta_ops_.RecvInitialMetadata(context_);
232 call_.PerformOps(&meta_ops_);
235 void Read(R* msg, void* tag) override {
237 read_ops_.set_output_tag(tag);
238 if (!context_->initial_metadata_received_) {
239 read_ops_.RecvInitialMetadata(context_);
241 read_ops_.RecvMessage(msg);
242 call_.PerformOps(&read_ops_);
245 /// See the \a ClientAsyncStreamingInterface.Finish method for semantics.
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 {
252 finish_ops_.set_output_tag(tag);
253 if (!context_->initial_metadata_received_) {
254 finish_ops_.RecvInitialMetadata(context_);
256 finish_ops_.ClientRecvStatus(context_, status);
257 call_.PerformOps(&finish_ops_);
261 friend class internal::ClientAsyncReaderFactory<R>;
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();
270 StartCallInternal(tag);
272 assert(tag == nullptr);
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_);
283 ClientContext* context_;
284 ::grpc::internal::Call call_;
286 ::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata,
287 ::grpc::internal::CallOpSendMessage,
288 ::grpc::internal::CallOpClientSendClose>
290 ::grpc::internal::CallOpSet<::grpc::internal::CallOpRecvInitialMetadata>
292 ::grpc::internal::CallOpSet<::grpc::internal::CallOpRecvInitialMetadata,
293 ::grpc::internal::CallOpRecvMessage<R>>
295 ::grpc::internal::CallOpSet<::grpc::internal::CallOpRecvInitialMetadata,
296 ::grpc::internal::CallOpClientRecvStatus>
300 /// Common interface for client side asynchronous writing.
302 class ClientAsyncWriterInterface
303 : public internal::ClientAsyncStreamingInterface,
304 public internal::AsyncWriterInterface<W> {
306 /// Signal the client is done with the writes (half-close the client stream).
307 /// Thread-safe with respect to \a AsyncReaderInterface::Read
309 /// \param[in] tag The tag identifying the operation.
310 virtual void WritesDone(void* tag) = 0;
315 class ClientAsyncWriterFactory {
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.
329 static ClientAsyncWriter<W>* Create(ChannelInterface* channel,
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);
340 } // namespace internal
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.
346 class ClientAsyncWriter final : public ClientAsyncWriterInterface<W> {
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));
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); }
360 void StartCall(void* tag) override {
363 StartCallInternal(tag);
366 /// See the \a ClientAsyncStreamingInterface.ReadInitialMetadata method for
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 {
375 GPR_CODEGEN_ASSERT(!context_->initial_metadata_received_);
377 meta_ops_.set_output_tag(tag);
378 meta_ops_.RecvInitialMetadata(context_);
379 call_.PerformOps(&meta_ops_);
382 void Write(const W& msg, void* tag) override {
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_);
390 void Write(const W& msg, WriteOptions options, void* tag) override {
392 write_ops_.set_output_tag(tag);
393 if (options.is_last_message()) {
394 options.set_buffer_hint();
395 write_ops_.ClientSendClose();
397 // TODO(ctiller): don't assert
398 GPR_CODEGEN_ASSERT(write_ops_.SendMessage(msg, options).ok());
399 call_.PerformOps(&write_ops_);
402 void WritesDone(void* tag) override {
404 write_ops_.set_output_tag(tag);
405 write_ops_.ClientSendClose();
406 call_.PerformOps(&write_ops_);
409 /// See the \a ClientAsyncStreamingInterface.Finish method for semantics.
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 {
418 finish_ops_.set_output_tag(tag);
419 if (!context_->initial_metadata_received_) {
420 finish_ops_.RecvInitialMetadata(context_);
422 finish_ops_.ClientRecvStatus(context_, status);
423 call_.PerformOps(&finish_ops_);
427 friend class internal::ClientAsyncWriterFactory<W>;
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();
435 StartCallInternal(tag);
437 assert(tag == nullptr);
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_);
452 ClientContext* context_;
453 ::grpc::internal::Call call_;
455 ::grpc::internal::CallOpSet<::grpc::internal::CallOpRecvInitialMetadata>
457 ::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata,
458 ::grpc::internal::CallOpSendMessage,
459 ::grpc::internal::CallOpClientSendClose>
461 ::grpc::internal::CallOpSet<::grpc::internal::CallOpRecvInitialMetadata,
462 ::grpc::internal::CallOpGenericRecvMessage,
463 ::grpc::internal::CallOpClientRecvStatus>
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> {
476 /// Signal the client is done with the writes (half-close the client stream).
477 /// Thread-safe with respect to \a AsyncReaderInterface::Read
479 /// \param[in] tag The tag identifying the operation.
480 virtual void WritesDone(void* tag) = 0;
484 template <class W, class R>
485 class ClientAsyncReaderWriterFactory {
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);
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);
505 } // namespace internal
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> {
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));
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); }
527 void StartCall(void* tag) override {
530 StartCallInternal(tag);
533 /// See the \a ClientAsyncStreamingInterface.ReadInitialMetadata method
534 /// for semantics of this method.
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 {
542 GPR_CODEGEN_ASSERT(!context_->initial_metadata_received_);
544 meta_ops_.set_output_tag(tag);
545 meta_ops_.RecvInitialMetadata(context_);
546 call_.PerformOps(&meta_ops_);
549 void Read(R* msg, void* tag) override {
551 read_ops_.set_output_tag(tag);
552 if (!context_->initial_metadata_received_) {
553 read_ops_.RecvInitialMetadata(context_);
555 read_ops_.RecvMessage(msg);
556 call_.PerformOps(&read_ops_);
559 void Write(const W& msg, void* tag) override {
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_);
567 void Write(const W& msg, WriteOptions options, void* tag) override {
569 write_ops_.set_output_tag(tag);
570 if (options.is_last_message()) {
571 options.set_buffer_hint();
572 write_ops_.ClientSendClose();
574 // TODO(ctiller): don't assert
575 GPR_CODEGEN_ASSERT(write_ops_.SendMessage(msg, options).ok());
576 call_.PerformOps(&write_ops_);
579 void WritesDone(void* tag) override {
581 write_ops_.set_output_tag(tag);
582 write_ops_.ClientSendClose();
583 call_.PerformOps(&write_ops_);
586 /// See the \a ClientAsyncStreamingInterface.Finish method for semantics.
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 {
592 finish_ops_.set_output_tag(tag);
593 if (!context_->initial_metadata_received_) {
594 finish_ops_.RecvInitialMetadata(context_);
596 finish_ops_.ClientRecvStatus(context_, status);
597 call_.PerformOps(&finish_ops_);
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) {
606 StartCallInternal(tag);
608 assert(tag == nullptr);
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_);
623 ClientContext* context_;
624 ::grpc::internal::Call call_;
626 ::grpc::internal::CallOpSet<::grpc::internal::CallOpRecvInitialMetadata>
628 ::grpc::internal::CallOpSet<::grpc::internal::CallOpRecvInitialMetadata,
629 ::grpc::internal::CallOpRecvMessage<R>>
631 ::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata,
632 ::grpc::internal::CallOpSendMessage,
633 ::grpc::internal::CallOpClientSendClose>
635 ::grpc::internal::CallOpSet<::grpc::internal::CallOpRecvInitialMetadata,
636 ::grpc::internal::CallOpClientRecvStatus>
640 template <class W, class R>
641 class ServerAsyncReaderInterface
642 : public internal::ServerAsyncStreamingInterface,
643 public internal::AsyncReaderInterface<R> {
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.
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').
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.
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.
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;
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.
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).
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.
684 /// gRPC doesn't take ownership or a reference to \a status, so it is safe to
685 /// to deallocate once FinishWithError returns.
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;
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> {
699 explicit ServerAsyncReader(ServerContext* ctx)
700 : call_(nullptr, nullptr, nullptr), ctx_(ctx) {}
702 /// See \a ServerAsyncStreamingInterface::SendInitialMetadata for semantics.
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_);
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());
716 ctx_->sent_initial_metadata_ = true;
717 call_.PerformOps(&meta_ops_);
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_);
726 /// See the \a ServerAsyncReaderInterface.Read method for semantics
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.
733 /// Note: \a msg is not sent if \a status has a non-OK code.
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());
745 ctx_->sent_initial_metadata_ = true;
747 // The response is dropped if the status is not OK.
749 finish_ops_.ServerSendStatus(&ctx_->trailing_metadata_,
750 finish_ops_.SendMessage(msg));
752 finish_ops_.ServerSendStatus(&ctx_->trailing_metadata_, status);
754 call_.PerformOps(&finish_ops_);
757 /// See the \a ServerAsyncReaderInterface.Read method for semantics
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.
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());
775 ctx_->sent_initial_metadata_ = true;
777 finish_ops_.ServerSendStatus(&ctx_->trailing_metadata_, status);
778 call_.PerformOps(&finish_ops_);
782 void BindCall(::grpc::internal::Call* call) override { call_ = *call; }
784 ::grpc::internal::Call call_;
786 ::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata>
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>
796 class ServerAsyncWriterInterface
797 : public internal::ServerAsyncStreamingInterface,
798 public internal::AsyncWriterInterface<W> {
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.
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.
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.
816 /// gRPC doesn't take ownership or a reference to \a status, so it is safe to
817 /// to deallocate once Finish returns.
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;
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.
827 /// WriteAndFinish is equivalent of performing WriteLast and Finish
828 /// in a single step.
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.
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;
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.
844 class ServerAsyncWriter final : public ServerAsyncWriterInterface<W> {
846 explicit ServerAsyncWriter(ServerContext* ctx)
847 : call_(nullptr, nullptr, nullptr), ctx_(ctx) {}
849 /// See \a ServerAsyncStreamingInterface::SendInitialMetadata for semantics.
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.
855 /// \param[in] tag Tag identifying this request.
856 void SendInitialMetadata(void* tag) override {
857 GPR_CODEGEN_ASSERT(!ctx_->sent_initial_metadata_);
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());
865 ctx_->sent_initial_metadata_ = true;
866 call_.PerformOps(&meta_ops_);
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_);
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();
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_);
889 /// See the \a ServerAsyncWriterInterface.WriteAndFinish method for semantics.
891 /// Implicit input parameter:
892 /// - the \a ServerContext associated with this call is used
893 /// for sending trailing (and initial) metadata to the client.
895 /// Note: \a status must have an OK code.
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_);
909 /// See the \a ServerAsyncWriterInterface.Finish method for semantics.
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.
915 /// Note: there are no restrictions are the code of
916 /// \a status,it may be non-OK
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_);
928 void BindCall(::grpc::internal::Call* call) override { call_ = *call; }
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());
938 ctx_->sent_initial_metadata_ = true;
942 ::grpc::internal::Call call_;
944 ::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata>
946 ::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata,
947 ::grpc::internal::CallOpSendMessage,
948 ::grpc::internal::CallOpServerSendStatus>
950 ::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata,
951 ::grpc::internal::CallOpServerSendStatus>
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> {
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.
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'
973 /// * it is desired to end the call early with some non-OK status code.
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.
979 /// gRPC doesn't take ownership or a reference to \a status, so it is safe to
980 /// to deallocate once Finish returns.
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;
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.
990 /// WriteAndFinish is equivalent of performing WriteLast and Finish in a
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.
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;
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> {
1012 explicit ServerAsyncReaderWriter(ServerContext* ctx)
1013 : call_(nullptr, nullptr, nullptr), ctx_(ctx) {}
1015 /// See \a ServerAsyncStreamingInterface::SendInitialMetadata for semantics.
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.
1021 /// \param[in] tag Tag identifying this request.
1022 void SendInitialMetadata(void* tag) override {
1023 GPR_CODEGEN_ASSERT(!ctx_->sent_initial_metadata_);
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());
1031 ctx_->sent_initial_metadata_ = true;
1032 call_.PerformOps(&meta_ops_);
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_);
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_);
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();
1054 EnsureInitialMetadataSent(&write_ops_);
1055 GPR_CODEGEN_ASSERT(write_ops_.SendMessage(msg, options).ok());
1056 call_.PerformOps(&write_ops_);
1059 /// See the \a ServerAsyncReaderWriterInterface.WriteAndFinish
1060 /// method for semantics.
1062 /// Implicit input parameter:
1063 /// - the \a ServerContext associated with this call is used
1064 /// for sending trailing (and initial) metadata to the client.
1066 /// Note: \a status must have an OK code.
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_);
1080 /// See the \a ServerAsyncReaderWriterInterface.Finish method for semantics.
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.
1086 /// Note: there are no restrictions are the code of \a status,
1087 /// it may be non-OK
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_);
1095 finish_ops_.ServerSendStatus(&ctx_->trailing_metadata_, status);
1096 call_.PerformOps(&finish_ops_);
1100 friend class ::grpc_impl::Server;
1102 void BindCall(::grpc::internal::Call* call) override { call_ = *call; }
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());
1112 ctx_->sent_initial_metadata_ = true;
1116 ::grpc::internal::Call call_;
1117 ServerContext* ctx_;
1118 ::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata>
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>
1125 ::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata,
1126 ::grpc::internal::CallOpServerSendStatus>
1132 #endif // GRPCPP_IMPL_CODEGEN_ASYNC_STREAM_H