mirror of https://github.com/grpc/grpc.git
Merge pull request #24082 from karthikravis/sync-stream
Move async, callback and sync implementation from grpc_impl to grpc namespacepull/24164/head
Karthik Ravi Shankar
4 years ago
committed by
GitHub
commit
30ed2f513f
35 changed files with 4678 additions and 5242 deletions
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
@ -1,314 +0,0 @@ |
||||
/*
|
||||
* |
||||
* 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 GRPCPP_IMPL_CODEGEN_ASYNC_UNARY_CALL_IMPL_H |
||||
#define GRPCPP_IMPL_CODEGEN_ASYNC_UNARY_CALL_IMPL_H |
||||
|
||||
#include <grpcpp/impl/codegen/call.h> |
||||
#include <grpcpp/impl/codegen/channel_interface.h> |
||||
#include <grpcpp/impl/codegen/client_context.h> |
||||
#include <grpcpp/impl/codegen/server_context.h> |
||||
#include <grpcpp/impl/codegen/service_type.h> |
||||
#include <grpcpp/impl/codegen/status.h> |
||||
|
||||
namespace grpc_impl { |
||||
|
||||
/// An interface relevant for async client side unary RPCs (which send
|
||||
/// one request message to a server and receive one response message).
|
||||
template <class R> |
||||
class ClientAsyncResponseReaderInterface { |
||||
public: |
||||
virtual ~ClientAsyncResponseReaderInterface() {} |
||||
|
||||
/// 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() = 0; |
||||
|
||||
/// Request notification of the reading of 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 Finish method.
|
||||
///
|
||||
/// \param[in] tag Tag identifying this request.
|
||||
virtual void ReadInitialMetadata(void* tag) = 0; |
||||
|
||||
/// Request to receive the server's response \a msg and final \a status for
|
||||
/// the call, and to notify \a tag on this call's completion queue when
|
||||
/// finished.
|
||||
///
|
||||
/// This function will return when either:
|
||||
/// - when the server's response message and status have been received.
|
||||
/// - when the server has returned a non-OK status (no message expected in
|
||||
/// this case).
|
||||
/// - when the call failed for some reason and the library generated a
|
||||
/// non-OK status.
|
||||
///
|
||||
/// \param[in] tag Tag identifying this request.
|
||||
/// \param[out] status To be updated with the operation status.
|
||||
/// \param[out] msg To be filled in with the server's response message.
|
||||
virtual void Finish(R* msg, ::grpc::Status* status, void* tag) = 0; |
||||
}; |
||||
|
||||
namespace internal { |
||||
template <class R> |
||||
class ClientAsyncResponseReaderFactory { |
||||
public: |
||||
/// Start a call and write the request out 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, 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 ClientAsyncResponseReader<R>* Create( |
||||
::grpc::ChannelInterface* channel, ::grpc::CompletionQueue* cq, |
||||
const ::grpc::internal::RpcMethod& method, ::grpc::ClientContext* context, |
||||
const W& request, bool start) { |
||||
::grpc::internal::Call call = channel->CreateCall(method, context, cq); |
||||
return new (::grpc::g_core_codegen_interface->grpc_call_arena_alloc( |
||||
call.call(), sizeof(ClientAsyncResponseReader<R>))) |
||||
ClientAsyncResponseReader<R>(call, context, request, start); |
||||
} |
||||
}; |
||||
} // namespace internal
|
||||
|
||||
/// Async API for client-side unary RPCs, where the message response
|
||||
/// received from the server is of type \a R.
|
||||
template <class R> |
||||
class ClientAsyncResponseReader final |
||||
: public ClientAsyncResponseReaderInterface<R> { |
||||
public: |
||||
// always allocated against a call arena, no memory free required
|
||||
static void operator delete(void* /*ptr*/, std::size_t size) { |
||||
GPR_CODEGEN_ASSERT(size == sizeof(ClientAsyncResponseReader)); |
||||
} |
||||
|
||||
// This operator should never be called as the memory should be freed as part
|
||||
// of the arena destruction. It only exists to provide a matching operator
|
||||
// delete to the operator new so that some compilers will not complain (see
|
||||
// https://github.com/grpc/grpc/issues/11301) Note at the time of adding this
|
||||
// there are no tests catching the compiler warning.
|
||||
static void operator delete(void*, void*) { GPR_CODEGEN_ASSERT(false); } |
||||
|
||||
void StartCall() override { |
||||
GPR_CODEGEN_ASSERT(!started_); |
||||
started_ = true; |
||||
StartCallInternal(); |
||||
} |
||||
|
||||
/// See \a ClientAsyncResponseReaderInterface::ReadInitialMetadata for
|
||||
/// semantics.
|
||||
///
|
||||
/// Side effect:
|
||||
/// - the \a ClientContext associated with this call is updated with
|
||||
/// possible initial and trailing metadata sent from the server.
|
||||
void ReadInitialMetadata(void* tag) override { |
||||
GPR_CODEGEN_ASSERT(started_); |
||||
GPR_CODEGEN_ASSERT(!context_->initial_metadata_received_); |
||||
|
||||
single_buf.set_output_tag(tag); |
||||
single_buf.RecvInitialMetadata(context_); |
||||
call_.PerformOps(&single_buf); |
||||
initial_metadata_read_ = true; |
||||
} |
||||
|
||||
/// See \a ClientAysncResponseReaderInterface::Finish 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(R* msg, ::grpc::Status* status, void* tag) override { |
||||
GPR_CODEGEN_ASSERT(started_); |
||||
if (initial_metadata_read_) { |
||||
finish_buf.set_output_tag(tag); |
||||
finish_buf.RecvMessage(msg); |
||||
finish_buf.AllowNoMessage(); |
||||
finish_buf.ClientRecvStatus(context_, status); |
||||
call_.PerformOps(&finish_buf); |
||||
} else { |
||||
single_buf.set_output_tag(tag); |
||||
single_buf.RecvInitialMetadata(context_); |
||||
single_buf.RecvMessage(msg); |
||||
single_buf.AllowNoMessage(); |
||||
single_buf.ClientRecvStatus(context_, status); |
||||
call_.PerformOps(&single_buf); |
||||
} |
||||
} |
||||
|
||||
private: |
||||
friend class internal::ClientAsyncResponseReaderFactory<R>; |
||||
::grpc::ClientContext* const context_; |
||||
::grpc::internal::Call call_; |
||||
bool started_; |
||||
bool initial_metadata_read_ = false; |
||||
|
||||
template <class W> |
||||
ClientAsyncResponseReader(::grpc::internal::Call call, |
||||
::grpc::ClientContext* context, const W& request, |
||||
bool start) |
||||
: context_(context), call_(call), started_(start) { |
||||
// Bind the metadata at time of StartCallInternal but set up the rest here
|
||||
// TODO(ctiller): don't assert
|
||||
GPR_CODEGEN_ASSERT(single_buf.SendMessage(request).ok()); |
||||
single_buf.ClientSendClose(); |
||||
if (start) StartCallInternal(); |
||||
} |
||||
|
||||
void StartCallInternal() { |
||||
single_buf.SendInitialMetadata(&context_->send_initial_metadata_, |
||||
context_->initial_metadata_flags()); |
||||
} |
||||
|
||||
// disable operator new
|
||||
static void* operator new(std::size_t size); |
||||
static void* operator new(std::size_t /*size*/, void* p) { return p; } |
||||
|
||||
::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata, |
||||
::grpc::internal::CallOpSendMessage, |
||||
::grpc::internal::CallOpClientSendClose, |
||||
::grpc::internal::CallOpRecvInitialMetadata, |
||||
::grpc::internal::CallOpRecvMessage<R>, |
||||
::grpc::internal::CallOpClientRecvStatus> |
||||
single_buf; |
||||
::grpc::internal::CallOpSet<::grpc::internal::CallOpRecvMessage<R>, |
||||
::grpc::internal::CallOpClientRecvStatus> |
||||
finish_buf; |
||||
}; |
||||
|
||||
/// Async server-side API for handling unary calls, where the single
|
||||
/// response message sent to the client is of type \a W.
|
||||
template <class W> |
||||
class ServerAsyncResponseWriter final |
||||
: public ::grpc::internal::ServerAsyncStreamingInterface { |
||||
public: |
||||
explicit ServerAsyncResponseWriter(::grpc::ServerContext* ctx) |
||||
: call_(nullptr, nullptr, nullptr), ctx_(ctx) {} |
||||
|
||||
/// See \a ServerAsyncStreamingInterface::SendInitialMetadata for semantics.
|
||||
///
|
||||
/// Side effect:
|
||||
/// 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_); |
||||
|
||||
meta_buf_.set_output_tag(tag); |
||||
meta_buf_.SendInitialMetadata(&ctx_->initial_metadata_, |
||||
ctx_->initial_metadata_flags()); |
||||
if (ctx_->compression_level_set()) { |
||||
meta_buf_.set_compression_level(ctx_->compression_level()); |
||||
} |
||||
ctx_->sent_initial_metadata_ = true; |
||||
call_.PerformOps(&meta_buf_); |
||||
} |
||||
|
||||
/// Indicate that the stream is to be finished and request notification
|
||||
/// when the server has sent the appropriate signals to the client to
|
||||
/// end the call. Should not be used concurrently with other operations.
|
||||
///
|
||||
/// \param[in] tag Tag identifying this request.
|
||||
/// \param[in] status To be sent to the client as the result of the call.
|
||||
/// \param[in] msg Message to be sent to the client.
|
||||
///
|
||||
/// Side effect:
|
||||
/// - also sends initial metadata if not already sent (using the
|
||||
/// \a ServerContext associated with this call).
|
||||
///
|
||||
/// Note: if \a status has a non-OK code, then \a msg will not be sent,
|
||||
/// and the client will receive only the status with possible trailing
|
||||
/// metadata.
|
||||
void Finish(const W& msg, const ::grpc::Status& status, void* tag) { |
||||
finish_buf_.set_output_tag(tag); |
||||
finish_buf_.set_core_cq_tag(&finish_buf_); |
||||
if (!ctx_->sent_initial_metadata_) { |
||||
finish_buf_.SendInitialMetadata(&ctx_->initial_metadata_, |
||||
ctx_->initial_metadata_flags()); |
||||
if (ctx_->compression_level_set()) { |
||||
finish_buf_.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_buf_.ServerSendStatus(&ctx_->trailing_metadata_, |
||||
finish_buf_.SendMessage(msg)); |
||||
} else { |
||||
finish_buf_.ServerSendStatus(&ctx_->trailing_metadata_, status); |
||||
} |
||||
call_.PerformOps(&finish_buf_); |
||||
} |
||||
|
||||
/// Indicate that the stream is to be finished with a non-OK status,
|
||||
/// and request notification for when the server has finished sending the
|
||||
/// appropriate signals to the client to end the call.
|
||||
/// Should not be used concurrently with other operations.
|
||||
///
|
||||
/// \param[in] tag Tag identifying this request.
|
||||
/// \param[in] status To be sent to the client as the result of the call.
|
||||
/// - Note: \a status must have a non-OK code.
|
||||
///
|
||||
/// Side effect:
|
||||
/// - also sends initial metadata if not already sent (using the
|
||||
/// \a ServerContext associated with this call).
|
||||
void FinishWithError(const ::grpc::Status& status, void* tag) { |
||||
GPR_CODEGEN_ASSERT(!status.ok()); |
||||
finish_buf_.set_output_tag(tag); |
||||
if (!ctx_->sent_initial_metadata_) { |
||||
finish_buf_.SendInitialMetadata(&ctx_->initial_metadata_, |
||||
ctx_->initial_metadata_flags()); |
||||
if (ctx_->compression_level_set()) { |
||||
finish_buf_.set_compression_level(ctx_->compression_level()); |
||||
} |
||||
ctx_->sent_initial_metadata_ = true; |
||||
} |
||||
finish_buf_.ServerSendStatus(&ctx_->trailing_metadata_, status); |
||||
call_.PerformOps(&finish_buf_); |
||||
} |
||||
|
||||
private: |
||||
void BindCall(::grpc::internal::Call* call) override { call_ = *call; } |
||||
|
||||
::grpc::internal::Call call_; |
||||
::grpc::ServerContext* ctx_; |
||||
::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata> |
||||
meta_buf_; |
||||
::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata, |
||||
::grpc::internal::CallOpSendMessage, |
||||
::grpc::internal::CallOpServerSendStatus> |
||||
finish_buf_; |
||||
}; |
||||
|
||||
} // namespace grpc_impl
|
||||
|
||||
namespace std { |
||||
template <class R> |
||||
class default_delete<::grpc_impl::ClientAsyncResponseReader<R>> { |
||||
public: |
||||
void operator()(void* /*p*/) {} |
||||
}; |
||||
template <class R> |
||||
class default_delete<::grpc_impl::ClientAsyncResponseReaderInterface<R>> { |
||||
public: |
||||
void operator()(void* /*p*/) {} |
||||
}; |
||||
} // namespace std
|
||||
|
||||
#endif // GRPCPP_IMPL_CODEGEN_ASYNC_UNARY_CALL_IMPL_H
|
||||
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
@ -1,385 +0,0 @@ |
||||
/*
|
||||
* |
||||
* 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 GRPCPP_IMPL_CODEGEN_METHOD_HANDLER_IMPL_H |
||||
#define GRPCPP_IMPL_CODEGEN_METHOD_HANDLER_IMPL_H |
||||
|
||||
#include <grpcpp/impl/codegen/byte_buffer.h> |
||||
#include <grpcpp/impl/codegen/core_codegen_interface.h> |
||||
#include <grpcpp/impl/codegen/rpc_service_method.h> |
||||
#include <grpcpp/impl/codegen/sync_stream_impl.h> |
||||
|
||||
namespace grpc_impl { |
||||
|
||||
namespace internal { |
||||
|
||||
// Invoke the method handler, fill in the status, and
|
||||
// return whether or not we finished safely (without an exception).
|
||||
// Note that exception handling is 0-cost in most compiler/library
|
||||
// implementations (except when an exception is actually thrown),
|
||||
// so this process doesn't require additional overhead in the common case.
|
||||
// Additionally, we don't need to return if we caught an exception or not;
|
||||
// the handling is the same in either case.
|
||||
template <class Callable> |
||||
::grpc::Status CatchingFunctionHandler(Callable&& handler) { |
||||
#if GRPC_ALLOW_EXCEPTIONS |
||||
try { |
||||
return handler(); |
||||
} catch (...) { |
||||
return ::grpc::Status(::grpc::StatusCode::UNKNOWN, |
||||
"Unexpected error in RPC handling"); |
||||
} |
||||
#else // GRPC_ALLOW_EXCEPTIONS
|
||||
return handler(); |
||||
#endif // GRPC_ALLOW_EXCEPTIONS
|
||||
} |
||||
|
||||
/// A wrapper class of an application provided rpc method handler.
|
||||
template <class ServiceType, class RequestType, class ResponseType> |
||||
class RpcMethodHandler : public ::grpc::internal::MethodHandler { |
||||
public: |
||||
RpcMethodHandler( |
||||
std::function<::grpc::Status(ServiceType*, ::grpc::ServerContext*, |
||||
const RequestType*, ResponseType*)> |
||||
func, |
||||
ServiceType* service) |
||||
: func_(func), service_(service) {} |
||||
|
||||
void RunHandler(const HandlerParameter& param) final { |
||||
ResponseType rsp; |
||||
::grpc::Status status = param.status; |
||||
if (status.ok()) { |
||||
status = CatchingFunctionHandler([this, ¶m, &rsp] { |
||||
return func_(service_, |
||||
static_cast<::grpc::ServerContext*>(param.server_context), |
||||
static_cast<RequestType*>(param.request), &rsp); |
||||
}); |
||||
static_cast<RequestType*>(param.request)->~RequestType(); |
||||
} |
||||
|
||||
GPR_CODEGEN_ASSERT(!param.server_context->sent_initial_metadata_); |
||||
::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata, |
||||
::grpc::internal::CallOpSendMessage, |
||||
::grpc::internal::CallOpServerSendStatus> |
||||
ops; |
||||
ops.SendInitialMetadata(¶m.server_context->initial_metadata_, |
||||
param.server_context->initial_metadata_flags()); |
||||
if (param.server_context->compression_level_set()) { |
||||
ops.set_compression_level(param.server_context->compression_level()); |
||||
} |
||||
if (status.ok()) { |
||||
status = ops.SendMessagePtr(&rsp); |
||||
} |
||||
ops.ServerSendStatus(¶m.server_context->trailing_metadata_, status); |
||||
param.call->PerformOps(&ops); |
||||
param.call->cq()->Pluck(&ops); |
||||
} |
||||
|
||||
void* Deserialize(grpc_call* call, grpc_byte_buffer* req, |
||||
::grpc::Status* status, void** /*handler_data*/) final { |
||||
::grpc::ByteBuffer buf; |
||||
buf.set_buffer(req); |
||||
auto* request = |
||||
new (::grpc::g_core_codegen_interface->grpc_call_arena_alloc( |
||||
call, sizeof(RequestType))) RequestType(); |
||||
*status = |
||||
::grpc::SerializationTraits<RequestType>::Deserialize(&buf, request); |
||||
buf.Release(); |
||||
if (status->ok()) { |
||||
return request; |
||||
} |
||||
request->~RequestType(); |
||||
return nullptr; |
||||
} |
||||
|
||||
private: |
||||
/// Application provided rpc handler function.
|
||||
std::function<::grpc::Status(ServiceType*, ::grpc::ServerContext*, |
||||
const RequestType*, ResponseType*)> |
||||
func_; |
||||
// The class the above handler function lives in.
|
||||
ServiceType* service_; |
||||
}; |
||||
|
||||
/// A wrapper class of an application provided client streaming handler.
|
||||
template <class ServiceType, class RequestType, class ResponseType> |
||||
class ClientStreamingHandler : public ::grpc::internal::MethodHandler { |
||||
public: |
||||
ClientStreamingHandler( |
||||
std::function<::grpc::Status(ServiceType*, ::grpc::ServerContext*, |
||||
::grpc_impl::ServerReader<RequestType>*, |
||||
ResponseType*)> |
||||
func, |
||||
ServiceType* service) |
||||
: func_(func), service_(service) {} |
||||
|
||||
void RunHandler(const HandlerParameter& param) final { |
||||
::grpc_impl::ServerReader<RequestType> reader( |
||||
param.call, static_cast<::grpc::ServerContext*>(param.server_context)); |
||||
ResponseType rsp; |
||||
::grpc::Status status = CatchingFunctionHandler([this, ¶m, &reader, |
||||
&rsp] { |
||||
return func_(service_, |
||||
static_cast<::grpc::ServerContext*>(param.server_context), |
||||
&reader, &rsp); |
||||
}); |
||||
|
||||
::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata, |
||||
::grpc::internal::CallOpSendMessage, |
||||
::grpc::internal::CallOpServerSendStatus> |
||||
ops; |
||||
if (!param.server_context->sent_initial_metadata_) { |
||||
ops.SendInitialMetadata(¶m.server_context->initial_metadata_, |
||||
param.server_context->initial_metadata_flags()); |
||||
if (param.server_context->compression_level_set()) { |
||||
ops.set_compression_level(param.server_context->compression_level()); |
||||
} |
||||
} |
||||
if (status.ok()) { |
||||
status = ops.SendMessagePtr(&rsp); |
||||
} |
||||
ops.ServerSendStatus(¶m.server_context->trailing_metadata_, status); |
||||
param.call->PerformOps(&ops); |
||||
param.call->cq()->Pluck(&ops); |
||||
} |
||||
|
||||
private: |
||||
std::function<::grpc::Status(ServiceType*, ::grpc::ServerContext*, |
||||
::grpc_impl::ServerReader<RequestType>*, |
||||
ResponseType*)> |
||||
func_; |
||||
ServiceType* service_; |
||||
}; |
||||
|
||||
/// A wrapper class of an application provided server streaming handler.
|
||||
template <class ServiceType, class RequestType, class ResponseType> |
||||
class ServerStreamingHandler : public ::grpc::internal::MethodHandler { |
||||
public: |
||||
ServerStreamingHandler( |
||||
std::function<::grpc::Status(ServiceType*, ::grpc::ServerContext*, |
||||
const RequestType*, |
||||
::grpc_impl::ServerWriter<ResponseType>*)> |
||||
func, |
||||
ServiceType* service) |
||||
: func_(func), service_(service) {} |
||||
|
||||
void RunHandler(const HandlerParameter& param) final { |
||||
::grpc::Status status = param.status; |
||||
if (status.ok()) { |
||||
::grpc_impl::ServerWriter<ResponseType> writer( |
||||
param.call, |
||||
static_cast<::grpc::ServerContext*>(param.server_context)); |
||||
status = CatchingFunctionHandler([this, ¶m, &writer] { |
||||
return func_(service_, |
||||
static_cast<::grpc::ServerContext*>(param.server_context), |
||||
static_cast<RequestType*>(param.request), &writer); |
||||
}); |
||||
static_cast<RequestType*>(param.request)->~RequestType(); |
||||
} |
||||
|
||||
::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata, |
||||
::grpc::internal::CallOpServerSendStatus> |
||||
ops; |
||||
if (!param.server_context->sent_initial_metadata_) { |
||||
ops.SendInitialMetadata(¶m.server_context->initial_metadata_, |
||||
param.server_context->initial_metadata_flags()); |
||||
if (param.server_context->compression_level_set()) { |
||||
ops.set_compression_level(param.server_context->compression_level()); |
||||
} |
||||
} |
||||
ops.ServerSendStatus(¶m.server_context->trailing_metadata_, status); |
||||
param.call->PerformOps(&ops); |
||||
if (param.server_context->has_pending_ops_) { |
||||
param.call->cq()->Pluck(¶m.server_context->pending_ops_); |
||||
} |
||||
param.call->cq()->Pluck(&ops); |
||||
} |
||||
|
||||
void* Deserialize(grpc_call* call, grpc_byte_buffer* req, |
||||
::grpc::Status* status, void** /*handler_data*/) final { |
||||
::grpc::ByteBuffer buf; |
||||
buf.set_buffer(req); |
||||
auto* request = |
||||
new (::grpc::g_core_codegen_interface->grpc_call_arena_alloc( |
||||
call, sizeof(RequestType))) RequestType(); |
||||
*status = |
||||
::grpc::SerializationTraits<RequestType>::Deserialize(&buf, request); |
||||
buf.Release(); |
||||
if (status->ok()) { |
||||
return request; |
||||
} |
||||
request->~RequestType(); |
||||
return nullptr; |
||||
} |
||||
|
||||
private: |
||||
std::function<::grpc::Status(ServiceType*, ::grpc::ServerContext*, |
||||
const RequestType*, |
||||
::grpc_impl::ServerWriter<ResponseType>*)> |
||||
func_; |
||||
ServiceType* service_; |
||||
}; |
||||
|
||||
/// A wrapper class of an application provided bidi-streaming handler.
|
||||
/// This also applies to server-streamed implementation of a unary method
|
||||
/// with the additional requirement that such methods must have done a
|
||||
/// write for status to be ok
|
||||
/// Since this is used by more than 1 class, the service is not passed in.
|
||||
/// Instead, it is expected to be an implicitly-captured argument of func
|
||||
/// (through bind or something along those lines)
|
||||
template <class Streamer, bool WriteNeeded> |
||||
class TemplatedBidiStreamingHandler : public ::grpc::internal::MethodHandler { |
||||
public: |
||||
TemplatedBidiStreamingHandler( |
||||
std::function<::grpc::Status(::grpc::ServerContext*, Streamer*)> func) |
||||
: func_(func), write_needed_(WriteNeeded) {} |
||||
|
||||
void RunHandler(const HandlerParameter& param) final { |
||||
Streamer stream(param.call, |
||||
static_cast<::grpc::ServerContext*>(param.server_context)); |
||||
::grpc::Status status = CatchingFunctionHandler([this, ¶m, &stream] { |
||||
return func_(static_cast<::grpc::ServerContext*>(param.server_context), |
||||
&stream); |
||||
}); |
||||
|
||||
::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata, |
||||
::grpc::internal::CallOpServerSendStatus> |
||||
ops; |
||||
if (!param.server_context->sent_initial_metadata_) { |
||||
ops.SendInitialMetadata(¶m.server_context->initial_metadata_, |
||||
param.server_context->initial_metadata_flags()); |
||||
if (param.server_context->compression_level_set()) { |
||||
ops.set_compression_level(param.server_context->compression_level()); |
||||
} |
||||
if (write_needed_ && status.ok()) { |
||||
// If we needed a write but never did one, we need to mark the
|
||||
// status as a fail
|
||||
status = ::grpc::Status(::grpc::StatusCode::INTERNAL, |
||||
"Service did not provide response message"); |
||||
} |
||||
} |
||||
ops.ServerSendStatus(¶m.server_context->trailing_metadata_, status); |
||||
param.call->PerformOps(&ops); |
||||
if (param.server_context->has_pending_ops_) { |
||||
param.call->cq()->Pluck(¶m.server_context->pending_ops_); |
||||
} |
||||
param.call->cq()->Pluck(&ops); |
||||
} |
||||
|
||||
private: |
||||
std::function<::grpc::Status(::grpc::ServerContext*, Streamer*)> func_; |
||||
const bool write_needed_; |
||||
}; |
||||
|
||||
template <class ServiceType, class RequestType, class ResponseType> |
||||
class BidiStreamingHandler |
||||
: public TemplatedBidiStreamingHandler< |
||||
::grpc_impl::ServerReaderWriter<ResponseType, RequestType>, false> { |
||||
public: |
||||
BidiStreamingHandler( |
||||
std::function<::grpc::Status( |
||||
ServiceType*, ::grpc::ServerContext*, |
||||
::grpc_impl::ServerReaderWriter<ResponseType, RequestType>*)> |
||||
func, |
||||
ServiceType* service) |
||||
// TODO(vjpai): When gRPC supports C++14, move-capture func in the below
|
||||
: TemplatedBidiStreamingHandler< |
||||
::grpc_impl::ServerReaderWriter<ResponseType, RequestType>, false>( |
||||
[func, service]( |
||||
::grpc::ServerContext* ctx, |
||||
::grpc_impl::ServerReaderWriter<ResponseType, RequestType>* |
||||
streamer) { return func(service, ctx, streamer); }) {} |
||||
}; |
||||
|
||||
template <class RequestType, class ResponseType> |
||||
class StreamedUnaryHandler |
||||
: public TemplatedBidiStreamingHandler< |
||||
::grpc_impl::ServerUnaryStreamer<RequestType, ResponseType>, true> { |
||||
public: |
||||
explicit StreamedUnaryHandler( |
||||
std::function<::grpc::Status( |
||||
::grpc::ServerContext*, |
||||
::grpc_impl::ServerUnaryStreamer<RequestType, ResponseType>*)> |
||||
func) |
||||
: TemplatedBidiStreamingHandler< |
||||
::grpc_impl::ServerUnaryStreamer<RequestType, ResponseType>, true>( |
||||
std::move(func)) {} |
||||
}; |
||||
|
||||
template <class RequestType, class ResponseType> |
||||
class SplitServerStreamingHandler |
||||
: public TemplatedBidiStreamingHandler< |
||||
::grpc_impl::ServerSplitStreamer<RequestType, ResponseType>, false> { |
||||
public: |
||||
explicit SplitServerStreamingHandler( |
||||
std::function<::grpc::Status( |
||||
::grpc::ServerContext*, |
||||
::grpc_impl::ServerSplitStreamer<RequestType, ResponseType>*)> |
||||
func) |
||||
: TemplatedBidiStreamingHandler< |
||||
::grpc_impl::ServerSplitStreamer<RequestType, ResponseType>, false>( |
||||
std::move(func)) {} |
||||
}; |
||||
|
||||
/// General method handler class for errors that prevent real method use
|
||||
/// e.g., handle unknown method by returning UNIMPLEMENTED error.
|
||||
template <::grpc::StatusCode code> |
||||
class ErrorMethodHandler : public ::grpc::internal::MethodHandler { |
||||
public: |
||||
template <class T> |
||||
static void FillOps(::grpc::ServerContextBase* context, T* ops) { |
||||
::grpc::Status status(code, ""); |
||||
if (!context->sent_initial_metadata_) { |
||||
ops->SendInitialMetadata(&context->initial_metadata_, |
||||
context->initial_metadata_flags()); |
||||
if (context->compression_level_set()) { |
||||
ops->set_compression_level(context->compression_level()); |
||||
} |
||||
context->sent_initial_metadata_ = true; |
||||
} |
||||
ops->ServerSendStatus(&context->trailing_metadata_, status); |
||||
} |
||||
|
||||
void RunHandler(const HandlerParameter& param) final { |
||||
::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata, |
||||
::grpc::internal::CallOpServerSendStatus> |
||||
ops; |
||||
FillOps(param.server_context, &ops); |
||||
param.call->PerformOps(&ops); |
||||
param.call->cq()->Pluck(&ops); |
||||
} |
||||
|
||||
void* Deserialize(grpc_call* /*call*/, grpc_byte_buffer* req, |
||||
::grpc::Status* /*status*/, void** /*handler_data*/) final { |
||||
// We have to destroy any request payload
|
||||
if (req != nullptr) { |
||||
::grpc::g_core_codegen_interface->grpc_byte_buffer_destroy(req); |
||||
} |
||||
return nullptr; |
||||
} |
||||
}; |
||||
|
||||
typedef ErrorMethodHandler<::grpc::StatusCode::UNIMPLEMENTED> |
||||
UnknownMethodHandler; |
||||
typedef ErrorMethodHandler<::grpc::StatusCode::RESOURCE_EXHAUSTED> |
||||
ResourceExhaustedHandler; |
||||
|
||||
} // namespace internal
|
||||
} // namespace grpc_impl
|
||||
|
||||
#endif // GRPCPP_IMPL_CODEGEN_METHOD_HANDLER_IMPL_H
|
||||
@ -1,783 +0,0 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2019 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 GRPCPP_IMPL_CODEGEN_SERVER_CALLBACK_IMPL_H |
||||
#define GRPCPP_IMPL_CODEGEN_SERVER_CALLBACK_IMPL_H |
||||
|
||||
#include <atomic> |
||||
#include <functional> |
||||
#include <type_traits> |
||||
|
||||
#include <grpcpp/impl/codegen/call.h> |
||||
#include <grpcpp/impl/codegen/call_op_set.h> |
||||
#include <grpcpp/impl/codegen/callback_common.h> |
||||
#include <grpcpp/impl/codegen/config.h> |
||||
#include <grpcpp/impl/codegen/core_codegen_interface.h> |
||||
#include <grpcpp/impl/codegen/message_allocator.h> |
||||
#include <grpcpp/impl/codegen/status.h> |
||||
|
||||
namespace grpc_impl { |
||||
|
||||
// Declare base class of all reactors as internal
|
||||
namespace internal { |
||||
|
||||
// Forward declarations
|
||||
template <class Request, class Response> |
||||
class CallbackUnaryHandler; |
||||
template <class Request, class Response> |
||||
class CallbackClientStreamingHandler; |
||||
template <class Request, class Response> |
||||
class CallbackServerStreamingHandler; |
||||
template <class Request, class Response> |
||||
class CallbackBidiHandler; |
||||
|
||||
class ServerReactor { |
||||
public: |
||||
virtual ~ServerReactor() = default; |
||||
virtual void OnDone() = 0; |
||||
virtual void OnCancel() = 0; |
||||
|
||||
// The following is not API. It is for internal use only and specifies whether
|
||||
// all reactions of this Reactor can be run without an extra executor
|
||||
// scheduling. This should only be used for internally-defined reactors with
|
||||
// trivial reactions.
|
||||
virtual bool InternalInlineable() { return false; } |
||||
|
||||
private: |
||||
template <class Request, class Response> |
||||
friend class CallbackUnaryHandler; |
||||
template <class Request, class Response> |
||||
friend class CallbackClientStreamingHandler; |
||||
template <class Request, class Response> |
||||
friend class CallbackServerStreamingHandler; |
||||
template <class Request, class Response> |
||||
friend class CallbackBidiHandler; |
||||
}; |
||||
|
||||
/// The base class of ServerCallbackUnary etc.
|
||||
class ServerCallbackCall { |
||||
public: |
||||
virtual ~ServerCallbackCall() {} |
||||
|
||||
// This object is responsible for tracking when it is safe to call OnDone and
|
||||
// OnCancel. OnDone should not be called until the method handler is complete,
|
||||
// Finish has been called, the ServerContext CompletionOp (which tracks
|
||||
// cancellation or successful completion) has completed, and all outstanding
|
||||
// Read/Write actions have seen their reactions. OnCancel should not be called
|
||||
// until after the method handler is done and the RPC has completed with a
|
||||
// cancellation. This is tracked by counting how many of these conditions have
|
||||
// been met and calling OnCancel when none remain unmet.
|
||||
|
||||
// Public versions of MaybeDone: one where we don't know the reactor in
|
||||
// advance (used for the ServerContext CompletionOp), and one for where we
|
||||
// know the inlineability of the OnDone reaction. You should set the inline
|
||||
// flag to true if either the Reactor is InternalInlineable() or if this
|
||||
// callback is already being forced to run dispatched to an executor
|
||||
// (typically because it contains additional work than just the MaybeDone).
|
||||
|
||||
void MaybeDone() { |
||||
if (GPR_UNLIKELY(Unref() == 1)) { |
||||
ScheduleOnDone(reactor()->InternalInlineable()); |
||||
} |
||||
} |
||||
|
||||
void MaybeDone(bool inline_ondone) { |
||||
if (GPR_UNLIKELY(Unref() == 1)) { |
||||
ScheduleOnDone(inline_ondone); |
||||
} |
||||
} |
||||
|
||||
// Fast version called with known reactor passed in, used from derived
|
||||
// classes, typically in non-cancel case
|
||||
void MaybeCallOnCancel(ServerReactor* reactor) { |
||||
if (GPR_UNLIKELY(UnblockCancellation())) { |
||||
CallOnCancel(reactor); |
||||
} |
||||
} |
||||
|
||||
// Slower version called from object that doesn't know the reactor a priori
|
||||
// (such as the ServerContext CompletionOp which is formed before the
|
||||
// reactor). This is used in cancel cases only, so it's ok to be slower and
|
||||
// invoke a virtual function.
|
||||
void MaybeCallOnCancel() { |
||||
if (GPR_UNLIKELY(UnblockCancellation())) { |
||||
CallOnCancel(reactor()); |
||||
} |
||||
} |
||||
|
||||
protected: |
||||
/// Increases the reference count
|
||||
void Ref() { callbacks_outstanding_.fetch_add(1, std::memory_order_relaxed); } |
||||
|
||||
private: |
||||
virtual ServerReactor* reactor() = 0; |
||||
|
||||
// CallOnDone performs the work required at completion of the RPC: invoking
|
||||
// the OnDone function and doing all necessary cleanup. This function is only
|
||||
// ever invoked on a fully-Unref'fed ServerCallbackCall.
|
||||
virtual void CallOnDone() = 0; |
||||
|
||||
// If the OnDone reaction is inlineable, execute it inline. Otherwise send it
|
||||
// to an executor.
|
||||
void ScheduleOnDone(bool inline_ondone); |
||||
|
||||
// If the OnCancel reaction is inlineable, execute it inline. Otherwise send
|
||||
// it to an executor.
|
||||
void CallOnCancel(ServerReactor* reactor); |
||||
|
||||
// Implement the cancellation constraint counter. Return true if OnCancel
|
||||
// should be called, false otherwise.
|
||||
bool UnblockCancellation() { |
||||
return on_cancel_conditions_remaining_.fetch_sub( |
||||
1, std::memory_order_acq_rel) == 1; |
||||
} |
||||
|
||||
/// Decreases the reference count and returns the previous value
|
||||
int Unref() { |
||||
return callbacks_outstanding_.fetch_sub(1, std::memory_order_acq_rel); |
||||
} |
||||
|
||||
std::atomic_int on_cancel_conditions_remaining_{2}; |
||||
std::atomic_int callbacks_outstanding_{ |
||||
3}; // reserve for start, Finish, and CompletionOp
|
||||
}; |
||||
|
||||
template <class Request, class Response> |
||||
class DefaultMessageHolder |
||||
: public ::grpc::experimental::MessageHolder<Request, Response> { |
||||
public: |
||||
DefaultMessageHolder() { |
||||
this->set_request(&request_obj_); |
||||
this->set_response(&response_obj_); |
||||
} |
||||
void Release() override { |
||||
// the object is allocated in the call arena.
|
||||
this->~DefaultMessageHolder<Request, Response>(); |
||||
} |
||||
|
||||
private: |
||||
Request request_obj_; |
||||
Response response_obj_; |
||||
}; |
||||
|
||||
} // namespace internal
|
||||
|
||||
// Forward declarations
|
||||
class ServerUnaryReactor; |
||||
template <class Request> |
||||
class ServerReadReactor; |
||||
template <class Response> |
||||
class ServerWriteReactor; |
||||
template <class Request, class Response> |
||||
class ServerBidiReactor; |
||||
|
||||
// NOTE: The actual call/stream object classes are provided as API only to
|
||||
// support mocking. There are no implementations of these class interfaces in
|
||||
// the API.
|
||||
class ServerCallbackUnary : public internal::ServerCallbackCall { |
||||
public: |
||||
virtual ~ServerCallbackUnary() {} |
||||
virtual void Finish(::grpc::Status s) = 0; |
||||
virtual void SendInitialMetadata() = 0; |
||||
|
||||
protected: |
||||
// Use a template rather than explicitly specifying ServerUnaryReactor to
|
||||
// delay binding and avoid a circular forward declaration issue
|
||||
template <class Reactor> |
||||
void BindReactor(Reactor* reactor) { |
||||
reactor->InternalBindCall(this); |
||||
} |
||||
}; |
||||
|
||||
template <class Request> |
||||
class ServerCallbackReader : public internal::ServerCallbackCall { |
||||
public: |
||||
virtual ~ServerCallbackReader() {} |
||||
virtual void Finish(::grpc::Status s) = 0; |
||||
virtual void SendInitialMetadata() = 0; |
||||
virtual void Read(Request* msg) = 0; |
||||
|
||||
protected: |
||||
void BindReactor(ServerReadReactor<Request>* reactor) { |
||||
reactor->InternalBindReader(this); |
||||
} |
||||
}; |
||||
|
||||
template <class Response> |
||||
class ServerCallbackWriter : public internal::ServerCallbackCall { |
||||
public: |
||||
virtual ~ServerCallbackWriter() {} |
||||
|
||||
virtual void Finish(::grpc::Status s) = 0; |
||||
virtual void SendInitialMetadata() = 0; |
||||
virtual void Write(const Response* msg, ::grpc::WriteOptions options) = 0; |
||||
virtual void WriteAndFinish(const Response* msg, ::grpc::WriteOptions options, |
||||
::grpc::Status s) = 0; |
||||
|
||||
protected: |
||||
void BindReactor(ServerWriteReactor<Response>* reactor) { |
||||
reactor->InternalBindWriter(this); |
||||
} |
||||
}; |
||||
|
||||
template <class Request, class Response> |
||||
class ServerCallbackReaderWriter : public internal::ServerCallbackCall { |
||||
public: |
||||
virtual ~ServerCallbackReaderWriter() {} |
||||
|
||||
virtual void Finish(::grpc::Status s) = 0; |
||||
virtual void SendInitialMetadata() = 0; |
||||
virtual void Read(Request* msg) = 0; |
||||
virtual void Write(const Response* msg, ::grpc::WriteOptions options) = 0; |
||||
virtual void WriteAndFinish(const Response* msg, ::grpc::WriteOptions options, |
||||
::grpc::Status s) = 0; |
||||
|
||||
protected: |
||||
void BindReactor(ServerBidiReactor<Request, Response>* reactor) { |
||||
reactor->InternalBindStream(this); |
||||
} |
||||
}; |
||||
|
||||
// The following classes are the reactor interfaces that are to be implemented
|
||||
// by the user, returned as the output parameter of the method handler for a
|
||||
// callback method. Note that none of the classes are pure; all reactions have a
|
||||
// default empty reaction so that the user class only needs to override those
|
||||
// classes that it cares about.
|
||||
|
||||
/// \a ServerBidiReactor is the interface for a bidirectional streaming RPC.
|
||||
template <class Request, class Response> |
||||
class ServerBidiReactor : public internal::ServerReactor { |
||||
public: |
||||
// NOTE: Initializing stream_ as a constructor initializer rather than a
|
||||
// default initializer because gcc-4.x requires a copy constructor for
|
||||
// default initializing a templated member, which isn't ok for atomic.
|
||||
// TODO(vjpai): Switch to default constructor and default initializer when
|
||||
// gcc-4.x is no longer supported
|
||||
ServerBidiReactor() : stream_(nullptr) {} |
||||
~ServerBidiReactor() = default; |
||||
|
||||
/// Send any initial metadata stored in the RPC context. If not invoked,
|
||||
/// any initial metadata will be passed along with the first Write or the
|
||||
/// Finish (if there are no writes).
|
||||
void StartSendInitialMetadata() { |
||||
ServerCallbackReaderWriter<Request, Response>* stream = |
||||
stream_.load(std::memory_order_acquire); |
||||
if (stream == nullptr) { |
||||
grpc::internal::MutexLock l(&stream_mu_); |
||||
stream = stream_.load(std::memory_order_relaxed); |
||||
if (stream == nullptr) { |
||||
backlog_.send_initial_metadata_wanted = true; |
||||
return; |
||||
} |
||||
} |
||||
stream->SendInitialMetadata(); |
||||
} |
||||
|
||||
/// Initiate a read operation.
|
||||
///
|
||||
/// \param[out] req Where to eventually store the read message. Valid when
|
||||
/// the library calls OnReadDone
|
||||
void StartRead(Request* req) { |
||||
ServerCallbackReaderWriter<Request, Response>* stream = |
||||
stream_.load(std::memory_order_acquire); |
||||
if (stream == nullptr) { |
||||
grpc::internal::MutexLock l(&stream_mu_); |
||||
stream = stream_.load(std::memory_order_relaxed); |
||||
if (stream == nullptr) { |
||||
backlog_.read_wanted = req; |
||||
return; |
||||
} |
||||
} |
||||
stream->Read(req); |
||||
} |
||||
|
||||
/// Initiate a write operation.
|
||||
///
|
||||
/// \param[in] resp The message to be written. The library does not take
|
||||
/// ownership but the caller must ensure that the message is
|
||||
/// not deleted or modified until OnWriteDone is called.
|
||||
void StartWrite(const Response* resp) { |
||||
StartWrite(resp, ::grpc::WriteOptions()); |
||||
} |
||||
|
||||
/// Initiate a write operation with specified options.
|
||||
///
|
||||
/// \param[in] resp The message to be written. The library does not take
|
||||
/// ownership but the caller must ensure that the message is
|
||||
/// not deleted or modified until OnWriteDone is called.
|
||||
/// \param[in] options The WriteOptions to use for writing this message
|
||||
void StartWrite(const Response* resp, ::grpc::WriteOptions options) { |
||||
ServerCallbackReaderWriter<Request, Response>* stream = |
||||
stream_.load(std::memory_order_acquire); |
||||
if (stream == nullptr) { |
||||
grpc::internal::MutexLock l(&stream_mu_); |
||||
stream = stream_.load(std::memory_order_relaxed); |
||||
if (stream == nullptr) { |
||||
backlog_.write_wanted = resp; |
||||
backlog_.write_options_wanted = std::move(options); |
||||
return; |
||||
} |
||||
} |
||||
stream->Write(resp, std::move(options)); |
||||
} |
||||
|
||||
/// Initiate a write operation with specified options and final RPC Status,
|
||||
/// which also causes any trailing metadata for this RPC to be sent out.
|
||||
/// StartWriteAndFinish is like merging StartWriteLast and Finish into a
|
||||
/// single step. A key difference, though, is that this operation doesn't have
|
||||
/// an OnWriteDone reaction - it is considered complete only when OnDone is
|
||||
/// available. An RPC can either have StartWriteAndFinish or Finish, but not
|
||||
/// both.
|
||||
///
|
||||
/// \param[in] resp The message to be written. The library does not take
|
||||
/// ownership but the caller must ensure that the message is
|
||||
/// not deleted or modified until OnDone is called.
|
||||
/// \param[in] options The WriteOptions to use for writing this message
|
||||
/// \param[in] s The status outcome of this RPC
|
||||
void StartWriteAndFinish(const Response* resp, ::grpc::WriteOptions options, |
||||
::grpc::Status s) { |
||||
ServerCallbackReaderWriter<Request, Response>* stream = |
||||
stream_.load(std::memory_order_acquire); |
||||
if (stream == nullptr) { |
||||
grpc::internal::MutexLock l(&stream_mu_); |
||||
stream = stream_.load(std::memory_order_relaxed); |
||||
if (stream == nullptr) { |
||||
backlog_.write_and_finish_wanted = true; |
||||
backlog_.write_wanted = resp; |
||||
backlog_.write_options_wanted = std::move(options); |
||||
backlog_.status_wanted = std::move(s); |
||||
return; |
||||
} |
||||
} |
||||
stream->WriteAndFinish(resp, std::move(options), std::move(s)); |
||||
} |
||||
|
||||
/// Inform system of a planned write operation with specified options, but
|
||||
/// allow the library to schedule the actual write coalesced with the writing
|
||||
/// of trailing metadata (which takes place on a Finish call).
|
||||
///
|
||||
/// \param[in] resp The message to be written. The library does not take
|
||||
/// ownership but the caller must ensure that the message is
|
||||
/// not deleted or modified until OnWriteDone is called.
|
||||
/// \param[in] options The WriteOptions to use for writing this message
|
||||
void StartWriteLast(const Response* resp, ::grpc::WriteOptions options) { |
||||
StartWrite(resp, std::move(options.set_last_message())); |
||||
} |
||||
|
||||
/// Indicate that the stream is to be finished and the trailing metadata and
|
||||
/// RPC status are to be sent. Every RPC MUST be finished using either Finish
|
||||
/// or StartWriteAndFinish (but not both), even if the RPC is already
|
||||
/// cancelled.
|
||||
///
|
||||
/// \param[in] s The status outcome of this RPC
|
||||
void Finish(::grpc::Status s) { |
||||
ServerCallbackReaderWriter<Request, Response>* stream = |
||||
stream_.load(std::memory_order_acquire); |
||||
if (stream == nullptr) { |
||||
grpc::internal::MutexLock l(&stream_mu_); |
||||
stream = stream_.load(std::memory_order_relaxed); |
||||
if (stream == nullptr) { |
||||
backlog_.finish_wanted = true; |
||||
backlog_.status_wanted = std::move(s); |
||||
return; |
||||
} |
||||
} |
||||
stream->Finish(std::move(s)); |
||||
} |
||||
|
||||
/// Notifies the application that an explicit StartSendInitialMetadata
|
||||
/// operation completed. Not used when the sending of initial metadata
|
||||
/// piggybacks onto the first write.
|
||||
///
|
||||
/// \param[in] ok Was it successful? If false, no further write-side operation
|
||||
/// will succeed.
|
||||
virtual void OnSendInitialMetadataDone(bool /*ok*/) {} |
||||
|
||||
/// Notifies the application that a StartRead operation completed.
|
||||
///
|
||||
/// \param[in] ok Was it successful? If false, no further read-side operation
|
||||
/// will succeed.
|
||||
virtual void OnReadDone(bool /*ok*/) {} |
||||
|
||||
/// Notifies the application that a StartWrite (or StartWriteLast) operation
|
||||
/// completed.
|
||||
///
|
||||
/// \param[in] ok Was it successful? If false, no further write-side operation
|
||||
/// will succeed.
|
||||
virtual void OnWriteDone(bool /*ok*/) {} |
||||
|
||||
/// Notifies the application that all operations associated with this RPC
|
||||
/// have completed. This is an override (from the internal base class) but
|
||||
/// still abstract, so derived classes MUST override it to be instantiated.
|
||||
void OnDone() override = 0; |
||||
|
||||
/// Notifies the application that this RPC has been cancelled. This is an
|
||||
/// override (from the internal base class) but not final, so derived classes
|
||||
/// should override it if they want to take action.
|
||||
void OnCancel() override {} |
||||
|
||||
private: |
||||
friend class ServerCallbackReaderWriter<Request, Response>; |
||||
// May be overridden by internal implementation details. This is not a public
|
||||
// customization point.
|
||||
virtual void InternalBindStream( |
||||
ServerCallbackReaderWriter<Request, Response>* stream) { |
||||
// TODO(vjpai): When stream_or_backlog_ becomes a variant (see below), use
|
||||
// a scoped MutexLock and std::swap stream_or_backlog_ with a variant that
|
||||
// has stream, then std::get<PreBindBacklog> out of that after the lock.
|
||||
// Do likewise with the remaining InternalBind* functions as well.
|
||||
grpc::internal::ReleasableMutexLock l(&stream_mu_); |
||||
PreBindBacklog ops(std::move(backlog_)); |
||||
stream_.store(stream, std::memory_order_release); |
||||
l.Unlock(); |
||||
|
||||
if (ops.send_initial_metadata_wanted) { |
||||
stream->SendInitialMetadata(); |
||||
} |
||||
if (ops.read_wanted != nullptr) { |
||||
stream->Read(ops.read_wanted); |
||||
} |
||||
if (ops.write_and_finish_wanted) { |
||||
stream->WriteAndFinish(ops.write_wanted, |
||||
std::move(ops.write_options_wanted), |
||||
std::move(ops.status_wanted)); |
||||
} else { |
||||
if (ops.write_wanted != nullptr) { |
||||
stream->Write(ops.write_wanted, std::move(ops.write_options_wanted)); |
||||
} |
||||
if (ops.finish_wanted) { |
||||
stream->Finish(std::move(ops.status_wanted)); |
||||
} |
||||
} |
||||
} |
||||
|
||||
grpc::internal::Mutex stream_mu_; |
||||
// TODO(vjpai): Make stream_or_backlog_ into a std::variant or absl::variant
|
||||
// once C++17 or ABSL is supported since stream and backlog are
|
||||
// mutually exclusive in this class. Do likewise with the
|
||||
// remaining reactor classes and their backlogs as well.
|
||||
std::atomic<ServerCallbackReaderWriter<Request, Response>*> stream_{nullptr}; |
||||
struct PreBindBacklog { |
||||
bool send_initial_metadata_wanted = false; |
||||
bool write_and_finish_wanted = false; |
||||
bool finish_wanted = false; |
||||
Request* read_wanted = nullptr; |
||||
const Response* write_wanted = nullptr; |
||||
::grpc::WriteOptions write_options_wanted; |
||||
::grpc::Status status_wanted; |
||||
}; |
||||
PreBindBacklog backlog_ /* GUARDED_BY(stream_mu_) */; |
||||
}; |
||||
|
||||
/// \a ServerReadReactor is the interface for a client-streaming RPC.
|
||||
template <class Request> |
||||
class ServerReadReactor : public internal::ServerReactor { |
||||
public: |
||||
ServerReadReactor() : reader_(nullptr) {} |
||||
~ServerReadReactor() = default; |
||||
|
||||
/// The following operation initiations are exactly like ServerBidiReactor.
|
||||
void StartSendInitialMetadata() { |
||||
ServerCallbackReader<Request>* reader = |
||||
reader_.load(std::memory_order_acquire); |
||||
if (reader == nullptr) { |
||||
grpc::internal::MutexLock l(&reader_mu_); |
||||
reader = reader_.load(std::memory_order_relaxed); |
||||
if (reader == nullptr) { |
||||
backlog_.send_initial_metadata_wanted = true; |
||||
return; |
||||
} |
||||
} |
||||
reader->SendInitialMetadata(); |
||||
} |
||||
void StartRead(Request* req) { |
||||
ServerCallbackReader<Request>* reader = |
||||
reader_.load(std::memory_order_acquire); |
||||
if (reader == nullptr) { |
||||
grpc::internal::MutexLock l(&reader_mu_); |
||||
reader = reader_.load(std::memory_order_relaxed); |
||||
if (reader == nullptr) { |
||||
backlog_.read_wanted = req; |
||||
return; |
||||
} |
||||
} |
||||
reader->Read(req); |
||||
} |
||||
void Finish(::grpc::Status s) { |
||||
ServerCallbackReader<Request>* reader = |
||||
reader_.load(std::memory_order_acquire); |
||||
if (reader == nullptr) { |
||||
grpc::internal::MutexLock l(&reader_mu_); |
||||
reader = reader_.load(std::memory_order_relaxed); |
||||
if (reader == nullptr) { |
||||
backlog_.finish_wanted = true; |
||||
backlog_.status_wanted = std::move(s); |
||||
return; |
||||
} |
||||
} |
||||
reader->Finish(std::move(s)); |
||||
} |
||||
|
||||
/// The following notifications are exactly like ServerBidiReactor.
|
||||
virtual void OnSendInitialMetadataDone(bool /*ok*/) {} |
||||
virtual void OnReadDone(bool /*ok*/) {} |
||||
void OnDone() override = 0; |
||||
void OnCancel() override {} |
||||
|
||||
private: |
||||
friend class ServerCallbackReader<Request>; |
||||
|
||||
// May be overridden by internal implementation details. This is not a public
|
||||
// customization point.
|
||||
virtual void InternalBindReader(ServerCallbackReader<Request>* reader) { |
||||
grpc::internal::ReleasableMutexLock l(&reader_mu_); |
||||
PreBindBacklog ops(std::move(backlog_)); |
||||
reader_.store(reader, std::memory_order_release); |
||||
l.Unlock(); |
||||
|
||||
if (ops.send_initial_metadata_wanted) { |
||||
reader->SendInitialMetadata(); |
||||
} |
||||
if (ops.read_wanted != nullptr) { |
||||
reader->Read(ops.read_wanted); |
||||
} |
||||
if (ops.finish_wanted) { |
||||
reader->Finish(std::move(ops.status_wanted)); |
||||
} |
||||
} |
||||
|
||||
grpc::internal::Mutex reader_mu_; |
||||
std::atomic<ServerCallbackReader<Request>*> reader_{nullptr}; |
||||
struct PreBindBacklog { |
||||
bool send_initial_metadata_wanted = false; |
||||
bool finish_wanted = false; |
||||
Request* read_wanted = nullptr; |
||||
::grpc::Status status_wanted; |
||||
}; |
||||
PreBindBacklog backlog_ /* GUARDED_BY(reader_mu_) */; |
||||
}; |
||||
|
||||
/// \a ServerWriteReactor is the interface for a server-streaming RPC.
|
||||
template <class Response> |
||||
class ServerWriteReactor : public internal::ServerReactor { |
||||
public: |
||||
ServerWriteReactor() : writer_(nullptr) {} |
||||
~ServerWriteReactor() = default; |
||||
|
||||
/// The following operation initiations are exactly like ServerBidiReactor.
|
||||
void StartSendInitialMetadata() { |
||||
ServerCallbackWriter<Response>* writer = |
||||
writer_.load(std::memory_order_acquire); |
||||
if (writer == nullptr) { |
||||
grpc::internal::MutexLock l(&writer_mu_); |
||||
writer = writer_.load(std::memory_order_relaxed); |
||||
if (writer == nullptr) { |
||||
backlog_.send_initial_metadata_wanted = true; |
||||
return; |
||||
} |
||||
} |
||||
writer->SendInitialMetadata(); |
||||
} |
||||
void StartWrite(const Response* resp) { |
||||
StartWrite(resp, ::grpc::WriteOptions()); |
||||
} |
||||
void StartWrite(const Response* resp, ::grpc::WriteOptions options) { |
||||
ServerCallbackWriter<Response>* writer = |
||||
writer_.load(std::memory_order_acquire); |
||||
if (writer == nullptr) { |
||||
grpc::internal::MutexLock l(&writer_mu_); |
||||
writer = writer_.load(std::memory_order_relaxed); |
||||
if (writer == nullptr) { |
||||
backlog_.write_wanted = resp; |
||||
backlog_.write_options_wanted = std::move(options); |
||||
return; |
||||
} |
||||
} |
||||
writer->Write(resp, std::move(options)); |
||||
} |
||||
void StartWriteAndFinish(const Response* resp, ::grpc::WriteOptions options, |
||||
::grpc::Status s) { |
||||
ServerCallbackWriter<Response>* writer = |
||||
writer_.load(std::memory_order_acquire); |
||||
if (writer == nullptr) { |
||||
grpc::internal::MutexLock l(&writer_mu_); |
||||
writer = writer_.load(std::memory_order_relaxed); |
||||
if (writer == nullptr) { |
||||
backlog_.write_and_finish_wanted = true; |
||||
backlog_.write_wanted = resp; |
||||
backlog_.write_options_wanted = std::move(options); |
||||
backlog_.status_wanted = std::move(s); |
||||
return; |
||||
} |
||||
} |
||||
writer->WriteAndFinish(resp, std::move(options), std::move(s)); |
||||
} |
||||
void StartWriteLast(const Response* resp, ::grpc::WriteOptions options) { |
||||
StartWrite(resp, std::move(options.set_last_message())); |
||||
} |
||||
void Finish(::grpc::Status s) { |
||||
ServerCallbackWriter<Response>* writer = |
||||
writer_.load(std::memory_order_acquire); |
||||
if (writer == nullptr) { |
||||
grpc::internal::MutexLock l(&writer_mu_); |
||||
writer = writer_.load(std::memory_order_relaxed); |
||||
if (writer == nullptr) { |
||||
backlog_.finish_wanted = true; |
||||
backlog_.status_wanted = std::move(s); |
||||
return; |
||||
} |
||||
} |
||||
writer->Finish(std::move(s)); |
||||
} |
||||
|
||||
/// The following notifications are exactly like ServerBidiReactor.
|
||||
virtual void OnSendInitialMetadataDone(bool /*ok*/) {} |
||||
virtual void OnWriteDone(bool /*ok*/) {} |
||||
void OnDone() override = 0; |
||||
void OnCancel() override {} |
||||
|
||||
private: |
||||
friend class ServerCallbackWriter<Response>; |
||||
// May be overridden by internal implementation details. This is not a public
|
||||
// customization point.
|
||||
virtual void InternalBindWriter(ServerCallbackWriter<Response>* writer) { |
||||
grpc::internal::ReleasableMutexLock l(&writer_mu_); |
||||
PreBindBacklog ops(std::move(backlog_)); |
||||
writer_.store(writer, std::memory_order_release); |
||||
l.Unlock(); |
||||
|
||||
if (ops.send_initial_metadata_wanted) { |
||||
writer->SendInitialMetadata(); |
||||
} |
||||
if (ops.write_and_finish_wanted) { |
||||
writer->WriteAndFinish(ops.write_wanted, |
||||
std::move(ops.write_options_wanted), |
||||
std::move(ops.status_wanted)); |
||||
} else { |
||||
if (ops.write_wanted != nullptr) { |
||||
writer->Write(ops.write_wanted, std::move(ops.write_options_wanted)); |
||||
} |
||||
if (ops.finish_wanted) { |
||||
writer->Finish(std::move(ops.status_wanted)); |
||||
} |
||||
} |
||||
} |
||||
|
||||
grpc::internal::Mutex writer_mu_; |
||||
std::atomic<ServerCallbackWriter<Response>*> writer_{nullptr}; |
||||
struct PreBindBacklog { |
||||
bool send_initial_metadata_wanted = false; |
||||
bool write_and_finish_wanted = false; |
||||
bool finish_wanted = false; |
||||
const Response* write_wanted = nullptr; |
||||
::grpc::WriteOptions write_options_wanted; |
||||
::grpc::Status status_wanted; |
||||
}; |
||||
PreBindBacklog backlog_ /* GUARDED_BY(writer_mu_) */; |
||||
}; |
||||
|
||||
class ServerUnaryReactor : public internal::ServerReactor { |
||||
public: |
||||
ServerUnaryReactor() : call_(nullptr) {} |
||||
~ServerUnaryReactor() = default; |
||||
|
||||
/// StartSendInitialMetadata is exactly like ServerBidiReactor.
|
||||
void StartSendInitialMetadata() { |
||||
ServerCallbackUnary* call = call_.load(std::memory_order_acquire); |
||||
if (call == nullptr) { |
||||
grpc::internal::MutexLock l(&call_mu_); |
||||
call = call_.load(std::memory_order_relaxed); |
||||
if (call == nullptr) { |
||||
backlog_.send_initial_metadata_wanted = true; |
||||
return; |
||||
} |
||||
} |
||||
call->SendInitialMetadata(); |
||||
} |
||||
/// Finish is similar to ServerBidiReactor except for one detail.
|
||||
/// If the status is non-OK, any message will not be sent. Instead,
|
||||
/// the client will only receive the status and any trailing metadata.
|
||||
void Finish(::grpc::Status s) { |
||||
ServerCallbackUnary* call = call_.load(std::memory_order_acquire); |
||||
if (call == nullptr) { |
||||
grpc::internal::MutexLock l(&call_mu_); |
||||
call = call_.load(std::memory_order_relaxed); |
||||
if (call == nullptr) { |
||||
backlog_.finish_wanted = true; |
||||
backlog_.status_wanted = std::move(s); |
||||
return; |
||||
} |
||||
} |
||||
call->Finish(std::move(s)); |
||||
} |
||||
|
||||
/// The following notifications are exactly like ServerBidiReactor.
|
||||
virtual void OnSendInitialMetadataDone(bool /*ok*/) {} |
||||
void OnDone() override = 0; |
||||
void OnCancel() override {} |
||||
|
||||
private: |
||||
friend class ServerCallbackUnary; |
||||
// May be overridden by internal implementation details. This is not a public
|
||||
// customization point.
|
||||
virtual void InternalBindCall(ServerCallbackUnary* call) { |
||||
grpc::internal::ReleasableMutexLock l(&call_mu_); |
||||
PreBindBacklog ops(std::move(backlog_)); |
||||
call_.store(call, std::memory_order_release); |
||||
l.Unlock(); |
||||
|
||||
if (ops.send_initial_metadata_wanted) { |
||||
call->SendInitialMetadata(); |
||||
} |
||||
if (ops.finish_wanted) { |
||||
call->Finish(std::move(ops.status_wanted)); |
||||
} |
||||
} |
||||
|
||||
grpc::internal::Mutex call_mu_; |
||||
std::atomic<ServerCallbackUnary*> call_{nullptr}; |
||||
struct PreBindBacklog { |
||||
bool send_initial_metadata_wanted = false; |
||||
bool finish_wanted = false; |
||||
::grpc::Status status_wanted; |
||||
}; |
||||
PreBindBacklog backlog_ /* GUARDED_BY(call_mu_) */; |
||||
}; |
||||
|
||||
namespace internal { |
||||
|
||||
template <class Base> |
||||
class FinishOnlyReactor : public Base { |
||||
public: |
||||
explicit FinishOnlyReactor(::grpc::Status s) { this->Finish(std::move(s)); } |
||||
void OnDone() override { this->~FinishOnlyReactor(); } |
||||
}; |
||||
|
||||
using UnimplementedUnaryReactor = FinishOnlyReactor<ServerUnaryReactor>; |
||||
template <class Request> |
||||
using UnimplementedReadReactor = FinishOnlyReactor<ServerReadReactor<Request>>; |
||||
template <class Response> |
||||
using UnimplementedWriteReactor = |
||||
FinishOnlyReactor<ServerWriteReactor<Response>>; |
||||
template <class Request, class Response> |
||||
using UnimplementedBidiReactor = |
||||
FinishOnlyReactor<ServerBidiReactor<Request, Response>>; |
||||
|
||||
} // namespace internal
|
||||
} // namespace grpc_impl
|
||||
|
||||
#endif // GRPCPP_IMPL_CODEGEN_SERVER_CALLBACK_IMPL_H
|
||||
@ -1,943 +0,0 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2019 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 GRPCPP_IMPL_CODEGEN_SYNC_STREAM_IMPL_H |
||||
#define GRPCPP_IMPL_CODEGEN_SYNC_STREAM_IMPL_H |
||||
|
||||
#include <grpcpp/impl/codegen/call.h> |
||||
#include <grpcpp/impl/codegen/channel_interface.h> |
||||
#include <grpcpp/impl/codegen/client_context.h> |
||||
#include <grpcpp/impl/codegen/completion_queue.h> |
||||
#include <grpcpp/impl/codegen/core_codegen_interface.h> |
||||
#include <grpcpp/impl/codegen/server_context.h> |
||||
#include <grpcpp/impl/codegen/service_type.h> |
||||
#include <grpcpp/impl/codegen/status.h> |
||||
|
||||
namespace grpc_impl { |
||||
|
||||
namespace internal { |
||||
/// Common interface for all synchronous client side streaming.
|
||||
class ClientStreamingInterface { |
||||
public: |
||||
virtual ~ClientStreamingInterface() {} |
||||
|
||||
/// Block waiting until the stream finishes and a final status of the call is
|
||||
/// available.
|
||||
///
|
||||
/// It is appropriate to call this method exactly once when both:
|
||||
/// * the calling code (client-side) has no more message to send
|
||||
/// (this can be declared implicitly by calling this method, or
|
||||
/// explicitly through an earlier call to <i>WritesDone</i> method of the
|
||||
/// class in use, e.g. \a ClientWriterInterface::WritesDone or
|
||||
/// \a ClientReaderWriterInterface::WritesDone).
|
||||
/// * there are no more messages to be received from the server (which can
|
||||
/// be known implicitly, or explicitly from an earlier call to \a
|
||||
/// ReaderInterface::Read that returned "false").
|
||||
///
|
||||
/// This function will return either:
|
||||
/// - when all incoming messages have been read and the server has
|
||||
/// returned status.
|
||||
/// - when the server has returned a non-OK status.
|
||||
/// - OR when the call failed for some reason and the library generated a
|
||||
/// status.
|
||||
///
|
||||
/// Return values:
|
||||
/// - \a Status contains the status code, message and details for the call
|
||||
/// - the \a ClientContext associated with this call is updated with
|
||||
/// possible trailing metadata sent from the server.
|
||||
virtual ::grpc::Status Finish() = 0; |
||||
}; |
||||
|
||||
/// Common interface for all synchronous server side streaming.
|
||||
class ServerStreamingInterface { |
||||
public: |
||||
virtual ~ServerStreamingInterface() {} |
||||
|
||||
/// Block to send initial metadata to client.
|
||||
/// This call is optional, but if it is used, it cannot be used concurrently
|
||||
/// with or after the \a Finish method.
|
||||
///
|
||||
/// The initial metadata that will be sent to the client will be
|
||||
/// taken from the \a ServerContext associated with the call.
|
||||
virtual void SendInitialMetadata() = 0; |
||||
}; |
||||
|
||||
/// An interface that yields a sequence of messages of type \a R.
|
||||
template <class R> |
||||
class ReaderInterface { |
||||
public: |
||||
virtual ~ReaderInterface() {} |
||||
|
||||
/// Get an upper bound on the next message size available for reading on this
|
||||
/// stream.
|
||||
virtual bool NextMessageSize(uint32_t* sz) = 0; |
||||
|
||||
/// Block to read a message and parse to \a msg. Returns \a true on success.
|
||||
/// This is thread-safe with respect to \a Write or \WritesDone methods on
|
||||
/// the same stream. It should not be called concurrently with another \a
|
||||
/// Read on the same stream as the order of delivery will not be defined.
|
||||
///
|
||||
/// \param[out] msg The read message.
|
||||
///
|
||||
/// \return \a false when there will be no more incoming messages, either
|
||||
/// because the other side has called \a WritesDone() or the stream has failed
|
||||
/// (or been cancelled).
|
||||
virtual bool Read(R* msg) = 0; |
||||
}; |
||||
|
||||
/// An interface that can be fed a sequence of messages of type \a W.
|
||||
template <class W> |
||||
class WriterInterface { |
||||
public: |
||||
virtual ~WriterInterface() {} |
||||
|
||||
/// Block to write \a msg to the stream with WriteOptions \a options.
|
||||
/// This is thread-safe with respect to \a ReaderInterface::Read
|
||||
///
|
||||
/// \param msg The message to be written to the stream.
|
||||
/// \param options The WriteOptions affecting the write operation.
|
||||
///
|
||||
/// \return \a true on success, \a false when the stream has been closed.
|
||||
virtual bool Write(const W& msg, ::grpc::WriteOptions options) = 0; |
||||
|
||||
/// Block to write \a msg to the stream with default write options.
|
||||
/// This is thread-safe with respect to \a ReaderInterface::Read
|
||||
///
|
||||
/// \param msg The message to be written to the stream.
|
||||
///
|
||||
/// \return \a true on success, \a false when the stream has been closed.
|
||||
inline bool Write(const W& msg) { return Write(msg, ::grpc::WriteOptions()); } |
||||
|
||||
/// Write \a msg and coalesce it with the writing of trailing metadata, using
|
||||
/// WriteOptions \a options.
|
||||
///
|
||||
/// For client, WriteLast is equivalent of performing Write and WritesDone in
|
||||
/// a single step. \a msg and trailing metadata are coalesced and sent on wire
|
||||
/// by calling this function. For server, WriteLast buffers the \a msg.
|
||||
/// The writing of \a msg is held until the service handler returns,
|
||||
/// where \a msg and trailing metadata are coalesced and sent on wire.
|
||||
/// 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 to the stream.
|
||||
/// \param[in] options The WriteOptions to be used to write this message.
|
||||
void WriteLast(const W& msg, ::grpc::WriteOptions options) { |
||||
Write(msg, options.set_last_message()); |
||||
} |
||||
}; |
||||
|
||||
} // namespace internal
|
||||
|
||||
/// Client-side interface for streaming reads of message of type \a R.
|
||||
template <class R> |
||||
class ClientReaderInterface : public internal::ClientStreamingInterface, |
||||
public internal::ReaderInterface<R> { |
||||
public: |
||||
/// Block to wait for initial metadata from server. The received metadata
|
||||
/// can only be accessed after this call returns. Should only be called before
|
||||
/// the first read. Calling this method is optional, and if it is not called
|
||||
/// the metadata will be available in ClientContext after the first read.
|
||||
virtual void WaitForInitialMetadata() = 0; |
||||
}; |
||||
|
||||
namespace internal { |
||||
template <class R> |
||||
class ClientReaderFactory { |
||||
public: |
||||
template <class W> |
||||
static ClientReader<R>* Create(::grpc::ChannelInterface* channel, |
||||
const ::grpc::internal::RpcMethod& method, |
||||
::grpc::ClientContext* context, |
||||
const W& request) { |
||||
return new ClientReader<R>(channel, method, context, request); |
||||
} |
||||
}; |
||||
} // namespace internal
|
||||
|
||||
/// Synchronous (blocking) client-side API for doing server-streaming RPCs,
|
||||
/// where the stream of messages coming from the server has messages
|
||||
/// of type \a R.
|
||||
template <class R> |
||||
class ClientReader final : public ClientReaderInterface<R> { |
||||
public: |
||||
/// See the \a ClientStreamingInterface.WaitForInitialMetadata method for
|
||||
/// semantics.
|
||||
///
|
||||
// Side effect:
|
||||
/// Once complete, the initial metadata read from
|
||||
/// the server will be accessible through the \a ClientContext used to
|
||||
/// construct this object.
|
||||
void WaitForInitialMetadata() override { |
||||
GPR_CODEGEN_ASSERT(!context_->initial_metadata_received_); |
||||
|
||||
::grpc::internal::CallOpSet<::grpc::internal::CallOpRecvInitialMetadata> |
||||
ops; |
||||
ops.RecvInitialMetadata(context_); |
||||
call_.PerformOps(&ops); |
||||
cq_.Pluck(&ops); /// status ignored
|
||||
} |
||||
|
||||
bool NextMessageSize(uint32_t* sz) override { |
||||
int result = call_.max_receive_message_size(); |
||||
*sz = (result > 0) ? result : UINT32_MAX; |
||||
return true; |
||||
} |
||||
|
||||
/// See the \a ReaderInterface.Read method for semantics.
|
||||
/// Side effect:
|
||||
/// This also receives initial metadata from the server, if not
|
||||
/// already received (if initial metadata is received, it can be then
|
||||
/// accessed through the \a ClientContext associated with this call).
|
||||
bool Read(R* msg) override { |
||||
::grpc::internal::CallOpSet<::grpc::internal::CallOpRecvInitialMetadata, |
||||
::grpc::internal::CallOpRecvMessage<R>> |
||||
ops; |
||||
if (!context_->initial_metadata_received_) { |
||||
ops.RecvInitialMetadata(context_); |
||||
} |
||||
ops.RecvMessage(msg); |
||||
call_.PerformOps(&ops); |
||||
return cq_.Pluck(&ops) && ops.got_message; |
||||
} |
||||
|
||||
/// See the \a ClientStreamingInterface.Finish method for semantics.
|
||||
///
|
||||
/// Side effect:
|
||||
/// The \a ClientContext associated with this call is updated with
|
||||
/// possible metadata received from the server.
|
||||
::grpc::Status Finish() override { |
||||
::grpc::internal::CallOpSet<::grpc::internal::CallOpClientRecvStatus> ops; |
||||
::grpc::Status status; |
||||
ops.ClientRecvStatus(context_, &status); |
||||
call_.PerformOps(&ops); |
||||
GPR_CODEGEN_ASSERT(cq_.Pluck(&ops)); |
||||
return status; |
||||
} |
||||
|
||||
private: |
||||
friend class internal::ClientReaderFactory<R>; |
||||
::grpc::ClientContext* context_; |
||||
::grpc::CompletionQueue cq_; |
||||
::grpc::internal::Call call_; |
||||
|
||||
/// Block to create a stream and write the initial metadata and \a request
|
||||
/// 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> |
||||
ClientReader(::grpc::ChannelInterface* channel, |
||||
const ::grpc::internal::RpcMethod& method, |
||||
::grpc::ClientContext* context, const W& request) |
||||
: context_(context), |
||||
cq_(grpc_completion_queue_attributes{ |
||||
GRPC_CQ_CURRENT_VERSION, GRPC_CQ_PLUCK, GRPC_CQ_DEFAULT_POLLING, |
||||
nullptr}), // Pluckable cq
|
||||
call_(channel->CreateCall(method, context, &cq_)) { |
||||
::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata, |
||||
::grpc::internal::CallOpSendMessage, |
||||
::grpc::internal::CallOpClientSendClose> |
||||
ops; |
||||
ops.SendInitialMetadata(&context->send_initial_metadata_, |
||||
context->initial_metadata_flags()); |
||||
// TODO(ctiller): don't assert
|
||||
GPR_CODEGEN_ASSERT(ops.SendMessagePtr(&request).ok()); |
||||
ops.ClientSendClose(); |
||||
call_.PerformOps(&ops); |
||||
cq_.Pluck(&ops); |
||||
} |
||||
}; |
||||
|
||||
/// Client-side interface for streaming writes of message type \a W.
|
||||
template <class W> |
||||
class ClientWriterInterface : public internal::ClientStreamingInterface, |
||||
public internal::WriterInterface<W> { |
||||
public: |
||||
/// Half close writing from the client. (signal that the stream of messages
|
||||
/// coming from the client is complete).
|
||||
/// Blocks until currently-pending writes are completed.
|
||||
/// Thread safe with respect to \a ReaderInterface::Read operations only
|
||||
///
|
||||
/// \return Whether the writes were successful.
|
||||
virtual bool WritesDone() = 0; |
||||
}; |
||||
|
||||
namespace internal { |
||||
template <class W> |
||||
class ClientWriterFactory { |
||||
public: |
||||
template <class R> |
||||
static ClientWriter<W>* Create(::grpc::ChannelInterface* channel, |
||||
const ::grpc::internal::RpcMethod& method, |
||||
::grpc::ClientContext* context, R* response) { |
||||
return new ClientWriter<W>(channel, method, context, response); |
||||
} |
||||
}; |
||||
} // namespace internal
|
||||
|
||||
/// Synchronous (blocking) client-side API for doing client-streaming RPCs,
|
||||
/// where the outgoing message stream coming from the client has messages of
|
||||
/// type \a W.
|
||||
template <class W> |
||||
class ClientWriter : public ClientWriterInterface<W> { |
||||
public: |
||||
/// See the \a ClientStreamingInterface.WaitForInitialMetadata method for
|
||||
/// semantics.
|
||||
///
|
||||
// Side effect:
|
||||
/// Once complete, the initial metadata read from the server will be
|
||||
/// accessible through the \a ClientContext used to construct this object.
|
||||
void WaitForInitialMetadata() { |
||||
GPR_CODEGEN_ASSERT(!context_->initial_metadata_received_); |
||||
|
||||
::grpc::internal::CallOpSet<::grpc::internal::CallOpRecvInitialMetadata> |
||||
ops; |
||||
ops.RecvInitialMetadata(context_); |
||||
call_.PerformOps(&ops); |
||||
cq_.Pluck(&ops); // status ignored
|
||||
} |
||||
|
||||
/// See the WriterInterface.Write(const W& msg, WriteOptions options) method
|
||||
/// for semantics.
|
||||
///
|
||||
/// Side effect:
|
||||
/// Also sends initial metadata if not already sent (using the
|
||||
/// \a ClientContext associated with this call).
|
||||
using internal::WriterInterface<W>::Write; |
||||
bool Write(const W& msg, ::grpc::WriteOptions options) override { |
||||
::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata, |
||||
::grpc::internal::CallOpSendMessage, |
||||
::grpc::internal::CallOpClientSendClose> |
||||
ops; |
||||
|
||||
if (options.is_last_message()) { |
||||
options.set_buffer_hint(); |
||||
ops.ClientSendClose(); |
||||
} |
||||
if (context_->initial_metadata_corked_) { |
||||
ops.SendInitialMetadata(&context_->send_initial_metadata_, |
||||
context_->initial_metadata_flags()); |
||||
context_->set_initial_metadata_corked(false); |
||||
} |
||||
if (!ops.SendMessagePtr(&msg, options).ok()) { |
||||
return false; |
||||
} |
||||
|
||||
call_.PerformOps(&ops); |
||||
return cq_.Pluck(&ops); |
||||
} |
||||
|
||||
bool WritesDone() override { |
||||
::grpc::internal::CallOpSet<::grpc::internal::CallOpClientSendClose> ops; |
||||
ops.ClientSendClose(); |
||||
call_.PerformOps(&ops); |
||||
return cq_.Pluck(&ops); |
||||
} |
||||
|
||||
/// See the ClientStreamingInterface.Finish method for semantics.
|
||||
/// Side effects:
|
||||
/// - Also receives initial metadata if not already received.
|
||||
/// - Attempts to fill in the \a response parameter passed
|
||||
/// to the constructor of this instance with the response
|
||||
/// message from the server.
|
||||
::grpc::Status Finish() override { |
||||
::grpc::Status status; |
||||
if (!context_->initial_metadata_received_) { |
||||
finish_ops_.RecvInitialMetadata(context_); |
||||
} |
||||
finish_ops_.ClientRecvStatus(context_, &status); |
||||
call_.PerformOps(&finish_ops_); |
||||
GPR_CODEGEN_ASSERT(cq_.Pluck(&finish_ops_)); |
||||
return status; |
||||
} |
||||
|
||||
private: |
||||
friend class internal::ClientWriterFactory<W>; |
||||
|
||||
/// Block to create a stream (i.e. send request headers and other initial
|
||||
/// metadata to the server). Note that \a context will be used to fill
|
||||
/// in custom initial metadata. \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> |
||||
ClientWriter(::grpc::ChannelInterface* channel, |
||||
const ::grpc::internal::RpcMethod& method, |
||||
::grpc::ClientContext* context, R* response) |
||||
: context_(context), |
||||
cq_(grpc_completion_queue_attributes{ |
||||
GRPC_CQ_CURRENT_VERSION, GRPC_CQ_PLUCK, GRPC_CQ_DEFAULT_POLLING, |
||||
nullptr}), // Pluckable cq
|
||||
call_(channel->CreateCall(method, context, &cq_)) { |
||||
finish_ops_.RecvMessage(response); |
||||
finish_ops_.AllowNoMessage(); |
||||
|
||||
if (!context_->initial_metadata_corked_) { |
||||
::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata> |
||||
ops; |
||||
ops.SendInitialMetadata(&context->send_initial_metadata_, |
||||
context->initial_metadata_flags()); |
||||
call_.PerformOps(&ops); |
||||
cq_.Pluck(&ops); |
||||
} |
||||
} |
||||
|
||||
::grpc::ClientContext* context_; |
||||
::grpc::internal::CallOpSet<::grpc::internal::CallOpRecvInitialMetadata, |
||||
::grpc::internal::CallOpGenericRecvMessage, |
||||
::grpc::internal::CallOpClientRecvStatus> |
||||
finish_ops_; |
||||
::grpc::CompletionQueue cq_; |
||||
::grpc::internal::Call call_; |
||||
}; |
||||
|
||||
/// Client-side interface for bi-directional streaming with
|
||||
/// client-to-server stream messages of type \a W and
|
||||
/// server-to-client stream messages of type \a R.
|
||||
template <class W, class R> |
||||
class ClientReaderWriterInterface : public internal::ClientStreamingInterface, |
||||
public internal::WriterInterface<W>, |
||||
public internal::ReaderInterface<R> { |
||||
public: |
||||
/// Block to wait for initial metadata from server. The received metadata
|
||||
/// can only be accessed after this call returns. Should only be called before
|
||||
/// the first read. Calling this method is optional, and if it is not called
|
||||
/// the metadata will be available in ClientContext after the first read.
|
||||
virtual void WaitForInitialMetadata() = 0; |
||||
|
||||
/// Half close writing from the client. (signal that the stream of messages
|
||||
/// coming from the client is complete).
|
||||
/// Blocks until currently-pending writes are completed.
|
||||
/// Thread-safe with respect to \a ReaderInterface::Read
|
||||
///
|
||||
/// \return Whether the writes were successful.
|
||||
virtual bool WritesDone() = 0; |
||||
}; |
||||
|
||||
namespace internal { |
||||
template <class W, class R> |
||||
class ClientReaderWriterFactory { |
||||
public: |
||||
static ClientReaderWriter<W, R>* Create( |
||||
::grpc::ChannelInterface* channel, |
||||
const ::grpc::internal::RpcMethod& method, |
||||
::grpc::ClientContext* context) { |
||||
return new ClientReaderWriter<W, R>(channel, method, context); |
||||
} |
||||
}; |
||||
} // namespace internal
|
||||
|
||||
/// Synchronous (blocking) client-side API for bi-directional streaming RPCs,
|
||||
/// where the outgoing message stream coming from the client has messages of
|
||||
/// type \a W, and the incoming messages stream coming from the server has
|
||||
/// messages of type \a R.
|
||||
template <class W, class R> |
||||
class ClientReaderWriter final : public ClientReaderWriterInterface<W, R> { |
||||
public: |
||||
/// Block waiting to read initial metadata from the server.
|
||||
/// This call is optional, but if it is used, it cannot be used concurrently
|
||||
/// with or after the \a Finish method.
|
||||
///
|
||||
/// Once complete, the initial metadata read from the server will be
|
||||
/// accessible through the \a ClientContext used to construct this object.
|
||||
void WaitForInitialMetadata() override { |
||||
GPR_CODEGEN_ASSERT(!context_->initial_metadata_received_); |
||||
|
||||
::grpc::internal::CallOpSet<::grpc::internal::CallOpRecvInitialMetadata> |
||||
ops; |
||||
ops.RecvInitialMetadata(context_); |
||||
call_.PerformOps(&ops); |
||||
cq_.Pluck(&ops); // status ignored
|
||||
} |
||||
|
||||
bool NextMessageSize(uint32_t* sz) override { |
||||
int result = call_.max_receive_message_size(); |
||||
*sz = (result > 0) ? result : UINT32_MAX; |
||||
return true; |
||||
} |
||||
|
||||
/// See the \a ReaderInterface.Read method for semantics.
|
||||
/// Side effect:
|
||||
/// Also receives initial metadata if not already received (updates the \a
|
||||
/// ClientContext associated with this call in that case).
|
||||
bool Read(R* msg) override { |
||||
::grpc::internal::CallOpSet<::grpc::internal::CallOpRecvInitialMetadata, |
||||
::grpc::internal::CallOpRecvMessage<R>> |
||||
ops; |
||||
if (!context_->initial_metadata_received_) { |
||||
ops.RecvInitialMetadata(context_); |
||||
} |
||||
ops.RecvMessage(msg); |
||||
call_.PerformOps(&ops); |
||||
return cq_.Pluck(&ops) && ops.got_message; |
||||
} |
||||
|
||||
/// See the \a WriterInterface.Write method for semantics.
|
||||
///
|
||||
/// Side effect:
|
||||
/// Also sends initial metadata if not already sent (using the
|
||||
/// \a ClientContext associated with this call to fill in values).
|
||||
using internal::WriterInterface<W>::Write; |
||||
bool Write(const W& msg, ::grpc::WriteOptions options) override { |
||||
::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata, |
||||
::grpc::internal::CallOpSendMessage, |
||||
::grpc::internal::CallOpClientSendClose> |
||||
ops; |
||||
|
||||
if (options.is_last_message()) { |
||||
options.set_buffer_hint(); |
||||
ops.ClientSendClose(); |
||||
} |
||||
if (context_->initial_metadata_corked_) { |
||||
ops.SendInitialMetadata(&context_->send_initial_metadata_, |
||||
context_->initial_metadata_flags()); |
||||
context_->set_initial_metadata_corked(false); |
||||
} |
||||
if (!ops.SendMessagePtr(&msg, options).ok()) { |
||||
return false; |
||||
} |
||||
|
||||
call_.PerformOps(&ops); |
||||
return cq_.Pluck(&ops); |
||||
} |
||||
|
||||
bool WritesDone() override { |
||||
::grpc::internal::CallOpSet<::grpc::internal::CallOpClientSendClose> ops; |
||||
ops.ClientSendClose(); |
||||
call_.PerformOps(&ops); |
||||
return cq_.Pluck(&ops); |
||||
} |
||||
|
||||
/// See the ClientStreamingInterface.Finish method for semantics.
|
||||
///
|
||||
/// Side effect:
|
||||
/// - the \a ClientContext associated with this call is updated with
|
||||
/// possible trailing metadata sent from the server.
|
||||
::grpc::Status Finish() override { |
||||
::grpc::internal::CallOpSet<::grpc::internal::CallOpRecvInitialMetadata, |
||||
::grpc::internal::CallOpClientRecvStatus> |
||||
ops; |
||||
if (!context_->initial_metadata_received_) { |
||||
ops.RecvInitialMetadata(context_); |
||||
} |
||||
::grpc::Status status; |
||||
ops.ClientRecvStatus(context_, &status); |
||||
call_.PerformOps(&ops); |
||||
GPR_CODEGEN_ASSERT(cq_.Pluck(&ops)); |
||||
return status; |
||||
} |
||||
|
||||
private: |
||||
friend class internal::ClientReaderWriterFactory<W, R>; |
||||
|
||||
::grpc::ClientContext* context_; |
||||
::grpc::CompletionQueue cq_; |
||||
::grpc::internal::Call call_; |
||||
|
||||
/// Block to create a stream and write the initial metadata and \a request
|
||||
/// out. Note that \a context will be used to fill in custom initial metadata
|
||||
/// used to send to the server when starting the call.
|
||||
ClientReaderWriter(::grpc::ChannelInterface* channel, |
||||
const ::grpc::internal::RpcMethod& method, |
||||
::grpc::ClientContext* context) |
||||
: context_(context), |
||||
cq_(grpc_completion_queue_attributes{ |
||||
GRPC_CQ_CURRENT_VERSION, GRPC_CQ_PLUCK, GRPC_CQ_DEFAULT_POLLING, |
||||
nullptr}), // Pluckable cq
|
||||
call_(channel->CreateCall(method, context, &cq_)) { |
||||
if (!context_->initial_metadata_corked_) { |
||||
::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata> |
||||
ops; |
||||
ops.SendInitialMetadata(&context->send_initial_metadata_, |
||||
context->initial_metadata_flags()); |
||||
call_.PerformOps(&ops); |
||||
cq_.Pluck(&ops); |
||||
} |
||||
} |
||||
}; |
||||
|
||||
/// Server-side interface for streaming reads of message of type \a R.
|
||||
template <class R> |
||||
class ServerReaderInterface : public internal::ServerStreamingInterface, |
||||
public internal::ReaderInterface<R> {}; |
||||
|
||||
/// Synchronous (blocking) server-side API for doing client-streaming RPCs,
|
||||
/// where the incoming message stream coming from the client has messages of
|
||||
/// type \a R.
|
||||
template <class R> |
||||
class ServerReader final : public ServerReaderInterface<R> { |
||||
public: |
||||
/// See the \a ServerStreamingInterface.SendInitialMetadata method
|
||||
/// for semantics. Note that initial metadata will be affected by the
|
||||
/// \a ServerContext associated with this call.
|
||||
void SendInitialMetadata() override { |
||||
GPR_CODEGEN_ASSERT(!ctx_->sent_initial_metadata_); |
||||
|
||||
::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata> |
||||
ops; |
||||
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_->PerformOps(&ops); |
||||
call_->cq()->Pluck(&ops); |
||||
} |
||||
|
||||
bool NextMessageSize(uint32_t* sz) override { |
||||
int result = call_->max_receive_message_size(); |
||||
*sz = (result > 0) ? result : UINT32_MAX; |
||||
return true; |
||||
} |
||||
|
||||
bool Read(R* msg) override { |
||||
::grpc::internal::CallOpSet<::grpc::internal::CallOpRecvMessage<R>> ops; |
||||
ops.RecvMessage(msg); |
||||
call_->PerformOps(&ops); |
||||
return call_->cq()->Pluck(&ops) && ops.got_message; |
||||
} |
||||
|
||||
private: |
||||
::grpc::internal::Call* const call_; |
||||
::grpc::ServerContext* const ctx_; |
||||
|
||||
template <class ServiceType, class RequestType, class ResponseType> |
||||
friend class ::grpc_impl::internal::ClientStreamingHandler; |
||||
|
||||
ServerReader(::grpc::internal::Call* call, ::grpc::ServerContext* ctx) |
||||
: call_(call), ctx_(ctx) {} |
||||
}; |
||||
|
||||
/// Server-side interface for streaming writes of message of type \a W.
|
||||
template <class W> |
||||
class ServerWriterInterface : public internal::ServerStreamingInterface, |
||||
public internal::WriterInterface<W> {}; |
||||
|
||||
/// Synchronous (blocking) server-side API for doing for doing a
|
||||
/// server-streaming RPCs, where the outgoing message stream coming from the
|
||||
/// server has messages of type \a W.
|
||||
template <class W> |
||||
class ServerWriter final : public ServerWriterInterface<W> { |
||||
public: |
||||
/// See the \a ServerStreamingInterface.SendInitialMetadata method
|
||||
/// for semantics.
|
||||
/// Note that initial metadata will be affected by the
|
||||
/// \a ServerContext associated with this call.
|
||||
void SendInitialMetadata() override { |
||||
GPR_CODEGEN_ASSERT(!ctx_->sent_initial_metadata_); |
||||
|
||||
::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata> |
||||
ops; |
||||
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_->PerformOps(&ops); |
||||
call_->cq()->Pluck(&ops); |
||||
} |
||||
|
||||
/// See the \a WriterInterface.Write method for semantics.
|
||||
///
|
||||
/// Side effect:
|
||||
/// Also sends initial metadata if not already sent (using the
|
||||
/// \a ClientContext associated with this call to fill in values).
|
||||
using internal::WriterInterface<W>::Write; |
||||
bool Write(const W& msg, ::grpc::WriteOptions options) override { |
||||
if (options.is_last_message()) { |
||||
options.set_buffer_hint(); |
||||
} |
||||
|
||||
if (!ctx_->pending_ops_.SendMessagePtr(&msg, options).ok()) { |
||||
return false; |
||||
} |
||||
if (!ctx_->sent_initial_metadata_) { |
||||
ctx_->pending_ops_.SendInitialMetadata(&ctx_->initial_metadata_, |
||||
ctx_->initial_metadata_flags()); |
||||
if (ctx_->compression_level_set()) { |
||||
ctx_->pending_ops_.set_compression_level(ctx_->compression_level()); |
||||
} |
||||
ctx_->sent_initial_metadata_ = true; |
||||
} |
||||
call_->PerformOps(&ctx_->pending_ops_); |
||||
// if this is the last message we defer the pluck until AFTER we start
|
||||
// the trailing md op. This prevents hangs. See
|
||||
// https://github.com/grpc/grpc/issues/11546
|
||||
if (options.is_last_message()) { |
||||
ctx_->has_pending_ops_ = true; |
||||
return true; |
||||
} |
||||
ctx_->has_pending_ops_ = false; |
||||
return call_->cq()->Pluck(&ctx_->pending_ops_); |
||||
} |
||||
|
||||
private: |
||||
::grpc::internal::Call* const call_; |
||||
::grpc::ServerContext* const ctx_; |
||||
|
||||
template <class ServiceType, class RequestType, class ResponseType> |
||||
friend class ::grpc_impl::internal::ServerStreamingHandler; |
||||
|
||||
ServerWriter(::grpc::internal::Call* call, ::grpc::ServerContext* ctx) |
||||
: call_(call), ctx_(ctx) {} |
||||
}; |
||||
|
||||
/// Server-side interface for bi-directional streaming.
|
||||
template <class W, class R> |
||||
class ServerReaderWriterInterface : public internal::ServerStreamingInterface, |
||||
public internal::WriterInterface<W>, |
||||
public internal::ReaderInterface<R> {}; |
||||
|
||||
/// Actual implementation of bi-directional streaming
|
||||
namespace internal { |
||||
template <class W, class R> |
||||
class ServerReaderWriterBody final { |
||||
public: |
||||
ServerReaderWriterBody(grpc::internal::Call* call, ::grpc::ServerContext* ctx) |
||||
: call_(call), ctx_(ctx) {} |
||||
|
||||
void SendInitialMetadata() { |
||||
GPR_CODEGEN_ASSERT(!ctx_->sent_initial_metadata_); |
||||
|
||||
grpc::internal::CallOpSet<grpc::internal::CallOpSendInitialMetadata> ops; |
||||
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_->PerformOps(&ops); |
||||
call_->cq()->Pluck(&ops); |
||||
} |
||||
|
||||
bool NextMessageSize(uint32_t* sz) { |
||||
int result = call_->max_receive_message_size(); |
||||
*sz = (result > 0) ? result : UINT32_MAX; |
||||
return true; |
||||
} |
||||
|
||||
bool Read(R* msg) { |
||||
::grpc::internal::CallOpSet<::grpc::internal::CallOpRecvMessage<R>> ops; |
||||
ops.RecvMessage(msg); |
||||
call_->PerformOps(&ops); |
||||
return call_->cq()->Pluck(&ops) && ops.got_message; |
||||
} |
||||
|
||||
bool Write(const W& msg, ::grpc::WriteOptions options) { |
||||
if (options.is_last_message()) { |
||||
options.set_buffer_hint(); |
||||
} |
||||
if (!ctx_->pending_ops_.SendMessagePtr(&msg, options).ok()) { |
||||
return false; |
||||
} |
||||
if (!ctx_->sent_initial_metadata_) { |
||||
ctx_->pending_ops_.SendInitialMetadata(&ctx_->initial_metadata_, |
||||
ctx_->initial_metadata_flags()); |
||||
if (ctx_->compression_level_set()) { |
||||
ctx_->pending_ops_.set_compression_level(ctx_->compression_level()); |
||||
} |
||||
ctx_->sent_initial_metadata_ = true; |
||||
} |
||||
call_->PerformOps(&ctx_->pending_ops_); |
||||
// if this is the last message we defer the pluck until AFTER we start
|
||||
// the trailing md op. This prevents hangs. See
|
||||
// https://github.com/grpc/grpc/issues/11546
|
||||
if (options.is_last_message()) { |
||||
ctx_->has_pending_ops_ = true; |
||||
return true; |
||||
} |
||||
ctx_->has_pending_ops_ = false; |
||||
return call_->cq()->Pluck(&ctx_->pending_ops_); |
||||
} |
||||
|
||||
private: |
||||
grpc::internal::Call* const call_; |
||||
::grpc::ServerContext* const ctx_; |
||||
}; |
||||
|
||||
} // namespace internal
|
||||
|
||||
/// Synchronous (blocking) server-side API for a bidirectional
|
||||
/// streaming call, where the incoming message stream coming from the client has
|
||||
/// messages of type \a R, and the outgoing message streaming coming from
|
||||
/// the server has messages of type \a W.
|
||||
template <class W, class R> |
||||
class ServerReaderWriter final : public ServerReaderWriterInterface<W, R> { |
||||
public: |
||||
/// See the \a ServerStreamingInterface.SendInitialMetadata method
|
||||
/// for semantics. Note that initial metadata will be affected by the
|
||||
/// \a ServerContext associated with this call.
|
||||
void SendInitialMetadata() override { body_.SendInitialMetadata(); } |
||||
|
||||
bool NextMessageSize(uint32_t* sz) override { |
||||
return body_.NextMessageSize(sz); |
||||
} |
||||
|
||||
bool Read(R* msg) override { return body_.Read(msg); } |
||||
|
||||
/// See the \a WriterInterface.Write(const W& msg, WriteOptions options)
|
||||
/// method for semantics.
|
||||
/// Side effect:
|
||||
/// Also sends initial metadata if not already sent (using the \a
|
||||
/// ServerContext associated with this call).
|
||||
using internal::WriterInterface<W>::Write; |
||||
bool Write(const W& msg, ::grpc::WriteOptions options) override { |
||||
return body_.Write(msg, options); |
||||
} |
||||
|
||||
private: |
||||
internal::ServerReaderWriterBody<W, R> body_; |
||||
|
||||
friend class ::grpc_impl::internal::TemplatedBidiStreamingHandler< |
||||
ServerReaderWriter<W, R>, false>; |
||||
ServerReaderWriter(::grpc::internal::Call* call, ::grpc::ServerContext* ctx) |
||||
: body_(call, ctx) {} |
||||
}; |
||||
|
||||
/// A class to represent a flow-controlled unary call. This is something
|
||||
/// of a hybrid between conventional unary and streaming. This is invoked
|
||||
/// through a unary call on the client side, but the server responds to it
|
||||
/// as though it were a single-ping-pong streaming call. The server can use
|
||||
/// the \a NextMessageSize method to determine an upper-bound on the size of
|
||||
/// the message. A key difference relative to streaming: ServerUnaryStreamer
|
||||
/// must have exactly 1 Read and exactly 1 Write, in that order, to function
|
||||
/// correctly. Otherwise, the RPC is in error.
|
||||
template <class RequestType, class ResponseType> |
||||
class ServerUnaryStreamer final |
||||
: public ServerReaderWriterInterface<ResponseType, RequestType> { |
||||
public: |
||||
/// Block to send initial metadata to client.
|
||||
/// Implicit input parameter:
|
||||
/// - the \a ServerContext associated with this call will be used for
|
||||
/// sending initial metadata.
|
||||
void SendInitialMetadata() override { body_.SendInitialMetadata(); } |
||||
|
||||
/// Get an upper bound on the request message size from the client.
|
||||
bool NextMessageSize(uint32_t* sz) override { |
||||
return body_.NextMessageSize(sz); |
||||
} |
||||
|
||||
/// 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 ReaderInterface::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.
|
||||
bool Read(RequestType* request) override { |
||||
if (read_done_) { |
||||
return false; |
||||
} |
||||
read_done_ = true; |
||||
return body_.Read(request); |
||||
} |
||||
|
||||
/// Block to write \a msg to the stream with WriteOptions \a options.
|
||||
/// This is thread-safe with respect to \a ReaderInterface::Read
|
||||
///
|
||||
/// \param msg The message to be written to the stream.
|
||||
/// \param options The WriteOptions affecting the write operation.
|
||||
///
|
||||
/// \return \a true on success, \a false when the stream has been closed.
|
||||
using internal::WriterInterface<ResponseType>::Write; |
||||
bool Write(const ResponseType& response, |
||||
::grpc::WriteOptions options) override { |
||||
if (write_done_ || !read_done_) { |
||||
return false; |
||||
} |
||||
write_done_ = true; |
||||
return body_.Write(response, options); |
||||
} |
||||
|
||||
private: |
||||
internal::ServerReaderWriterBody<ResponseType, RequestType> body_; |
||||
bool read_done_; |
||||
bool write_done_; |
||||
|
||||
friend class ::grpc_impl::internal::TemplatedBidiStreamingHandler< |
||||
ServerUnaryStreamer<RequestType, ResponseType>, true>; |
||||
ServerUnaryStreamer(::grpc::internal::Call* call, ::grpc::ServerContext* ctx) |
||||
: body_(call, ctx), read_done_(false), write_done_(false) {} |
||||
}; |
||||
|
||||
/// A class to represent a flow-controlled server-side streaming call.
|
||||
/// This is something of a hybrid between server-side and bidi streaming.
|
||||
/// This is invoked through a server-side streaming call on the client side,
|
||||
/// but the server responds to it as though it were a bidi streaming call that
|
||||
/// must first have exactly 1 Read and then any number of Writes.
|
||||
template <class RequestType, class ResponseType> |
||||
class ServerSplitStreamer final |
||||
: public ServerReaderWriterInterface<ResponseType, RequestType> { |
||||
public: |
||||
/// Block to send initial metadata to client.
|
||||
/// Implicit input parameter:
|
||||
/// - the \a ServerContext associated with this call will be used for
|
||||
/// sending initial metadata.
|
||||
void SendInitialMetadata() override { body_.SendInitialMetadata(); } |
||||
|
||||
/// Get an upper bound on the request message size from the client.
|
||||
bool NextMessageSize(uint32_t* sz) override { |
||||
return body_.NextMessageSize(sz); |
||||
} |
||||
|
||||
/// 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 ReaderInterface::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.
|
||||
bool Read(RequestType* request) override { |
||||
if (read_done_) { |
||||
return false; |
||||
} |
||||
read_done_ = true; |
||||
return body_.Read(request); |
||||
} |
||||
|
||||
/// Block to write \a msg to the stream with WriteOptions \a options.
|
||||
/// This is thread-safe with respect to \a ReaderInterface::Read
|
||||
///
|
||||
/// \param msg The message to be written to the stream.
|
||||
/// \param options The WriteOptions affecting the write operation.
|
||||
///
|
||||
/// \return \a true on success, \a false when the stream has been closed.
|
||||
using internal::WriterInterface<ResponseType>::Write; |
||||
bool Write(const ResponseType& response, |
||||
::grpc::WriteOptions options) override { |
||||
return read_done_ && body_.Write(response, options); |
||||
} |
||||
|
||||
private: |
||||
internal::ServerReaderWriterBody<ResponseType, RequestType> body_; |
||||
bool read_done_; |
||||
|
||||
friend class ::grpc_impl::internal::TemplatedBidiStreamingHandler< |
||||
ServerSplitStreamer<RequestType, ResponseType>, false>; |
||||
ServerSplitStreamer(::grpc::internal::Call* call, ::grpc::ServerContext* ctx) |
||||
: body_(call, ctx), read_done_(false) {} |
||||
}; |
||||
|
||||
} // namespace grpc_impl
|
||||
|
||||
#endif // GRPCPP_IMPL_CODEGEN_SYNC_STREAM_IMPL_H
|
||||
Loading…
Reference in new issue