diff options
Diffstat (limited to 'include/grpc++/impl/codegen/async_stream.h')
| -rw-r--r-- | include/grpc++/impl/codegen/async_stream.h | 629 |
1 files changed, 450 insertions, 179 deletions
diff --git a/include/grpc++/impl/codegen/async_stream.h b/include/grpc++/impl/codegen/async_stream.h index f97d824baf..ddbf3e655e 100644 --- a/include/grpc++/impl/codegen/async_stream.h +++ b/include/grpc++/impl/codegen/async_stream.h @@ -1,33 +1,18 @@ /* * - * Copyright 2015, Google Inc. - * All rights reserved. + * Copyright 2015 gRPC authors. * - * Redistribution and use in source and binary forms, with or without - * modification, are permitted provided that the following conditions are - * met: + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at * - * * Redistributions of source code must retain the above copyright - * notice, this list of conditions and the following disclaimer. - * * Redistributions in binary form must reproduce the above - * copyright notice, this list of conditions and the following disclaimer - * in the documentation and/or other materials provided with the - * distribution. - * * Neither the name of Google Inc. nor the names of its - * contributors may be used to endorse or promote products derived from - * this software without specific prior written permission. + * http://www.apache.org/licenses/LICENSE-2.0 * - * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS - * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT - * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR - * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT - * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, - * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT - * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, - * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY - * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT - * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE - * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. * */ @@ -45,6 +30,7 @@ namespace grpc { class CompletionQueue; +namespace internal { /// Common interface for all client side asynchronous streaming. class ClientAsyncStreamingInterface { public: @@ -53,16 +39,38 @@ class ClientAsyncStreamingInterface { /// Request notification of the reading of the initial metadata. Completion /// will be notified by \a tag on the associated completion queue. /// This call is optional, but if it is used, it cannot be used concurrently - /// with or after the \a Read method. + /// with or after the \a AsyncReaderInterface::Read method. /// /// \param[in] tag Tag identifying this request. virtual void ReadInitialMetadata(void* tag) = 0; - /// Indicate that the stream is to be finished and request notification - /// Should not be used concurrently with other operations + /// Indicate that the stream is to be finished and request notification for + /// when the call has been ended. + /// Should not be used concurrently with other operations. + /// + /// It is appropriate to call this method when both: + /// * the client side has no more message to send + /// (this can be declared implicitly by calling this method, or + /// explicitly through an earlier call to the <i>WritesDone</i> method + /// of the class in use, e.g. \a ClientAsyncWriterInterface::WritesDone or + /// \a ClientAsyncReaderWriterInterface::WritesDone). + /// * there are no more messages to be received from the server (this can + /// be known implicitly by the calling code, or explicitly from an + /// earlier call to \a AsyncReaderInterface::Read that yielded a failed + /// result, e.g. cq->Next(&read_tag, &ok) filled in 'ok' with 'false'). + /// + /// This function will return when either: + /// - all incoming messages have been read and the server has returned + /// a status. + /// - the server has returned a non-OK status. + /// - the call failed for some reason and the library generated a + /// status. + /// + /// Note that implementations of this method attempt to receive initial + /// metadata from the server if initial metadata hasn't yet been received. /// - /// \param[out] status To be updated with the operation status. /// \param[in] tag Tag identifying this request. + /// \param[out] status To be updated with the operation status. virtual void Finish(Status* status, void* tag) = 0; }; @@ -77,11 +85,14 @@ class AsyncReaderInterface { /// This is thread-safe with respect to \a Write or \a WritesDone methods. It /// should not be called concurrently with other streaming APIs /// on the same stream. It is not meaningful to call it concurrently - /// with another \a Read on the same stream since reads on the same stream - /// are delivered in order. + /// with another \a AsyncReaderInterface::Read on the same stream since reads + /// on the same stream are delivered in order. /// /// \param[out] msg Where to eventually store the read message. /// \param[in] tag The tag identifying the operation. + /// + /// Side effect: note that this method attempt to receive initial metadata for + /// a stream if it hasn't yet been received. virtual void Read(R* msg, void* tag) = 0; }; @@ -96,7 +107,7 @@ class AsyncWriterInterface { /// Only one write may be outstanding at any given time. This means that /// after calling Write, one must wait to receive \a tag from the completion /// queue BEFORE calling Write again. - /// This is thread-safe with respect to \a Read + /// This is thread-safe with respect to \a AsyncReaderInterface::Read /// /// \param[in] msg The message to be written. /// \param[in] tag The tag identifying the operation. @@ -109,7 +120,7 @@ class AsyncWriterInterface { /// after calling Write, one must wait to receive \a tag from the completion /// queue BEFORE calling Write again. /// WriteOptions \a options is used to set the write options of this message. - /// This is thread-safe with respect to \a Read + /// This is thread-safe with respect to \a AsyncReaderInterface::Read /// /// \param[in] msg The message to be written. /// \param[in] options The WriteOptions to be used to write this message. @@ -117,11 +128,11 @@ class AsyncWriterInterface { virtual void Write(const W& msg, WriteOptions options, void* tag) = 0; /// Request the writing of \a msg and coalesce it with the writing - /// of trailing metadata, using WriteOptions \a options with identifying tag - /// \a tag. + /// of trailing metadata, using WriteOptions \a options with + /// identifying tag \a tag. /// - /// For client, WriteLast is equivalent of performing Write and WritesDone in - /// a single step. + /// For client, WriteLast is equivalent of performing Write and + /// WritesDone in a single step. /// For server, WriteLast buffers the \a msg. The writing of \a msg is held /// until Finish is called, where \a msg and trailing metadata are coalesced /// and write is initiated. Note that WriteLast can only buffer \a msg up to @@ -136,30 +147,80 @@ class AsyncWriterInterface { } }; +} // namespace internal + template <class R> -class ClientAsyncReaderInterface : public ClientAsyncStreamingInterface, - public AsyncReaderInterface<R> {}; +class ClientAsyncReaderInterface + : public internal::ClientAsyncStreamingInterface, + public internal::AsyncReaderInterface<R> {}; + +/// Common interface for client side asynchronous writing. +template <class W> +class ClientAsyncWriterInterface + : public internal::ClientAsyncStreamingInterface, + public internal::AsyncWriterInterface<W> { + public: + /// Signal the client is done with the writes (half-close the client stream). + /// Thread-safe with respect to \a AsyncReaderInterface::Read + /// + /// \param[in] tag The tag identifying the operation. + virtual void WritesDone(void* tag) = 0; +}; + +/// Async client-side interface for bi-directional streaming, +/// where the client-to-server message stream has messages of type \a W, +/// and the server-to-client message stream has messages of type \a R. +template <class W, class R> +class ClientAsyncReaderWriterInterface + : public internal::ClientAsyncStreamingInterface, + public internal::AsyncWriterInterface<W>, + public internal::AsyncReaderInterface<R> { + public: + /// Signal the client is done with the writes (half-close the client stream). + /// Thread-safe with respect to \a AsyncReaderInterface::Read + /// + /// \param[in] tag The tag identifying the operation. + virtual void WritesDone(void* tag) = 0; +}; +/// Async client-side API for doing server-streaming RPCs, +/// where the incoming message stream coming from the server has +/// messages of type \a R. template <class R> class ClientAsyncReader final : public ClientAsyncReaderInterface<R> { public: - /// Create a stream and write the first request out. - template <class W> - static ClientAsyncReader* Create(ChannelInterface* channel, - CompletionQueue* cq, const RpcMethod& method, - ClientContext* context, const W& request, - void* tag) { - Call call = channel->CreateCall(method, context, cq); - return new (g_core_codegen_interface->grpc_call_arena_alloc( - call.call(), sizeof(ClientAsyncReader))) - ClientAsyncReader(call, context, request, tag); - } + struct internal { + /// Create a stream and write the first request out. + /// \a tag will be notified on \a cq when the call has been started and + /// \a request has been written out. + /// Note that \a context will be used to fill in custom initial metadata + /// used to send to the server when starting the call. + template <class W> + static ClientAsyncReader* Create(::grpc::ChannelInterface* channel, + CompletionQueue* cq, + const ::grpc::internal::RpcMethod& method, + ClientContext* context, const W& request, + void* tag) { + ::grpc::internal::Call call = channel->CreateCall(method, context, cq); + return new (g_core_codegen_interface->grpc_call_arena_alloc( + call.call(), sizeof(ClientAsyncReader))) + ClientAsyncReader(call, context, request, tag); + } + }; // always allocated against a call arena, no memory free required static void operator delete(void* ptr, std::size_t size) { assert(size == sizeof(ClientAsyncReader)); } + /// See the \a ClientAsyncStreamingInterface.ReadInitialMetadata + /// method for semantics. + /// + /// Side effect: + /// - upon receiving initial metadata from the server, + /// the \a ClientContext associated with this call is updated, and the + /// calling code can access the received metadata through the + /// \a ClientContext. void ReadInitialMetadata(void* tag) override { GPR_CODEGEN_ASSERT(!context_->initial_metadata_received_); @@ -177,6 +238,11 @@ class ClientAsyncReader final : public ClientAsyncReaderInterface<R> { call_.PerformOps(&read_ops_); } + /// See the \a ClientAsyncStreamingInterface.Finish method for semantics. + /// + /// Side effect: + /// - the \a ClientContext associated with this call is updated with + /// possible initial and trailing metadata received from the server. void Finish(Status* status, void* tag) override { finish_ops_.set_output_tag(tag); if (!context_->initial_metadata_received_) { @@ -188,8 +254,8 @@ class ClientAsyncReader final : public ClientAsyncReaderInterface<R> { private: template <class W> - ClientAsyncReader(Call call, ClientContext* context, const W& request, - void* tag) + ClientAsyncReader(::grpc::internal::Call call, ClientContext* context, + const W& request, void* tag) : context_(context), call_(call) { init_ops_.set_output_tag(tag); init_ops_.SendInitialMetadata(context->send_initial_metadata_, @@ -201,45 +267,61 @@ class ClientAsyncReader final : public ClientAsyncReaderInterface<R> { } ClientContext* context_; - Call call_; - CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage, CallOpClientSendClose> + ::grpc::internal::Call call_; + ::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata, + ::grpc::internal::CallOpSendMessage, + ::grpc::internal::CallOpClientSendClose> init_ops_; - CallOpSet<CallOpRecvInitialMetadata> meta_ops_; - CallOpSet<CallOpRecvInitialMetadata, CallOpRecvMessage<R>> read_ops_; - CallOpSet<CallOpRecvInitialMetadata, CallOpClientRecvStatus> finish_ops_; -}; - -/// Common interface for client side asynchronous writing. -template <class W> -class ClientAsyncWriterInterface : public ClientAsyncStreamingInterface, - public AsyncWriterInterface<W> { - public: - /// Signal the client is done with the writes. - /// Thread-safe with respect to \a Read - /// - /// \param[in] tag The tag identifying the operation. - virtual void WritesDone(void* tag) = 0; + ::grpc::internal::CallOpSet<::grpc::internal::CallOpRecvInitialMetadata> + meta_ops_; + ::grpc::internal::CallOpSet<::grpc::internal::CallOpRecvInitialMetadata, + ::grpc::internal::CallOpRecvMessage<R>> + read_ops_; + ::grpc::internal::CallOpSet<::grpc::internal::CallOpRecvInitialMetadata, + ::grpc::internal::CallOpClientRecvStatus> + finish_ops_; }; +/// Async API on the client side for doing client-streaming RPCs, +/// where the outgoing message stream going to the server contains +/// messages of type \a W. template <class W> class ClientAsyncWriter final : public ClientAsyncWriterInterface<W> { public: - template <class R> - static ClientAsyncWriter* Create(ChannelInterface* channel, - CompletionQueue* cq, const RpcMethod& method, - ClientContext* context, R* response, - void* tag) { - Call call = channel->CreateCall(method, context, cq); - return new (g_core_codegen_interface->grpc_call_arena_alloc( - call.call(), sizeof(ClientAsyncWriter))) - ClientAsyncWriter(call, context, response, tag); - } + struct internal { + /// Create a stream and write the first request out. + /// \a tag will be notified on \a cq when the call has been started (i.e. + /// intitial metadata sent) and \a request has been written out. + /// Note that \a context will be used to fill in custom initial metadata + /// used to send to the server when starting the call. + /// \a response will be filled in with the single expected response + /// message from the server upon a successful call to the \a Finish + /// method of this instance. + template <class R> + static ClientAsyncWriter* Create(::grpc::ChannelInterface* channel, + CompletionQueue* cq, + const ::grpc::internal::RpcMethod& method, + ClientContext* context, R* response, + void* tag) { + ::grpc::internal::Call call = channel->CreateCall(method, context, cq); + return new (g_core_codegen_interface->grpc_call_arena_alloc( + call.call(), sizeof(ClientAsyncWriter))) + ClientAsyncWriter(call, context, response, tag); + } + }; // always allocated against a call arena, no memory free required static void operator delete(void* ptr, std::size_t size) { assert(size == sizeof(ClientAsyncWriter)); } + /// See the \a ClientAsyncStreamingInterface.ReadInitialMetadata method for + /// semantics. + /// + /// Side effect: + /// - upon receiving initial metadata from the server, the \a ClientContext + /// associated with this call is updated, and the calling code can access + /// the received metadata through the \a ClientContext. void ReadInitialMetadata(void* tag) override { GPR_CODEGEN_ASSERT(!context_->initial_metadata_received_); @@ -272,6 +354,13 @@ class ClientAsyncWriter final : public ClientAsyncWriterInterface<W> { call_.PerformOps(&write_ops_); } + /// See the \a ClientAsyncStreamingInterface.Finish method for semantics. + /// + /// Side effect: + /// - the \a ClientContext associated with this call is updated with + /// possible initial and trailing metadata received from the server. + /// - attempts to fill in the \a response parameter passed to this class's + /// constructor with the server's response message. void Finish(Status* status, void* tag) override { finish_ops_.set_output_tag(tag); if (!context_->initial_metadata_received_) { @@ -283,7 +372,8 @@ class ClientAsyncWriter final : public ClientAsyncWriterInterface<W> { private: template <class R> - ClientAsyncWriter(Call call, ClientContext* context, R* response, void* tag) + ClientAsyncWriter(::grpc::internal::Call call, ClientContext* context, + R* response, void* tag) : context_(context), call_(call) { finish_ops_.RecvMessage(response); finish_ops_.AllowNoMessage(); @@ -301,48 +391,57 @@ class ClientAsyncWriter final : public ClientAsyncWriterInterface<W> { } ClientContext* context_; - Call call_; - CallOpSet<CallOpRecvInitialMetadata> meta_ops_; - CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage, CallOpClientSendClose> + ::grpc::internal::Call call_; + ::grpc::internal::CallOpSet<::grpc::internal::CallOpRecvInitialMetadata> + meta_ops_; + ::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata, + ::grpc::internal::CallOpSendMessage, + ::grpc::internal::CallOpClientSendClose> write_ops_; - CallOpSet<CallOpRecvInitialMetadata, CallOpGenericRecvMessage, - CallOpClientRecvStatus> + ::grpc::internal::CallOpSet<::grpc::internal::CallOpRecvInitialMetadata, + ::grpc::internal::CallOpGenericRecvMessage, + ::grpc::internal::CallOpClientRecvStatus> finish_ops_; }; -/// Client-side interface for asynchronous bi-directional streaming. -template <class W, class R> -class ClientAsyncReaderWriterInterface : public ClientAsyncStreamingInterface, - public AsyncWriterInterface<W>, - public AsyncReaderInterface<R> { - public: - /// Signal the client is done with the writes. - /// Thread-safe with respect to \a Read - /// - /// \param[in] tag The tag identifying the operation. - virtual void WritesDone(void* tag) = 0; -}; - +/// Async client-side interface for bi-directional streaming, +/// where the outgoing message stream going to the server +/// has messages of type \a W, and the incoming message stream coming +/// from the server has messages of type \a R. template <class W, class R> class ClientAsyncReaderWriter final : public ClientAsyncReaderWriterInterface<W, R> { public: - static ClientAsyncReaderWriter* Create(ChannelInterface* channel, - CompletionQueue* cq, - const RpcMethod& method, - ClientContext* context, void* tag) { - Call call = channel->CreateCall(method, context, cq); - - return new (g_core_codegen_interface->grpc_call_arena_alloc( - call.call(), sizeof(ClientAsyncReaderWriter))) - ClientAsyncReaderWriter(call, context, tag); - } + struct internal { + /// Create a stream and write the first request out. + /// \a tag will be notified on \a cq when the call has been started (i.e. + /// intitial metadata sent). + /// Note that \a context will be used to fill in custom initial metadata + /// used to send to the server when starting the call. + static ClientAsyncReaderWriter* Create( + ::grpc::ChannelInterface* channel, CompletionQueue* cq, + const ::grpc::internal::RpcMethod& method, ClientContext* context, + void* tag) { + ::grpc::internal::Call call = channel->CreateCall(method, context, cq); + + return new (g_core_codegen_interface->grpc_call_arena_alloc( + call.call(), sizeof(ClientAsyncReaderWriter))) + ClientAsyncReaderWriter(call, context, tag); + } + }; // always allocated against a call arena, no memory free required static void operator delete(void* ptr, std::size_t size) { assert(size == sizeof(ClientAsyncReaderWriter)); } + /// See the \a ClientAsyncStreamingInterface.ReadInitialMetadata method + /// for semantics of this method. + /// + /// Side effect: + /// - upon receiving initial metadata from the server, the \a ClientContext + /// is updated with it, and then the receiving initial metadata can + /// be accessed through this \a ClientContext. void ReadInitialMetadata(void* tag) override { GPR_CODEGEN_ASSERT(!context_->initial_metadata_received_); @@ -384,6 +483,10 @@ class ClientAsyncReaderWriter final call_.PerformOps(&write_ops_); } + /// See the \a ClientAsyncStreamingInterface.Finish method for semantics. + /// Side effect + /// - the \a ClientContext associated with this call is updated with + /// possible initial and trailing metadata sent from the server. void Finish(Status* status, void* tag) override { finish_ops_.set_output_tag(tag); if (!context_->initial_metadata_received_) { @@ -394,7 +497,8 @@ class ClientAsyncReaderWriter final } private: - ClientAsyncReaderWriter(Call call, ClientContext* context, void* tag) + ClientAsyncReaderWriter(::grpc::internal::Call call, ClientContext* context, + void* tag) : context_(context), call_(call) { if (context_->initial_metadata_corked_) { // if corked bit is set in context, we buffer up the initial metadata to @@ -410,29 +514,165 @@ class ClientAsyncReaderWriter final } ClientContext* context_; - Call call_; - CallOpSet<CallOpRecvInitialMetadata> meta_ops_; - CallOpSet<CallOpRecvInitialMetadata, CallOpRecvMessage<R>> read_ops_; - CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage, CallOpClientSendClose> + ::grpc::internal::Call call_; + ::grpc::internal::CallOpSet<::grpc::internal::CallOpRecvInitialMetadata> + meta_ops_; + ::grpc::internal::CallOpSet<::grpc::internal::CallOpRecvInitialMetadata, + ::grpc::internal::CallOpRecvMessage<R>> + read_ops_; + ::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata, + ::grpc::internal::CallOpSendMessage, + ::grpc::internal::CallOpClientSendClose> write_ops_; - CallOpSet<CallOpRecvInitialMetadata, CallOpClientRecvStatus> finish_ops_; + ::grpc::internal::CallOpSet<::grpc::internal::CallOpRecvInitialMetadata, + ::grpc::internal::CallOpClientRecvStatus> + finish_ops_; }; template <class W, class R> -class ServerAsyncReaderInterface : public ServerAsyncStreamingInterface, - public AsyncReaderInterface<R> { +class ServerAsyncReaderInterface + : public internal::ServerAsyncStreamingInterface, + public internal::AsyncReaderInterface<R> { public: + /// Indicate that the stream is to be finished with a certain status code + /// and also send out \a msg response to the client. + /// Request notification for when the server has sent the response and the + /// appropriate signals to the client to end the call. + /// Should not be used concurrently with other operations. + /// + /// It is appropriate to call this method when: + /// * all messages from the client have been received (either known + /// implictly, or explicitly because a previous + /// \a AsyncReaderInterface::Read operation with a non-ok result, + /// e.g., cq->Next(&read_tag, &ok) filled in 'ok' with 'false'). + /// + /// This operation will end when the server has finished sending out initial + /// metadata (if not sent already), response message, and status, or if + /// some failure occurred when trying to do so. + /// + /// \param[in] tag Tag identifying this request. + /// \param[in] status To be sent to the client as the result of this call. + /// \param[in] msg To be sent to the client as the response for this call. virtual void Finish(const W& msg, const Status& status, void* tag) = 0; + /// Indicate that the stream is to be finished with a certain + /// non-OK status code. + /// Request notification for when the server has sent the appropriate + /// signals to the client to end the call. + /// Should not be used concurrently with other operations. + /// + /// This call is meant to end the call with some error, and can be called at + /// any point that the server would like to "fail" the call (though note + /// this shouldn't be called concurrently with any other "sending" call, like + /// \a AsyncWriterInterface::Write). + /// + /// This operation will end when the server has finished sending out initial + /// metadata (if not sent already), and status, or if some failure occurred + /// when trying to do so. + /// + /// \param[in] tag Tag identifying this request. + /// \param[in] status To be sent to the client as the result of this call. + /// - Note: \a status must have a non-OK code. virtual void FinishWithError(const Status& status, void* tag) = 0; }; +template <class W> +class ServerAsyncWriterInterface + : public internal::ServerAsyncStreamingInterface, + public internal::AsyncWriterInterface<W> { + public: + /// Indicate that the stream is to be finished with a certain status code. + /// Request notification for when the server has sent the appropriate + /// signals to the client to end the call. + /// Should not be used concurrently with other operations. + /// + /// It is appropriate to call this method when either: + /// * all messages from the client have been received (either known + /// implictly, or explicitly because a previous \a + /// AsyncReaderInterface::Read operation with a non-ok + /// result (e.g., cq->Next(&read_tag, &ok) filled in 'ok' with 'false'. + /// * it is desired to end the call early with some non-OK status code. + /// + /// This operation will end when the server has finished sending out initial + /// metadata (if not sent already), response message, and status, or if + /// some failure occurred when trying to do so. + /// + /// \param[in] tag Tag identifying this request. + /// \param[in] status To be sent to the client as the result of this call. + virtual void Finish(const Status& status, void* tag) = 0; + + /// Request the writing of \a msg and coalesce it with trailing metadata which + /// contains \a status, using WriteOptions options with + /// identifying tag \a tag. + /// + /// WriteAndFinish is equivalent of performing WriteLast and Finish + /// in a single step. + /// + /// \param[in] msg The message to be written. + /// \param[in] options The WriteOptions to be used to write this message. + /// \param[in] status The Status that server returns to client. + /// \param[in] tag The tag identifying the operation. + virtual void WriteAndFinish(const W& msg, WriteOptions options, + const Status& status, void* tag) = 0; +}; + +/// Server-side interface for asynchronous bi-directional streaming. +template <class W, class R> +class ServerAsyncReaderWriterInterface + : public internal::ServerAsyncStreamingInterface, + public internal::AsyncWriterInterface<W>, + public internal::AsyncReaderInterface<R> { + public: + /// Indicate that the stream is to be finished with a certain status code. + /// Request notification for when the server has sent the appropriate + /// signals to the client to end the call. + /// Should not be used concurrently with other operations. + /// + /// It is appropriate to call this method when either: + /// * all messages from the client have been received (either known + /// implictly, or explicitly because a previous \a + /// AsyncReaderInterface::Read operation + /// with a non-ok result (e.g., cq->Next(&read_tag, &ok) filled in 'ok' + /// with 'false'. + /// * it is desired to end the call early with some non-OK status code. + /// + /// This operation will end when the server has finished sending out initial + /// metadata (if not sent already), response message, and status, or if some + /// failure occurred when trying to do so. + /// + /// \param[in] tag Tag identifying this request. + /// \param[in] status To be sent to the client as the result of this call. + virtual void Finish(const Status& status, void* tag) = 0; + + /// Request the writing of \a msg and coalesce it with trailing metadata which + /// contains \a status, using WriteOptions options with + /// identifying tag \a tag. + /// + /// WriteAndFinish is equivalent of performing WriteLast and Finish in a + /// single step. + /// + /// \param[in] msg The message to be written. + /// \param[in] options The WriteOptions to be used to write this message. + /// \param[in] status The Status that server returns to client. + /// \param[in] tag The tag identifying the operation. + virtual void WriteAndFinish(const W& msg, WriteOptions options, + const Status& status, void* tag) = 0; +}; + +/// Async server-side API for doing client-streaming RPCs, +/// where the incoming message stream from the client has messages of type \a R, +/// and the single response message sent from the server is type \a W. template <class W, class R> class ServerAsyncReader final : public ServerAsyncReaderInterface<W, R> { public: explicit ServerAsyncReader(ServerContext* ctx) : call_(nullptr, nullptr, nullptr), ctx_(ctx) {} + /// See \a ServerAsyncStreamingInterface::SendInitialMetadata for semantics. + /// + /// Implicit input parameter: + /// - The initial metadata that will be sent to the client from this op will + /// be taken from the \a ServerContext associated with the call. void SendInitialMetadata(void* tag) override { GPR_CODEGEN_ASSERT(!ctx_->sent_initial_metadata_); @@ -452,6 +692,14 @@ class ServerAsyncReader final : public ServerAsyncReaderInterface<W, R> { call_.PerformOps(&read_ops_); } + /// See the \a ServerAsyncReaderInterface.Read method for semantics + /// + /// Side effect: + /// - also sends initial metadata if not alreay sent. + /// - uses the \a ServerContext associated with this call to send possible + /// initial and trailing metadata. + /// + /// Note: \a msg is not sent if \a status has a non-OK code. void Finish(const W& msg, const Status& status, void* tag) override { finish_ops_.set_output_tag(tag); if (!ctx_->sent_initial_metadata_) { @@ -472,6 +720,12 @@ class ServerAsyncReader final : public ServerAsyncReaderInterface<W, R> { call_.PerformOps(&finish_ops_); } + /// See the \a ServerAsyncReaderInterface.Read method for semantics + /// + /// Side effect: + /// - also sends initial metadata if not alreay sent. + /// - uses the \a ServerContext associated with this call to send possible + /// initial and trailing metadata. void FinishWithError(const Status& status, void* tag) override { GPR_CODEGEN_ASSERT(!status.ok()); finish_ops_.set_output_tag(tag); @@ -488,44 +742,34 @@ class ServerAsyncReader final : public ServerAsyncReaderInterface<W, R> { } private: - void BindCall(Call* call) override { call_ = *call; } + void BindCall(::grpc::internal::Call* call) override { call_ = *call; } - Call call_; + ::grpc::internal::Call call_; ServerContext* ctx_; - CallOpSet<CallOpSendInitialMetadata> meta_ops_; - CallOpSet<CallOpRecvMessage<R>> read_ops_; - CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage, - CallOpServerSendStatus> + ::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata> + meta_ops_; + ::grpc::internal::CallOpSet<::grpc::internal::CallOpRecvMessage<R>> read_ops_; + ::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata, + ::grpc::internal::CallOpSendMessage, + ::grpc::internal::CallOpServerSendStatus> finish_ops_; }; -template <class W> -class ServerAsyncWriterInterface : public ServerAsyncStreamingInterface, - public AsyncWriterInterface<W> { - public: - virtual void Finish(const Status& status, void* tag) = 0; - - /// Request the writing of \a msg and coalesce it with trailing metadata which - /// contains \a status, using WriteOptions options with identifying tag \a - /// tag. - /// - /// WriteAndFinish is equivalent of performing WriteLast and Finish in a - /// single step. - /// - /// \param[in] msg The message to be written. - /// \param[in] options The WriteOptions to be used to write this message. - /// \param[in] status The Status that server returns to client. - /// \param[in] tag The tag identifying the operation. - virtual void WriteAndFinish(const W& msg, WriteOptions options, - const Status& status, void* tag) = 0; -}; - +/// Async server-side API for doing server streaming RPCs, +/// where the outgoing message stream from the server has messages of type \a W. template <class W> class ServerAsyncWriter final : public ServerAsyncWriterInterface<W> { public: explicit ServerAsyncWriter(ServerContext* ctx) : call_(nullptr, nullptr, nullptr), ctx_(ctx) {} + /// See \a ServerAsyncStreamingInterface::SendInitialMetadata for semantics. + /// + /// Implicit input parameter: + /// - The initial metadata that will be sent to the client from this op will + /// be taken from the \a ServerContext associated with the call. + /// + /// \param[in] tag Tag identifying this request. void SendInitialMetadata(void* tag) override { GPR_CODEGEN_ASSERT(!ctx_->sent_initial_metadata_); @@ -559,6 +803,13 @@ class ServerAsyncWriter final : public ServerAsyncWriterInterface<W> { call_.PerformOps(&write_ops_); } + /// See the \a ServerAsyncWriterInterface.WriteAndFinish method for semantics. + /// + /// Implicit input parameter: + /// - the \a ServerContext associated with this call is used + /// for sending trailing (and initial) metadata to the client. + /// + /// Note: \a status must have an OK code. void WriteAndFinish(const W& msg, WriteOptions options, const Status& status, void* tag) override { write_ops_.set_output_tag(tag); @@ -569,6 +820,14 @@ class ServerAsyncWriter final : public ServerAsyncWriterInterface<W> { call_.PerformOps(&write_ops_); } + /// See the \a ServerAsyncWriterInterface.Finish method for semantics. + /// + /// Implicit input parameter: + /// - the \a ServerContext associated with this call is used for sending + /// trailing (and initial if not already sent) metadata to the client. + /// + /// Note: there are no restrictions are the code of + /// \a status,it may be non-OK void Finish(const Status& status, void* tag) override { finish_ops_.set_output_tag(tag); EnsureInitialMetadataSent(&finish_ops_); @@ -577,7 +836,7 @@ class ServerAsyncWriter final : public ServerAsyncWriterInterface<W> { } private: - void BindCall(Call* call) override { call_ = *call; } + void BindCall(::grpc::internal::Call* call) override { call_ = *call; } template <class T> void EnsureInitialMetadataSent(T* ops) { @@ -591,38 +850,23 @@ class ServerAsyncWriter final : public ServerAsyncWriterInterface<W> { } } - Call call_; + ::grpc::internal::Call call_; ServerContext* ctx_; - CallOpSet<CallOpSendInitialMetadata> meta_ops_; - CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage, - CallOpServerSendStatus> + ::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata> + meta_ops_; + ::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata, + ::grpc::internal::CallOpSendMessage, + ::grpc::internal::CallOpServerSendStatus> write_ops_; - CallOpSet<CallOpSendInitialMetadata, CallOpServerSendStatus> finish_ops_; -}; - -/// Server-side interface for asynchronous bi-directional streaming. -template <class W, class R> -class ServerAsyncReaderWriterInterface : public ServerAsyncStreamingInterface, - public AsyncWriterInterface<W>, - public AsyncReaderInterface<R> { - public: - virtual void Finish(const Status& status, void* tag) = 0; - - /// Request the writing of \a msg and coalesce it with trailing metadata which - /// contains \a status, using WriteOptions options with identifying tag \a - /// tag. - /// - /// WriteAndFinish is equivalent of performing WriteLast and Finish in a - /// single step. - /// - /// \param[in] msg The message to be written. - /// \param[in] options The WriteOptions to be used to write this message. - /// \param[in] status The Status that server returns to client. - /// \param[in] tag The tag identifying the operation. - virtual void WriteAndFinish(const W& msg, WriteOptions options, - const Status& status, void* tag) = 0; + ::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata, + ::grpc::internal::CallOpServerSendStatus> + finish_ops_; }; +/// Async server-side API for doing bidirectional streaming RPCs, +/// where the incoming message stream coming from the client has messages of +/// type \a R, and the outgoing message stream coming from the server has +/// messages of type \a W. template <class W, class R> class ServerAsyncReaderWriter final : public ServerAsyncReaderWriterInterface<W, R> { @@ -630,6 +874,13 @@ class ServerAsyncReaderWriter final explicit ServerAsyncReaderWriter(ServerContext* ctx) : call_(nullptr, nullptr, nullptr), ctx_(ctx) {} + /// See \a ServerAsyncStreamingInterface::SendInitialMetadata for semantics. + /// + /// Implicit input parameter: + /// - The initial metadata that will be sent to the client from this op will + /// be taken from the \a ServerContext associated with the call. + /// + /// \param[in] tag Tag identifying this request. void SendInitialMetadata(void* tag) override { GPR_CODEGEN_ASSERT(!ctx_->sent_initial_metadata_); @@ -667,6 +918,14 @@ class ServerAsyncReaderWriter final call_.PerformOps(&write_ops_); } + /// See the \a ServerAsyncReaderWriterInterface.WriteAndFinish + /// method for semantics. + /// + /// Implicit input parameter: + /// - the \a ServerContext associated with this call is used + /// for sending trailing (and initial) metadata to the client. + /// + /// Note: \a status must have an OK code. void WriteAndFinish(const W& msg, WriteOptions options, const Status& status, void* tag) override { write_ops_.set_output_tag(tag); @@ -677,6 +936,14 @@ class ServerAsyncReaderWriter final call_.PerformOps(&write_ops_); } + /// See the \a ServerAsyncReaderWriterInterface.Finish method for semantics. + /// + /// Implicit input parameter: + /// - the \a ServerContext associated with this call is used for sending + /// trailing (and initial if not already sent) metadata to the client. + /// + /// Note: there are no restrictions are the code of \a status, + /// it may be non-OK void Finish(const Status& status, void* tag) override { finish_ops_.set_output_tag(tag); EnsureInitialMetadataSent(&finish_ops_); @@ -688,7 +955,7 @@ class ServerAsyncReaderWriter final private: friend class ::grpc::Server; - void BindCall(Call* call) override { call_ = *call; } + void BindCall(::grpc::internal::Call* call) override { call_ = *call; } template <class T> void EnsureInitialMetadataSent(T* ops) { @@ -702,14 +969,18 @@ class ServerAsyncReaderWriter final } } - Call call_; + ::grpc::internal::Call call_; ServerContext* ctx_; - CallOpSet<CallOpSendInitialMetadata> meta_ops_; - CallOpSet<CallOpRecvMessage<R>> read_ops_; - CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage, - CallOpServerSendStatus> + ::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata> + meta_ops_; + ::grpc::internal::CallOpSet<::grpc::internal::CallOpRecvMessage<R>> read_ops_; + ::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata, + ::grpc::internal::CallOpSendMessage, + ::grpc::internal::CallOpServerSendStatus> write_ops_; - CallOpSet<CallOpSendInitialMetadata, CallOpServerSendStatus> finish_ops_; + ::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata, + ::grpc::internal::CallOpServerSendStatus> + finish_ops_; }; } // namespace grpc |