The C based gRPC (C++, Python, Ruby, Objective-C, PHP, C#) https://grpc.io/

1010 lines
39 KiB

/*
*
* Copyright 2015 gRPC authors.
*
* 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
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* 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.
*
*/
#ifndef GRPCXX_IMPL_CODEGEN_ASYNC_STREAM_H
#define GRPCXX_IMPL_CODEGEN_ASYNC_STREAM_H
9 years ago
#include <grpc++/impl/codegen/call.h>
#include <grpc++/impl/codegen/channel_interface.h>
#include <grpc++/impl/codegen/core_codegen_interface.h>
#include <grpc++/impl/codegen/server_context.h>
9 years ago
#include <grpc++/impl/codegen/service_type.h>
#include <grpc++/impl/codegen/status.h>
namespace grpc {
class CompletionQueue;
/// Common interface for all client side asynchronous streaming.
class ClientAsyncStreamingInterface {
public:
virtual ~ClientAsyncStreamingInterface() {}
/// Start the call that was set up by the constructor, but only if the
/// constructor was invoked through the "Prepare" API which doesn't actually
/// start the call
virtual void StartCall(void* tag) = 0;
/// 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 AsyncReaderInterface::Read method.
///
/// \param[in] tag Tag identifying this request.
virtual void ReadInitialMetadata(void* tag) = 0;
8 years ago
/// 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').
8 years ago
///
/// 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[in] tag Tag identifying this request.
8 years ago
/// \param[out] status To be updated with the operation status.
virtual void Finish(Status* status, void* tag) = 0;
};
/// An interface that yields a sequence of messages of type \a R.
template <class R>
class AsyncReaderInterface {
public:
virtual ~AsyncReaderInterface() {}
/// Read a message of type \a R into \a msg. Completion will be notified by \a
/// tag on the associated completion queue.
/// 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 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.
8 years ago
///
/// 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;
};
/// An interface that can be fed a sequence of messages of type \a W.
template <class W>
class AsyncWriterInterface {
public:
virtual ~AsyncWriterInterface() {}
/// Request the writing of \a msg with identifying tag \a tag.
///
/// 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 AsyncReaderInterface::Read
///
/// \param[in] msg The message to be written.
/// \param[in] tag The tag identifying the operation.
virtual void Write(const W& msg, void* tag) = 0;
/// Request the writing of \a msg using WriteOptions \a options with
/// identifying tag \a tag.
///
/// 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.
/// WriteOptions \a options is used to set the write options of this message.
/// 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.
/// \param[in] tag The tag identifying the operation.
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.
///
/// 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
/// the flow control window size. If \a msg size is larger than the window
/// size, it will be sent on wire without buffering.
///
/// \param[in] msg The message to be written.
/// \param[in] options The WriteOptions to be used to write this message.
/// \param[in] tag The tag identifying the operation.
void WriteLast(const W& msg, WriteOptions options, void* tag) {
Write(msg, options.set_last_message(), tag);
}
};
template <class R>
class ClientAsyncReaderInterface : public ClientAsyncStreamingInterface,
public AsyncReaderInterface<R> {};
8 years ago
/// 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 object.
/// Write the first request out if \a start is set.
/// \a tag will be notified on \a cq when the call has been started and
/// \a request has been written out. If \a start is not set, \a tag must be
/// nullptr and the actual call must be initiated by StartCall
/// 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(ChannelInterface* channel,
CompletionQueue* cq, const RpcMethod& method,
ClientContext* context, const W& request,
bool start, 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, start, tag);
}
8 years ago
// always allocated against a call arena, no memory free required
static void operator delete(void* ptr, std::size_t size) {
assert(size == sizeof(ClientAsyncReader));
}
void StartCall(void* tag) override {
assert(!started_);
started_ = true;
StartCallInternal(tag);
}
/// See the \a ClientAsyncStreamingInterface.ReadInitialMetadata
/// method for semantics.
8 years ago
///
/// 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 {
assert(started_);
GPR_CODEGEN_ASSERT(!context_->initial_metadata_received_);
meta_ops_.set_output_tag(tag);
meta_ops_.RecvInitialMetadata(context_);
call_.PerformOps(&meta_ops_);
}
void Read(R* msg, void* tag) override {
assert(started_);
read_ops_.set_output_tag(tag);
if (!context_->initial_metadata_received_) {
read_ops_.RecvInitialMetadata(context_);
}
read_ops_.RecvMessage(msg);
call_.PerformOps(&read_ops_);
}
8 years ago
/// 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 {
assert(started_);
finish_ops_.set_output_tag(tag);
if (!context_->initial_metadata_received_) {
finish_ops_.RecvInitialMetadata(context_);
}
finish_ops_.ClientRecvStatus(context_, status);
call_.PerformOps(&finish_ops_);
}
private:
template <class W>
ClientAsyncReader(Call call, ClientContext* context, const W& request,
bool start, void* tag)
: context_(context), call_(call), started_(start) {
// TODO(ctiller): don't assert
GPR_CODEGEN_ASSERT(init_ops_.SendMessage(request).ok());
init_ops_.ClientSendClose();
if (start) {
StartCallInternal(tag);
} else {
assert(tag == nullptr);
}
}
void StartCallInternal(void* tag) {
init_ops_.SendInitialMetadata(context_->send_initial_metadata_,
context_->initial_metadata_flags());
init_ops_.set_output_tag(tag);
call_.PerformOps(&init_ops_);
}
ClientContext* context_;
Call call_;
bool started_;
CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage, 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 (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 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:
/// Create a stream object.
/// Start the RPC if \a start is set
/// \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.
/// If \a start is not set, \a tag must be nullptr and the actual call
/// must be initiated by StartCall
/// 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(ChannelInterface* channel,
CompletionQueue* cq, const RpcMethod& method,
ClientContext* context, R* response,
bool start, 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, start, tag);
}
8 years ago
// always allocated against a call arena, no memory free required
static void operator delete(void* ptr, std::size_t size) {
assert(size == sizeof(ClientAsyncWriter));
}
void StartCall(void* tag) override {
assert(!started_);
started_ = true;
StartCallInternal(tag);
}
8 years ago
/// 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 {
assert(started_);
GPR_CODEGEN_ASSERT(!context_->initial_metadata_received_);
meta_ops_.set_output_tag(tag);
meta_ops_.RecvInitialMetadata(context_);
call_.PerformOps(&meta_ops_);
}
void Write(const W& msg, void* tag) override {
assert(started_);
write_ops_.set_output_tag(tag);
// TODO(ctiller): don't assert
GPR_CODEGEN_ASSERT(write_ops_.SendMessage(msg).ok());
call_.PerformOps(&write_ops_);
}
8 years ago
void Write(const W& msg, WriteOptions options, void* tag) override {
assert(started_);
write_ops_.set_output_tag(tag);
if (options.is_last_message()) {
options.set_buffer_hint();
write_ops_.ClientSendClose();
}
// TODO(ctiller): don't assert
GPR_CODEGEN_ASSERT(write_ops_.SendMessage(msg, options).ok());
call_.PerformOps(&write_ops_);
}
void WritesDone(void* tag) override {
assert(started_);
write_ops_.set_output_tag(tag);
write_ops_.ClientSendClose();
call_.PerformOps(&write_ops_);
}
8 years ago
/// 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 {
assert(started_);
finish_ops_.set_output_tag(tag);
if (!context_->initial_metadata_received_) {
finish_ops_.RecvInitialMetadata(context_);
}
finish_ops_.ClientRecvStatus(context_, status);
call_.PerformOps(&finish_ops_);
}
private:
template <class R>
ClientAsyncWriter(Call call, ClientContext* context, R* response, bool start,
void* tag)
: context_(context), call_(call), started_(start) {
finish_ops_.RecvMessage(response);
finish_ops_.AllowNoMessage();
if (start) {
StartCallInternal(tag);
} else {
assert(tag == nullptr);
}
}
void StartCallInternal(void* tag) {
write_ops_.SendInitialMetadata(context_->send_initial_metadata_,
context_->initial_metadata_flags());
// if corked bit is set in context, we just keep the initial metadata
// buffered up to coalesce with later message send. No op is performed.
if (!context_->initial_metadata_corked_) {
write_ops_.set_output_tag(tag);
call_.PerformOps(&write_ops_);
}
}
ClientContext* context_;
Call call_;
bool started_;
CallOpSet<CallOpRecvInitialMetadata> meta_ops_;
CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage, CallOpClientSendClose>
write_ops_;
CallOpSet<CallOpRecvInitialMetadata, CallOpGenericRecvMessage,
CallOpClientRecvStatus>
9 years ago
finish_ops_;
};
/// 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 ClientAsyncStreamingInterface,
public AsyncWriterInterface<W>,
public 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;
};
8 years ago
/// 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:
/// Create a stream object.
/// Start the RPC request if \a start is set.
/// \a tag will be notified on \a cq when the call has been started (i.e.
/// intitial metadata sent). If \a start is not set, \a tag must be
/// nullptr and the actual call must be initiated by StartCall
/// 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(ChannelInterface* channel,
CompletionQueue* cq,
const RpcMethod& method,
ClientContext* context, bool start,
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, start, tag);
}
8 years ago
// always allocated against a call arena, no memory free required
static void operator delete(void* ptr, std::size_t size) {
assert(size == sizeof(ClientAsyncReaderWriter));
}
void StartCall(void* tag) override {
assert(!started_);
started_ = true;
StartCallInternal(tag);
}
8 years ago
/// 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 {
assert(started_);
GPR_CODEGEN_ASSERT(!context_->initial_metadata_received_);
meta_ops_.set_output_tag(tag);
meta_ops_.RecvInitialMetadata(context_);
call_.PerformOps(&meta_ops_);
}
void Read(R* msg, void* tag) override {
assert(started_);
read_ops_.set_output_tag(tag);
if (!context_->initial_metadata_received_) {
read_ops_.RecvInitialMetadata(context_);
}
read_ops_.RecvMessage(msg);
call_.PerformOps(&read_ops_);
}
void Write(const W& msg, void* tag) override {
assert(started_);
write_ops_.set_output_tag(tag);
// TODO(ctiller): don't assert
GPR_CODEGEN_ASSERT(write_ops_.SendMessage(msg).ok());
call_.PerformOps(&write_ops_);
}
8 years ago
void Write(const W& msg, WriteOptions options, void* tag) override {
assert(started_);
write_ops_.set_output_tag(tag);
if (options.is_last_message()) {
options.set_buffer_hint();
write_ops_.ClientSendClose();
}
// TODO(ctiller): don't assert
GPR_CODEGEN_ASSERT(write_ops_.SendMessage(msg, options).ok());
call_.PerformOps(&write_ops_);
}
void WritesDone(void* tag) override {
assert(started_);
write_ops_.set_output_tag(tag);
write_ops_.ClientSendClose();
call_.PerformOps(&write_ops_);
}
8 years ago
/// 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 {
assert(started_);
finish_ops_.set_output_tag(tag);
if (!context_->initial_metadata_received_) {
finish_ops_.RecvInitialMetadata(context_);
}
finish_ops_.ClientRecvStatus(context_, status);
call_.PerformOps(&finish_ops_);
}
private:
ClientAsyncReaderWriter(Call call, ClientContext* context, bool start,
void* tag)
: context_(context), call_(call), started_(start) {
if (start) {
StartCallInternal(tag);
} else {
assert(tag == nullptr);
}
}
void StartCallInternal(void* tag) {
write_ops_.SendInitialMetadata(context_->send_initial_metadata_,
context_->initial_metadata_flags());
// if corked bit is set in context, we just keep the initial metadata
// buffered up to coalesce with later message send. No op is performed.
if (!context_->initial_metadata_corked_) {
write_ops_.set_output_tag(tag);
call_.PerformOps(&write_ops_);
}
}
ClientContext* context_;
Call call_;
bool started_;
CallOpSet<CallOpRecvInitialMetadata> meta_ops_;
CallOpSet<CallOpRecvInitialMetadata, CallOpRecvMessage<R>> read_ops_;
CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage, CallOpClientSendClose>
write_ops_;
CallOpSet<CallOpRecvInitialMetadata, CallOpClientRecvStatus> finish_ops_;
};
template <class W, class R>
class ServerAsyncReaderInterface : public ServerAsyncStreamingInterface,
public AsyncReaderInterface<R> {
public:
8 years ago
/// 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.
8 years ago
/// 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').
8 years ago
///
/// 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.
8 years ago
///
/// \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.
8 years ago
/// 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).
8 years ago
///
/// 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.
8 years ago
///
/// \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;
};
8 years ago
/// 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.
8 years ago
///
/// 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_);
meta_ops_.set_output_tag(tag);
meta_ops_.SendInitialMetadata(ctx_->initial_metadata_,
ctx_->initial_metadata_flags());
if (ctx_->compression_level_set()) {
meta_ops_.set_compression_level(ctx_->compression_level());
}
ctx_->sent_initial_metadata_ = true;
call_.PerformOps(&meta_ops_);
}
void Read(R* msg, void* tag) override {
read_ops_.set_output_tag(tag);
read_ops_.RecvMessage(msg);
call_.PerformOps(&read_ops_);
}
8 years ago
/// 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_) {
finish_ops_.SendInitialMetadata(ctx_->initial_metadata_,
ctx_->initial_metadata_flags());
if (ctx_->compression_level_set()) {
finish_ops_.set_compression_level(ctx_->compression_level());
}
ctx_->sent_initial_metadata_ = true;
}
// The response is dropped if the status is not OK.
if (status.ok()) {
finish_ops_.ServerSendStatus(ctx_->trailing_metadata_,
finish_ops_.SendMessage(msg));
} else {
finish_ops_.ServerSendStatus(ctx_->trailing_metadata_, status);
}
call_.PerformOps(&finish_ops_);
}
8 years ago
/// 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);
if (!ctx_->sent_initial_metadata_) {
finish_ops_.SendInitialMetadata(ctx_->initial_metadata_,
ctx_->initial_metadata_flags());
if (ctx_->compression_level_set()) {
finish_ops_.set_compression_level(ctx_->compression_level());
}
ctx_->sent_initial_metadata_ = true;
}
finish_ops_.ServerSendStatus(ctx_->trailing_metadata_, status);
call_.PerformOps(&finish_ops_);
}
private:
void BindCall(Call* call) override { call_ = *call; }
Call call_;
ServerContext* ctx_;
CallOpSet<CallOpSendInitialMetadata> meta_ops_;
CallOpSet<CallOpRecvMessage<R>> read_ops_;
CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage,
CallOpServerSendStatus>
finish_ops_;
};
template <class W>
class ServerAsyncWriterInterface : public ServerAsyncStreamingInterface,
public 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;
};
8 years ago
/// 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.
8 years ago
///
/// 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.
8 years ago
///
/// \param[in] tag Tag identifying this request.
void SendInitialMetadata(void* tag) override {
GPR_CODEGEN_ASSERT(!ctx_->sent_initial_metadata_);
meta_ops_.set_output_tag(tag);
meta_ops_.SendInitialMetadata(ctx_->initial_metadata_,
ctx_->initial_metadata_flags());
if (ctx_->compression_level_set()) {
meta_ops_.set_compression_level(ctx_->compression_level());
}
ctx_->sent_initial_metadata_ = true;
call_.PerformOps(&meta_ops_);
}
void Write(const W& msg, void* tag) override {
write_ops_.set_output_tag(tag);
EnsureInitialMetadataSent(&write_ops_);
// TODO(ctiller): don't assert
GPR_CODEGEN_ASSERT(write_ops_.SendMessage(msg).ok());
call_.PerformOps(&write_ops_);
}
void Write(const W& msg, WriteOptions options, void* tag) override {
write_ops_.set_output_tag(tag);
if (options.is_last_message()) {
options.set_buffer_hint();
}
EnsureInitialMetadataSent(&write_ops_);
// TODO(ctiller): don't assert
GPR_CODEGEN_ASSERT(write_ops_.SendMessage(msg, options).ok());
call_.PerformOps(&write_ops_);
}
8 years ago
/// 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);
EnsureInitialMetadataSent(&write_ops_);
options.set_buffer_hint();
GPR_CODEGEN_ASSERT(write_ops_.SendMessage(msg, options).ok());
write_ops_.ServerSendStatus(ctx_->trailing_metadata_, status);
call_.PerformOps(&write_ops_);
}
8 years ago
/// 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.
8 years ago
///
/// 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_);
finish_ops_.ServerSendStatus(ctx_->trailing_metadata_, status);
call_.PerformOps(&finish_ops_);
}
private:
void BindCall(Call* call) override { call_ = *call; }
template <class T>
void EnsureInitialMetadataSent(T* ops) {
if (!ctx_->sent_initial_metadata_) {
ops->SendInitialMetadata(ctx_->initial_metadata_,
ctx_->initial_metadata_flags());
if (ctx_->compression_level_set()) {
ops->set_compression_level(ctx_->compression_level());
}
ctx_->sent_initial_metadata_ = true;
}
}
Call call_;
ServerContext* ctx_;
CallOpSet<CallOpSendInitialMetadata> meta_ops_;
CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage,
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:
/// 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;
};
8 years ago
/// 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> {
public:
explicit ServerAsyncReaderWriter(ServerContext* ctx)
: call_(nullptr, nullptr, nullptr), ctx_(ctx) {}
/// See \a ServerAsyncStreamingInterface::SendInitialMetadata for semantics.
8 years ago
///
/// 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.
8 years ago
///
/// \param[in] tag Tag identifying this request.
void SendInitialMetadata(void* tag) override {
GPR_CODEGEN_ASSERT(!ctx_->sent_initial_metadata_);
meta_ops_.set_output_tag(tag);
meta_ops_.SendInitialMetadata(ctx_->initial_metadata_,
ctx_->initial_metadata_flags());
if (ctx_->compression_level_set()) {
meta_ops_.set_compression_level(ctx_->compression_level());
}
ctx_->sent_initial_metadata_ = true;
call_.PerformOps(&meta_ops_);
}
void Read(R* msg, void* tag) override {
read_ops_.set_output_tag(tag);
read_ops_.RecvMessage(msg);
call_.PerformOps(&read_ops_);
}
void Write(const W& msg, void* tag) override {
write_ops_.set_output_tag(tag);
EnsureInitialMetadataSent(&write_ops_);
// TODO(ctiller): don't assert
GPR_CODEGEN_ASSERT(write_ops_.SendMessage(msg).ok());
call_.PerformOps(&write_ops_);
}
void Write(const W& msg, WriteOptions options, void* tag) override {
write_ops_.set_output_tag(tag);
if (options.is_last_message()) {
options.set_buffer_hint();
}
EnsureInitialMetadataSent(&write_ops_);
GPR_CODEGEN_ASSERT(write_ops_.SendMessage(msg, options).ok());
call_.PerformOps(&write_ops_);
}
/// See the \a ServerAsyncReaderWriterInterface.WriteAndFinish
/// method for semantics.
8 years ago
///
/// 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);
EnsureInitialMetadataSent(&write_ops_);
options.set_buffer_hint();
GPR_CODEGEN_ASSERT(write_ops_.SendMessage(msg, options).ok());
write_ops_.ServerSendStatus(ctx_->trailing_metadata_, status);
call_.PerformOps(&write_ops_);
}
8 years ago
/// 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.
8 years ago
///
/// 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_);
finish_ops_.ServerSendStatus(ctx_->trailing_metadata_, status);
call_.PerformOps(&finish_ops_);
}
private:
friend class ::grpc::Server;
void BindCall(Call* call) override { call_ = *call; }
template <class T>
void EnsureInitialMetadataSent(T* ops) {
if (!ctx_->sent_initial_metadata_) {
ops->SendInitialMetadata(ctx_->initial_metadata_,
ctx_->initial_metadata_flags());
if (ctx_->compression_level_set()) {
ops->set_compression_level(ctx_->compression_level());
}
ctx_->sent_initial_metadata_ = true;
}
}
Call call_;
ServerContext* ctx_;
CallOpSet<CallOpSendInitialMetadata> meta_ops_;
CallOpSet<CallOpRecvMessage<R>> read_ops_;
CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage,
CallOpServerSendStatus>
write_ops_;
CallOpSet<CallOpSendInitialMetadata, CallOpServerSendStatus> finish_ops_;
};
} // namespace grpc
#endif // GRPCXX_IMPL_CODEGEN_ASYNC_STREAM_H