Chiebot-Mirror
/
grpc
mirror of https://github.com/grpc/grpc.git
8
0
Fork 0
Code Issues Projects Releases Wiki Activity

Merge pull request #11928 from grpc/revert-11572-internalize

Browse Source 
Revert "Separate internal-only and public parts of C++ API"
pull/11932/head
Vijay Pai 8 years ago committed by GitHub
parent 3b914413a8 bc81eee4d4
commit 196d3d6ee8
32 changed files with 661 additions and 857 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 2
      include/grpc++/alarm.h
  2. 12
      include/grpc++/channel.h
  3. 443
      include/grpc++/impl/codegen/async_stream.h
  4. 66
      include/grpc++/impl/codegen/async_unary_call.h
  5. 10
      include/grpc++/impl/codegen/call.h
  6. 2
      include/grpc++/impl/codegen/call_hook.h
  7. 34
      include/grpc++/impl/codegen/channel_interface.h
  8. 28
      include/grpc++/impl/codegen/client_context.h
  9. 4
      include/grpc++/impl/codegen/client_unary_call.h
  10. 56
      include/grpc++/impl/codegen/completion_queue.h
  11. 2
      include/grpc++/impl/codegen/completion_queue_tag.h
  12. 2
      include/grpc++/impl/codegen/metadata_map.h
  13. 2
      include/grpc++/impl/codegen/method_handler_impl.h
  14. 3
      include/grpc++/impl/codegen/rpc_method.h
  15. 3
      include/grpc++/impl/codegen/rpc_service_method.h
  16. 27
      include/grpc++/impl/codegen/server_context.h
  17. 42
      include/grpc++/impl/codegen/server_interface.h
  18. 46
      include/grpc++/impl/codegen/service_type.h
  19. 398
      include/grpc++/impl/codegen/sync_stream.h
  20. 6
      include/grpc++/impl/codegen/time.h
  21. 3
      include/grpc++/server.h
  22. 1
      include/grpc++/server_builder.h
  23. 199
      src/compiler/cpp_generator.cc
  24. 12
      src/cpp/client/channel_cc.cc
  25. 6
      src/cpp/client/generic_stub.cc
  26. 2
      src/cpp/common/completion_queue_cc.cc
  27. 9
      src/cpp/server/health/default_health_check_service.cc
  28. 2
      src/cpp/server/health/default_health_check_service.h
  29. 69
      src/cpp/server/server_cc.cc
  30. 8
      src/cpp/server/server_context.cc
  31. 17
      test/cpp/codegen/compiler_test_golden
  32. 2
      test/cpp/microbenchmarks/bm_cq.cc

2
include/grpc++/alarm.h
Unescape View File

@ -77,7 +77,7 @@ class Alarm : private GrpcLibraryCodegen {
void Cancel() { grpc_alarm_cancel(alarm_); }
private:
class AlarmEntry : public internal::CompletionQueueTag {
class AlarmEntry : public CompletionQueueTag {
public:
AlarmEntry(void* tag) : tag_(tag) {}
bool FinalizeResult(void** tag, bool* status) override {

12
include/grpc++/channel.h
Unescape View File

@ -32,7 +32,7 @@ struct grpc_channel;
namespace grpc {
/// Channels represent a connection to an endpoint. Created by \a CreateChannel.
class Channel final : public ChannelInterface,
public internal::CallHook,
public CallHook,
public std::enable_shared_from_this<Channel>,
private GrpcLibraryCodegen {
public:
@ -52,7 +52,7 @@ class Channel final : public ChannelInterface,
private:
template <class InputMessage, class OutputMessage>
friend Status BlockingUnaryCall(ChannelInterface* channel,
const internal::RpcMethod& method,
const RpcMethod& method,
ClientContext* context,
const InputMessage& request,
OutputMessage* result);
@ -60,11 +60,9 @@ class Channel final : public ChannelInterface,
const grpc::string& host, grpc_channel* c_channel);
Channel(const grpc::string& host, grpc_channel* c_channel);
internal::Call CreateCall(const internal::RpcMethod& method,
ClientContext* context,
CompletionQueue* cq) override;
void PerformOpsOnCall(internal::CallOpSetInterface* ops,
internal::Call* call) override;
Call CreateCall(const RpcMethod& method, ClientContext* context,
CompletionQueue* cq) override;
void PerformOpsOnCall(CallOpSetInterface* ops, Call* call) override;
void* RegisterMethod(const char* method) override;
void NotifyOnStateChangeImpl(grpc_connectivity_state last_observed,

443
include/grpc++/impl/codegen/async_stream.h
Unescape View File

@ -30,7 +30,6 @@ namespace grpc {
class CompletionQueue;
namespace internal {
/// Common interface for all client side asynchronous streaming.
class ClientAsyncStreamingInterface {
public:
@ -147,41 +146,9 @@ class AsyncWriterInterface {
}
};
} // namespace internal
template <class R>
class ClientAsyncReaderInterface
: public internal::ClientAsyncStreamingInterface,
public internal::AsyncReaderInterface<R> {};
/// Common interface for client side asynchronous writing.
template <class W>
class ClientAsyncWriterInterface
: public internal::ClientAsyncStreamingInterface,
public internal::AsyncWriterInterface<W> {
public:
/// Signal the client is done with the writes (half-close the client stream).
/// Thread-safe with respect to \a AsyncReaderInterface::Read
///
/// \param[in] tag The tag identifying the operation.
virtual void WritesDone(void* tag) = 0;
};
/// Async client-side interface for bi-directional streaming,
/// where the client-to-server message stream has messages of type \a W,
/// and the server-to-client message stream has messages of type \a R.
template <class W, class R>
class ClientAsyncReaderWriterInterface
: public internal::ClientAsyncStreamingInterface,
public internal::AsyncWriterInterface<W>,
public internal::AsyncReaderInterface<R> {
public:
/// Signal the client is done with the writes (half-close the client stream).
/// Thread-safe with respect to \a AsyncReaderInterface::Read
///
/// \param[in] tag The tag identifying the operation.
virtual void WritesDone(void* tag) = 0;
};
class ClientAsyncReaderInterface : public ClientAsyncStreamingInterface,
public AsyncReaderInterface<R> {};
/// Async client-side API for doing server-streaming RPCs,
/// where the incoming message stream coming from the server has
@ -189,24 +156,21 @@ class ClientAsyncReaderWriterInterface
template <class R>
class ClientAsyncReader final : public ClientAsyncReaderInterface<R> {
public:
struct internal {
/// Create a stream and write the first request out.
/// \a tag will be notified on \a cq when the call has been started and
/// \a request has been written out.
/// Note that \a context will be used to fill in custom initial metadata
/// used to send to the server when starting the call.
template <class W>
static ClientAsyncReader* Create(::grpc::ChannelInterface* channel,
CompletionQueue* cq,
const ::grpc::internal::RpcMethod& method,
ClientContext* context, const W& request,
void* tag) {
::grpc::internal::Call call = channel->CreateCall(method, context, cq);
return new (g_core_codegen_interface->grpc_call_arena_alloc(
call.call(), sizeof(ClientAsyncReader)))
ClientAsyncReader(call, context, request, tag);
}
};
/// Create a stream and write the first request out.
/// \a tag will be notified on \a cq when the call has been started and
/// \a request has been written out.
/// Note that \a context will be used to fill in custom initial metadata
/// used to send to the server when starting the call.
template <class W>
static ClientAsyncReader* Create(ChannelInterface* channel,
CompletionQueue* cq, const RpcMethod& method,
ClientContext* context, const W& request,
void* tag) {
Call call = channel->CreateCall(method, context, cq);
return new (g_core_codegen_interface->grpc_call_arena_alloc(
call.call(), sizeof(ClientAsyncReader)))
ClientAsyncReader(call, context, request, tag);
}
// always allocated against a call arena, no memory free required
static void operator delete(void* ptr, std::size_t size) {
@ -254,8 +218,8 @@ class ClientAsyncReader final : public ClientAsyncReaderInterface<R> {
private:
template <class W>
ClientAsyncReader(::grpc::internal::Call call, ClientContext* context,
const W& request, void* tag)
ClientAsyncReader(Call call, ClientContext* context, const W& request,
void* tag)
: context_(context), call_(call) {
init_ops_.set_output_tag(tag);
init_ops_.SendInitialMetadata(context->send_initial_metadata_,
@ -267,19 +231,24 @@ class ClientAsyncReader final : public ClientAsyncReaderInterface<R> {
}
ClientContext* context_;
::grpc::internal::Call call_;
::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata,
::grpc::internal::CallOpSendMessage,
::grpc::internal::CallOpClientSendClose>
Call call_;
CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage, CallOpClientSendClose>
init_ops_;
::grpc::internal::CallOpSet<::grpc::internal::CallOpRecvInitialMetadata>
meta_ops_;
::grpc::internal::CallOpSet<::grpc::internal::CallOpRecvInitialMetadata,
::grpc::internal::CallOpRecvMessage<R>>
read_ops_;
::grpc::internal::CallOpSet<::grpc::internal::CallOpRecvInitialMetadata,
::grpc::internal::CallOpClientRecvStatus>
finish_ops_;
CallOpSet<CallOpRecvInitialMetadata> meta_ops_;
CallOpSet<CallOpRecvInitialMetadata, CallOpRecvMessage<R>> read_ops_;
CallOpSet<CallOpRecvInitialMetadata, CallOpClientRecvStatus> finish_ops_;
};
/// Common interface for client side asynchronous writing.
template <class W>
class ClientAsyncWriterInterface : public ClientAsyncStreamingInterface,
public AsyncWriterInterface<W> {
public:
/// Signal the client is done with the writes (half-close the client stream).
/// Thread-safe with respect to \a AsyncReaderInterface::Read
///
/// \param[in] tag The tag identifying the operation.
virtual void WritesDone(void* tag) = 0;
};
/// Async API on the client side for doing client-streaming RPCs,
@ -288,27 +257,24 @@ class ClientAsyncReader final : public ClientAsyncReaderInterface<R> {
template <class W>
class ClientAsyncWriter final : public ClientAsyncWriterInterface<W> {
public:
struct internal {
/// Create a stream and write the first request out.
/// \a tag will be notified on \a cq when the call has been started (i.e.
/// intitial metadata sent) and \a request has been written out.
/// Note that \a context will be used to fill in custom initial metadata
/// used to send to the server when starting the call.
/// \a response will be filled in with the single expected response
/// message from the server upon a successful call to the \a Finish
/// method of this instance.
template <class R>
static ClientAsyncWriter* Create(::grpc::ChannelInterface* channel,
CompletionQueue* cq,
const ::grpc::internal::RpcMethod& method,
ClientContext* context, R* response,
void* tag) {
::grpc::internal::Call call = channel->CreateCall(method, context, cq);
return new (g_core_codegen_interface->grpc_call_arena_alloc(
call.call(), sizeof(ClientAsyncWriter)))
ClientAsyncWriter(call, context, response, tag);
}
};
/// Create a stream and write the first request out.
/// \a tag will be notified on \a cq when the call has been started (i.e.
/// intitial metadata sent) and \a request has been written out.
/// Note that \a context will be used to fill in custom initial metadata
/// used to send to the server when starting the call.
/// \a response will be filled in with the single expected response
/// message from the server upon a successful call to the \a Finish
/// method of this instance.
template <class R>
static ClientAsyncWriter* Create(ChannelInterface* channel,
CompletionQueue* cq, const RpcMethod& method,
ClientContext* context, R* response,
void* tag) {
Call call = channel->CreateCall(method, context, cq);
return new (g_core_codegen_interface->grpc_call_arena_alloc(
call.call(), sizeof(ClientAsyncWriter)))
ClientAsyncWriter(call, context, response, tag);
}
// always allocated against a call arena, no memory free required
static void operator delete(void* ptr, std::size_t size) {
@ -372,8 +338,7 @@ class ClientAsyncWriter final : public ClientAsyncWriterInterface<W> {
private:
template <class R>
ClientAsyncWriter(::grpc::internal::Call call, ClientContext* context,
R* response, void* tag)
ClientAsyncWriter(Call call, ClientContext* context, R* response, void* tag)
: context_(context), call_(call) {
finish_ops_.RecvMessage(response);
finish_ops_.AllowNoMessage();
@ -391,19 +356,30 @@ class ClientAsyncWriter final : public ClientAsyncWriterInterface<W> {
}
ClientContext* context_;
::grpc::internal::Call call_;
::grpc::internal::CallOpSet<::grpc::internal::CallOpRecvInitialMetadata>
meta_ops_;
::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata,
::grpc::internal::CallOpSendMessage,
::grpc::internal::CallOpClientSendClose>
Call call_;
CallOpSet<CallOpRecvInitialMetadata> meta_ops_;
CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage, CallOpClientSendClose>
write_ops_;
::grpc::internal::CallOpSet<::grpc::internal::CallOpRecvInitialMetadata,
::grpc::internal::CallOpGenericRecvMessage,
::grpc::internal::CallOpClientRecvStatus>
CallOpSet<CallOpRecvInitialMetadata, CallOpGenericRecvMessage,
CallOpClientRecvStatus>
finish_ops_;
};
/// Async client-side interface for bi-directional streaming,
/// where the client-to-server message stream has messages of type \a W,
/// and the server-to-client message stream has messages of type \a R.
template <class W, class R>
class ClientAsyncReaderWriterInterface : public ClientAsyncStreamingInterface,
public AsyncWriterInterface<W>,
public AsyncReaderInterface<R> {
public:
/// Signal the client is done with the writes (half-close the client stream).
/// Thread-safe with respect to \a AsyncReaderInterface::Read
///
/// \param[in] tag The tag identifying the operation.
virtual void WritesDone(void* tag) = 0;
};
/// Async client-side interface for bi-directional streaming,
/// where the outgoing message stream going to the server
/// has messages of type \a W, and the incoming message stream coming
@ -412,23 +388,21 @@ template <class W, class R>
class ClientAsyncReaderWriter final
: public ClientAsyncReaderWriterInterface<W, R> {
public:
struct internal {
/// Create a stream and write the first request out.
/// \a tag will be notified on \a cq when the call has been started (i.e.
/// intitial metadata sent).
/// Note that \a context will be used to fill in custom initial metadata
/// used to send to the server when starting the call.
static ClientAsyncReaderWriter* Create(
::grpc::ChannelInterface* channel, CompletionQueue* cq,
const ::grpc::internal::RpcMethod& method, ClientContext* context,
void* tag) {
::grpc::internal::Call call = channel->CreateCall(method, context, cq);
return new (g_core_codegen_interface->grpc_call_arena_alloc(
call.call(), sizeof(ClientAsyncReaderWriter)))
ClientAsyncReaderWriter(call, context, tag);
}
};
/// Create a stream and write the first request out.
/// \a tag will be notified on \a cq when the call has been started (i.e.
/// intitial metadata sent).
/// Note that \a context will be used to fill in custom initial metadata
/// used to send to the server when starting the call.
static ClientAsyncReaderWriter* Create(ChannelInterface* channel,
CompletionQueue* cq,
const RpcMethod& method,
ClientContext* context, void* tag) {
Call call = channel->CreateCall(method, context, cq);
return new (g_core_codegen_interface->grpc_call_arena_alloc(
call.call(), sizeof(ClientAsyncReaderWriter)))
ClientAsyncReaderWriter(call, context, tag);
}
// always allocated against a call arena, no memory free required
static void operator delete(void* ptr, std::size_t size) {
@ -497,8 +471,7 @@ class ClientAsyncReaderWriter final
}
private:
ClientAsyncReaderWriter(::grpc::internal::Call call, ClientContext* context,
void* tag)
ClientAsyncReaderWriter(Call call, ClientContext* context, void* tag)
: context_(context), call_(call) {
if (context_->initial_metadata_corked_) {
// if corked bit is set in context, we buffer up the initial metadata to
@ -514,25 +487,17 @@ class ClientAsyncReaderWriter final
}
ClientContext* context_;
::grpc::internal::Call call_;
::grpc::internal::CallOpSet<::grpc::internal::CallOpRecvInitialMetadata>
meta_ops_;
::grpc::internal::CallOpSet<::grpc::internal::CallOpRecvInitialMetadata,
::grpc::internal::CallOpRecvMessage<R>>
read_ops_;
::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata,
::grpc::internal::CallOpSendMessage,
::grpc::internal::CallOpClientSendClose>
Call call_;
CallOpSet<CallOpRecvInitialMetadata> meta_ops_;
CallOpSet<CallOpRecvInitialMetadata, CallOpRecvMessage<R>> read_ops_;
CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage, CallOpClientSendClose>
write_ops_;
::grpc::internal::CallOpSet<::grpc::internal::CallOpRecvInitialMetadata,
::grpc::internal::CallOpClientRecvStatus>
finish_ops_;
CallOpSet<CallOpRecvInitialMetadata, CallOpClientRecvStatus> finish_ops_;
};
template <class W, class R>
class ServerAsyncReaderInterface
: public internal::ServerAsyncStreamingInterface,
public internal::AsyncReaderInterface<R> {
class ServerAsyncReaderInterface : public ServerAsyncStreamingInterface,
public AsyncReaderInterface<R> {
public:
/// Indicate that the stream is to be finished with a certain status code
/// and also send out \a msg response to the client.
@ -576,89 +541,6 @@ class ServerAsyncReaderInterface
virtual void FinishWithError(const Status& status, void* tag) = 0;
};
template <class W>
class ServerAsyncWriterInterface
: public internal::ServerAsyncStreamingInterface,
public internal::AsyncWriterInterface<W> {
public:
/// Indicate that the stream is to be finished with a certain status code.
/// Request notification for when the server has sent the appropriate
/// signals to the client to end the call.
/// Should not be used concurrently with other operations.
///
/// It is appropriate to call this method when either:
/// * all messages from the client have been received (either known
/// implictly, or explicitly because a previous \a
/// AsyncReaderInterface::Read operation with a non-ok
/// result (e.g., cq->Next(&read_tag, &ok) filled in 'ok' with 'false'.
/// * it is desired to end the call early with some non-OK status code.
///
/// This operation will end when the server has finished sending out initial
/// metadata (if not sent already), response message, and status, or if
/// some failure occurred when trying to do so.
///
/// \param[in] tag Tag identifying this request.
/// \param[in] status To be sent to the client as the result of this call.
virtual void Finish(const Status& status, void* tag) = 0;
/// Request the writing of \a msg and coalesce it with trailing metadata which
/// contains \a status, using WriteOptions options with
/// identifying tag \a tag.
///
/// WriteAndFinish is equivalent of performing WriteLast and Finish
/// in a single step.
///
/// \param[in] msg The message to be written.
/// \param[in] options The WriteOptions to be used to write this message.
/// \param[in] status The Status that server returns to client.
/// \param[in] tag The tag identifying the operation.
virtual void WriteAndFinish(const W& msg, WriteOptions options,
const Status& status, void* tag) = 0;
};
/// Server-side interface for asynchronous bi-directional streaming.
template <class W, class R>
class ServerAsyncReaderWriterInterface
: public internal::ServerAsyncStreamingInterface,
public internal::AsyncWriterInterface<W>,
public internal::AsyncReaderInterface<R> {
public:
/// Indicate that the stream is to be finished with a certain status code.
/// Request notification for when the server has sent the appropriate
/// signals to the client to end the call.
/// Should not be used concurrently with other operations.
///
/// It is appropriate to call this method when either:
/// * all messages from the client have been received (either known
/// implictly, or explicitly because a previous \a
/// AsyncReaderInterface::Read operation
/// with a non-ok result (e.g., cq->Next(&read_tag, &ok) filled in 'ok'
/// with 'false'.
/// * it is desired to end the call early with some non-OK status code.
///
/// This operation will end when the server has finished sending out initial
/// metadata (if not sent already), response message, and status, or if some
/// failure occurred when trying to do so.
///
/// \param[in] tag Tag identifying this request.
/// \param[in] status To be sent to the client as the result of this call.
virtual void Finish(const Status& status, void* tag) = 0;
/// Request the writing of \a msg and coalesce it with trailing metadata which
/// contains \a status, using WriteOptions options with
/// identifying tag \a tag.
///
/// WriteAndFinish is equivalent of performing WriteLast and Finish in a
/// single step.
///
/// \param[in] msg The message to be written.
/// \param[in] options The WriteOptions to be used to write this message.
/// \param[in] status The Status that server returns to client.
/// \param[in] tag The tag identifying the operation.
virtual void WriteAndFinish(const W& msg, WriteOptions options,
const Status& status, void* tag) = 0;
};
/// Async server-side API for doing client-streaming RPCs,
/// where the incoming message stream from the client has messages of type \a R,
/// and the single response message sent from the server is type \a W.
@ -742,19 +624,56 @@ class ServerAsyncReader final : public ServerAsyncReaderInterface<W, R> {
}
private:
void BindCall(::grpc::internal::Call* call) override { call_ = *call; }
void BindCall(Call* call) override { call_ = *call; }
::grpc::internal::Call call_;
Call call_;
ServerContext* ctx_;
::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata>
meta_ops_;
::grpc::internal::CallOpSet<::grpc::internal::CallOpRecvMessage<R>> read_ops_;
::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata,
::grpc::internal::CallOpSendMessage,
::grpc::internal::CallOpServerSendStatus>
CallOpSet<CallOpSendInitialMetadata> meta_ops_;
CallOpSet<CallOpRecvMessage<R>> read_ops_;
CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage,
CallOpServerSendStatus>
finish_ops_;
};
template <class W>
class ServerAsyncWriterInterface : public ServerAsyncStreamingInterface,
public AsyncWriterInterface<W> {
public:
/// Indicate that the stream is to be finished with a certain status code.
/// Request notification for when the server has sent the appropriate
/// signals to the client to end the call.
/// Should not be used concurrently with other operations.
///
/// It is appropriate to call this method when either:
/// * all messages from the client have been received (either known
/// implictly, or explicitly because a previous \a
/// AsyncReaderInterface::Read operation with a non-ok
/// result (e.g., cq->Next(&read_tag, &ok) filled in 'ok' with 'false'.
/// * it is desired to end the call early with some non-OK status code.
///
/// This operation will end when the server has finished sending out initial
/// metadata (if not sent already), response message, and status, or if
/// some failure occurred when trying to do so.
///
/// \param[in] tag Tag identifying this request.
/// \param[in] status To be sent to the client as the result of this call.
virtual void Finish(const Status& status, void* tag) = 0;
/// Request the writing of \a msg and coalesce it with trailing metadata which
/// contains \a status, using WriteOptions options with
/// identifying tag \a tag.
///
/// WriteAndFinish is equivalent of performing WriteLast and Finish
/// in a single step.
///
/// \param[in] msg The message to be written.
/// \param[in] options The WriteOptions to be used to write this message.
/// \param[in] status The Status that server returns to client.
/// \param[in] tag The tag identifying the operation.
virtual void WriteAndFinish(const W& msg, WriteOptions options,
const Status& status, void* tag) = 0;
};
/// Async server-side API for doing server streaming RPCs,
/// where the outgoing message stream from the server has messages of type \a W.
template <class W>
@ -836,7 +755,7 @@ class ServerAsyncWriter final : public ServerAsyncWriterInterface<W> {
}
private:
void BindCall(::grpc::internal::Call* call) override { call_ = *call; }
void BindCall(Call* call) override { call_ = *call; }
template <class T>
void EnsureInitialMetadataSent(T* ops) {
@ -850,17 +769,55 @@ class ServerAsyncWriter final : public ServerAsyncWriterInterface<W> {
}
}
::grpc::internal::Call call_;
Call call_;
ServerContext* ctx_;
::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata>
meta_ops_;
::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata,
::grpc::internal::CallOpSendMessage,
::grpc::internal::CallOpServerSendStatus>
CallOpSet<CallOpSendInitialMetadata> meta_ops_;
CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage,
CallOpServerSendStatus>
write_ops_;
::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata,
::grpc::internal::CallOpServerSendStatus>
finish_ops_;
CallOpSet<CallOpSendInitialMetadata, CallOpServerSendStatus> finish_ops_;
};
/// Server-side interface for asynchronous bi-directional streaming.
template <class W, class R>
class ServerAsyncReaderWriterInterface : public ServerAsyncStreamingInterface,
public AsyncWriterInterface<W>,
public AsyncReaderInterface<R> {
public:
/// Indicate that the stream is to be finished with a certain status code.
/// Request notification for when the server has sent the appropriate
/// signals to the client to end the call.
/// Should not be used concurrently with other operations.
///
/// It is appropriate to call this method when either:
/// * all messages from the client have been received (either known
/// implictly, or explicitly because a previous \a
/// AsyncReaderInterface::Read operation
/// with a non-ok result (e.g., cq->Next(&read_tag, &ok) filled in 'ok'
/// with 'false'.
/// * it is desired to end the call early with some non-OK status code.
///
/// This operation will end when the server has finished sending out initial
/// metadata (if not sent already), response message, and status, or if some
/// failure occurred when trying to do so.
///
/// \param[in] tag Tag identifying this request.
/// \param[in] status To be sent to the client as the result of this call.
virtual void Finish(const Status& status, void* tag) = 0;
/// Request the writing of \a msg and coalesce it with trailing metadata which
/// contains \a status, using WriteOptions options with
/// identifying tag \a tag.
///
/// WriteAndFinish is equivalent of performing WriteLast and Finish in a
/// single step.
///
/// \param[in] msg The message to be written.
/// \param[in] options The WriteOptions to be used to write this message.
/// \param[in] status The Status that server returns to client.
/// \param[in] tag The tag identifying the operation.
virtual void WriteAndFinish(const W& msg, WriteOptions options,
const Status& status, void* tag) = 0;
};
/// Async server-side API for doing bidirectional streaming RPCs,
@ -955,7 +912,7 @@ class ServerAsyncReaderWriter final
private:
friend class ::grpc::Server;
void BindCall(::grpc::internal::Call* call) override { call_ = *call; }
void BindCall(Call* call) override { call_ = *call; }
template <class T>
void EnsureInitialMetadataSent(T* ops) {
@ -969,18 +926,14 @@ class ServerAsyncReaderWriter final
}
}
::grpc::internal::Call call_;
Call call_;
ServerContext* ctx_;
::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata>
meta_ops_;
::grpc::internal::CallOpSet<::grpc::internal::CallOpRecvMessage<R>> read_ops_;
::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata,
::grpc::internal::CallOpSendMessage,
::grpc::internal::CallOpServerSendStatus>
CallOpSet<CallOpSendInitialMetadata> meta_ops_;
CallOpSet<CallOpRecvMessage<R>> read_ops_;
CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage,
CallOpServerSendStatus>
write_ops_;
::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata,
::grpc::internal::CallOpServerSendStatus>
finish_ops_;
CallOpSet<CallOpSendInitialMetadata, CallOpServerSendStatus> finish_ops_;
};
} // namespace grpc

66
include/grpc++/impl/codegen/async_unary_call.h
Unescape View File

@ -75,18 +75,17 @@ class ClientAsyncResponseReader final
/// intitial metadata sent) and \a request has been written out.
/// Note that \a context will be used to fill in custom initial metadata
/// used to send to the server when starting the call.
struct internal {
template <class W>
static ClientAsyncResponseReader* Create(
::grpc::ChannelInterface* channel, CompletionQueue* cq,
const ::grpc::internal::RpcMethod& method, ClientContext* context,
const W& request) {
::grpc::internal::Call call = channel->CreateCall(method, context, cq);
return new (g_core_codegen_interface->grpc_call_arena_alloc(
call.call(), sizeof(ClientAsyncResponseReader)))
ClientAsyncResponseReader(call, context, request);
}
};
template <class W>
static ClientAsyncResponseReader* Create(ChannelInterface* channel,
CompletionQueue* cq,
const RpcMethod& method,
ClientContext* context,
const W& request) {
Call call = channel->CreateCall(method, context, cq);
return new (g_core_codegen_interface->grpc_call_arena_alloc(
call.call(), sizeof(ClientAsyncResponseReader)))
ClientAsyncResponseReader(call, context, request);
}
/// TODO(vjpai): Delete the below constructor
/// PLEASE DO NOT USE THIS CONSTRUCTOR IN NEW CODE
@ -95,10 +94,9 @@ class ClientAsyncResponseReader final
/// created this struct rather than properly using a stub.
/// This code will not remain a valid public constructor for long.
template <class W>
ClientAsyncResponseReader(::grpc::ChannelInterface* channel,
CompletionQueue* cq,
const ::grpc::internal::RpcMethod& method,
ClientContext* context, const W& request)
ClientAsyncResponseReader(ChannelInterface* channel, CompletionQueue* cq,
const RpcMethod& method, ClientContext* context,
const W& request)
: context_(context),
call_(channel->CreateCall(method, context, cq)),
collection_(std::make_shared<Ops>()) {
@ -166,11 +164,10 @@ class ClientAsyncResponseReader final
private:
ClientContext* const context_;
::grpc::internal::Call call_;
Call call_;
template <class W>
ClientAsyncResponseReader(::grpc::internal::Call call, ClientContext* context,
const W& request)
ClientAsyncResponseReader(Call call, ClientContext* context, const W& request)
: context_(context), call_(call) {
ops_.init_buf.SendInitialMetadata(context->send_initial_metadata_,
context->initial_metadata_flags());
@ -186,17 +183,13 @@ class ClientAsyncResponseReader final
// TODO(vjpai): Remove the reference to CallOpSetCollectionInterface
// as soon as the related workaround (public constructor) is deleted
struct Ops : public ::grpc::internal::CallOpSetCollectionInterface {
::grpc::internal::SneakyCallOpSet<
::grpc::internal::CallOpSendInitialMetadata,
::grpc::internal::CallOpSendMessage,
::grpc::internal::CallOpClientSendClose>
struct Ops : public CallOpSetCollectionInterface {
SneakyCallOpSet<CallOpSendInitialMetadata, CallOpSendMessage,
CallOpClientSendClose>
init_buf;
::grpc::internal::CallOpSet<::grpc::internal::CallOpRecvInitialMetadata>
meta_buf;
::grpc::internal::CallOpSet<::grpc::internal::CallOpRecvInitialMetadata,
::grpc::internal::CallOpRecvMessage<R>,
::grpc::internal::CallOpClientRecvStatus>
CallOpSet<CallOpRecvInitialMetadata> meta_buf;
CallOpSet<CallOpRecvInitialMetadata, CallOpRecvMessage<R>,
CallOpClientRecvStatus>
finish_buf;
} ops_;
@ -208,8 +201,7 @@ class ClientAsyncResponseReader final
/// 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 internal::ServerAsyncStreamingInterface {
class ServerAsyncResponseWriter final : public ServerAsyncStreamingInterface {
public:
explicit ServerAsyncResponseWriter(ServerContext* ctx)
: call_(nullptr, nullptr, nullptr), ctx_(ctx) {}
@ -297,15 +289,13 @@ class ServerAsyncResponseWriter final
}
private:
void BindCall(::grpc::internal::Call* call) override { call_ = *call; }
void BindCall(Call* call) override { call_ = *call; }
::grpc::internal::Call call_;
Call call_;
ServerContext* ctx_;
::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata>
meta_buf_;
::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata,
::grpc::internal::CallOpSendMessage,
::grpc::internal::CallOpServerSendStatus>
CallOpSet<CallOpSendInitialMetadata> meta_buf_;
CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage,
CallOpServerSendStatus>
finish_buf_;
};

10
include/grpc++/impl/codegen/call.h
Unescape View File

@ -44,12 +44,10 @@ struct grpc_byte_buffer;
namespace grpc {
class ByteBuffer;
class CompletionQueue;
extern CoreCodegenInterface* g_core_codegen_interface;
namespace internal {
class Call;
class CallHook;
class CompletionQueue;
extern CoreCodegenInterface* g_core_codegen_interface;
const char kBinaryErrorDetailsKey[] = "grpc-status-details-bin";
@ -78,7 +76,6 @@ inline grpc_metadata* FillMetadataArray(
}
return metadata_array;
}
} // namespace internal
/// Per-message write options.
class WriteOptions {
@ -194,7 +191,6 @@ class WriteOptions {
bool last_message_;
};
namespace internal {
/// Default argument for CallOpSet. I is unused by the class, but can be
/// used for generating multiple names for the same thing.
template <int I>
@ -682,7 +678,7 @@ class Call final {
grpc_call* call_;
int max_receive_message_size_;
};
} // namespace internal
} // namespace grpc
#endif // GRPCXX_IMPL_CODEGEN_CALL_H

2
include/grpc++/impl/codegen/call_hook.h
Unescape View File

@ -21,7 +21,6 @@
namespace grpc {
namespace internal {
class CallOpSetInterface;
class Call;
@ -32,7 +31,6 @@ class CallHook {
virtual ~CallHook() {}
virtual void PerformOpsOnCall(CallOpSetInterface* ops, Call* call) = 0;
};
} // namespace internal
} // namespace grpc

34
include/grpc++/impl/codegen/channel_interface.h
Unescape View File

@ -24,8 +24,10 @@
#include <grpc/impl/codegen/connectivity_state.h>
namespace grpc {
class ChannelInterface;
class Call;
class ClientContext;
class RpcMethod;
class CallOpSetInterface;
class CompletionQueue;
template <class R>
@ -43,16 +45,6 @@ class ClientAsyncReaderWriter;
template <class R>
class ClientAsyncResponseReader;
namespace internal {
class Call;
class CallOpSetInterface;
class RpcMethod;
template <class InputMessage, class OutputMessage>
Status BlockingUnaryCall(ChannelInterface* channel, const RpcMethod& method,
ClientContext* context, const InputMessage& request,
OutputMessage* result);
} // namespace internal
/// Codegen interface for \a grpc::Channel.
class ChannelInterface {
public:
@ -104,16 +96,15 @@ class ChannelInterface {
template <class R>
friend class ::grpc::ClientAsyncResponseReader;
template <class InputMessage, class OutputMessage>
friend Status(::grpc::internal::BlockingUnaryCall)(
ChannelInterface* channel, const internal::RpcMethod& method,
ClientContext* context, const InputMessage& request,
OutputMessage* result);
friend class ::grpc::internal::RpcMethod;
virtual internal::Call CreateCall(const internal::RpcMethod& method,
ClientContext* context,
CompletionQueue* cq) = 0;
virtual void PerformOpsOnCall(internal::CallOpSetInterface* ops,
internal::Call* call) = 0;
friend Status BlockingUnaryCall(ChannelInterface* channel,
const RpcMethod& method,
ClientContext* context,
const InputMessage& request,
OutputMessage* result);
friend class ::grpc::RpcMethod;
virtual Call CreateCall(const RpcMethod& method, ClientContext* context,
CompletionQueue* cq) = 0;
virtual void PerformOpsOnCall(CallOpSetInterface* ops, Call* call) = 0;
virtual void* RegisterMethod(const char* method) = 0;
virtual void NotifyOnStateChangeImpl(grpc_connectivity_state last_observed,
gpr_timespec deadline,
@ -121,6 +112,7 @@ class ChannelInterface {
virtual bool WaitForStateChangeImpl(grpc_connectivity_state last_observed,
gpr_timespec deadline) = 0;
};
} // namespace grpc
#endif // GRPCXX_IMPL_CODEGEN_CHANNEL_INTERFACE_H

28
include/grpc++/impl/codegen/client_context.h
Unescape View File

@ -60,18 +60,7 @@ class Channel;
class ChannelInterface;
class CompletionQueue;
class CallCredentials;
class ClientContext;
namespace internal {
class RpcMethod;
class CallOpClientRecvStatus;
class CallOpRecvInitialMetadata;
template <class InputMessage, class OutputMessage>
Status BlockingUnaryCall(ChannelInterface* channel, const RpcMethod& method,
ClientContext* context, const InputMessage& request,
OutputMessage* result);
} // namespace internal
template <class R>
class ClientReader;
template <class W>
@ -356,8 +345,8 @@ class ClientContext {
ClientContext& operator=(const ClientContext&);
friend class ::grpc::testing::InteropClientContextInspector;
friend class ::grpc::internal::CallOpClientRecvStatus;
friend class ::grpc::internal::CallOpRecvInitialMetadata;
friend class CallOpClientRecvStatus;
friend class CallOpRecvInitialMetadata;
friend class Channel;
template <class R>
friend class ::grpc::ClientReader;
@ -374,10 +363,11 @@ class ClientContext {
template <class R>
friend class ::grpc::ClientAsyncResponseReader;
template <class InputMessage, class OutputMessage>
friend Status(::grpc::internal::BlockingUnaryCall)(
ChannelInterface* channel, const internal::RpcMethod& method,
ClientContext* context, const InputMessage& request,
OutputMessage* result);
friend Status BlockingUnaryCall(ChannelInterface* channel,
const RpcMethod& method,
ClientContext* context,
const InputMessage& request,
OutputMessage* result);
grpc_call* call() const { return call_; }
void set_call(grpc_call* call, const std::shared_ptr<Channel>& channel);
@ -409,8 +399,8 @@ class ClientContext {
mutable std::shared_ptr<const AuthContext> auth_context_;
struct census_context* census_context_;
std::multimap<grpc::string, grpc::string> send_initial_metadata_;
internal::MetadataMap recv_initial_metadata_;
internal::MetadataMap trailing_metadata_;
MetadataMap recv_initial_metadata_;
MetadataMap trailing_metadata_;
grpc_call* propagate_from_call_;
PropagationOptions propagation_options_;

4
include/grpc++/impl/codegen/client_unary_call.h
Unescape View File

@ -30,9 +30,8 @@ namespace grpc {
class Channel;
class ClientContext;
class CompletionQueue;
namespace internal {
class RpcMethod;
/// Wrapper that performs a blocking unary call
template <class InputMessage, class OutputMessage>
Status BlockingUnaryCall(ChannelInterface* channel, const RpcMethod& method,
@ -68,7 +67,6 @@ Status BlockingUnaryCall(ChannelInterface* channel, const RpcMethod& method,
return status;
}
} // namespace internal
} // namespace grpc
#endif // GRPCXX_IMPL_CODEGEN_CLIENT_UNARY_CALL_H

56
include/grpc++/impl/codegen/completion_queue.h
Unescape View File

@ -56,19 +56,7 @@ class ServerWriter;
namespace internal {
template <class W, class R>
class ServerReaderWriterBody;
} // namespace internal
class Channel;
class ChannelInterface;
class ClientContext;
class CompletionQueue;
class Server;
class ServerBuilder;
class ServerContext;
namespace internal {
class CompletionQueueTag;
class RpcMethod;
}
template <class ServiceType, class RequestType, class ResponseType>
class RpcMethodHandler;
template <class ServiceType, class RequestType, class ResponseType>
@ -78,13 +66,16 @@ class ServerStreamingHandler;
template <class ServiceType, class RequestType, class ResponseType>
class BidiStreamingHandler;
class UnknownMethodHandler;
template <class Streamer, bool WriteNeeded>
class TemplatedBidiStreamingHandler;
template <class InputMessage, class OutputMessage>
Status BlockingUnaryCall(ChannelInterface* channel, const RpcMethod& method,
ClientContext* context, const InputMessage& request,
OutputMessage* result);
} // namespace internal
class Channel;
class ChannelInterface;
class ClientContext;
class CompletionQueueTag;
class CompletionQueue;
class RpcMethod;
class Server;
class ServerBuilder;
class ServerContext;
extern CoreCodegenInterface* g_core_codegen_interface;
@ -205,27 +196,28 @@ class CompletionQueue : private GrpcLibraryCodegen {
template <class W, class R>
friend class ::grpc::internal::ServerReaderWriterBody;
template <class ServiceType, class RequestType, class ResponseType>
friend class ::grpc::internal::RpcMethodHandler;
friend class RpcMethodHandler;
template <class ServiceType, class RequestType, class ResponseType>
friend class ::grpc::internal::ClientStreamingHandler;
friend class ClientStreamingHandler;
template <class ServiceType, class RequestType, class ResponseType>
friend class ::grpc::internal::ServerStreamingHandler;
friend class ServerStreamingHandler;
template <class Streamer, bool WriteNeeded>
friend class ::grpc::internal::TemplatedBidiStreamingHandler;
friend class ::grpc::internal::UnknownMethodHandler;
friend class TemplatedBidiStreamingHandler;
friend class UnknownMethodHandler;
friend class ::grpc::Server;
friend class ::grpc::ServerContext;
template <class InputMessage, class OutputMessage>
friend Status(::grpc::internal::BlockingUnaryCall)(
ChannelInterface* channel, const internal::RpcMethod& method,
ClientContext* context, const InputMessage& request,
OutputMessage* result);
friend Status BlockingUnaryCall(ChannelInterface* channel,
const RpcMethod& method,
ClientContext* context,
const InputMessage& request,
OutputMessage* result);
NextStatus AsyncNextInternal(void** tag, bool* ok, gpr_timespec deadline);
/// Wraps \a grpc_completion_queue_pluck.
/// \warning Must not be mixed with calls to \a Next.
bool Pluck(internal::CompletionQueueTag* tag) {
bool Pluck(CompletionQueueTag* tag) {
auto deadline =
g_core_codegen_interface->gpr_inf_future(GPR_CLOCK_REALTIME);
auto ev = g_core_codegen_interface->grpc_completion_queue_pluck(
@ -246,7 +238,7 @@ class CompletionQueue : private GrpcLibraryCodegen {
/// implementation to simple call the other TryPluck function with a zero
/// timeout. i.e:
/// TryPluck(tag, gpr_time_0(GPR_CLOCK_REALTIME))
void TryPluck(internal::CompletionQueueTag* tag) {
void TryPluck(CompletionQueueTag* tag) {
auto deadline = g_core_codegen_interface->gpr_time_0(GPR_CLOCK_REALTIME);
auto ev = g_core_codegen_interface->grpc_completion_queue_pluck(
cq_, tag, deadline, nullptr);
@ -262,7 +254,7 @@ class CompletionQueue : private GrpcLibraryCodegen {
///
/// This exects tag->FinalizeResult (if called) to return 'false' i.e expects
/// that the tag is internal not something that is returned to the user.
void TryPluck(internal::CompletionQueueTag* tag, gpr_timespec deadline) {
void TryPluck(CompletionQueueTag* tag, gpr_timespec deadline) {
auto ev = g_core_codegen_interface->grpc_completion_queue_pluck(
cq_, tag, deadline, nullptr);
if (ev.type == GRPC_QUEUE_TIMEOUT || ev.type == GRPC_QUEUE_SHUTDOWN) {

2
include/grpc++/impl/codegen/completion_queue_tag.h
Unescape View File

@ -21,7 +21,6 @@
namespace grpc {
namespace internal {
/// An interface allowing implementors to process and filter event tags.
class CompletionQueueTag {
public:
@ -32,7 +31,6 @@ class CompletionQueueTag {
/// queue
virtual bool FinalizeResult(void** tag, bool* status) = 0;
};
} // namespace internal
} // namespace grpc

2
include/grpc++/impl/codegen/metadata_map.h
Unescape View File

@ -23,7 +23,6 @@
namespace grpc {
namespace internal {
class MetadataMap {
public:
MetadataMap() { memset(&arr_, 0, sizeof(arr_)); }
@ -51,7 +50,6 @@ class MetadataMap {
grpc_metadata_array arr_;
std::multimap<grpc::string_ref, grpc::string_ref> map_;
};
} // namespace internal
} // namespace grpc

2
include/grpc++/impl/codegen/method_handler_impl.h
Unescape View File

@ -25,7 +25,6 @@
namespace grpc {
namespace internal {
/// A wrapper class of an application provided rpc method handler.
template <class ServiceType, class RequestType, class ResponseType>
class RpcMethodHandler : public MethodHandler {
@ -266,7 +265,6 @@ class UnknownMethodHandler : public MethodHandler {
}
};
} // namespace internal
} // namespace grpc
#endif // GRPCXX_IMPL_CODEGEN_METHOD_HANDLER_IMPL_H

3
include/grpc++/impl/codegen/rpc_method.h
Unescape View File

@ -24,7 +24,7 @@
#include <grpc++/impl/codegen/channel_interface.h>
namespace grpc {
namespace internal {
/// Descriptor of an RPC method
class RpcMethod {
public:
@ -55,7 +55,6 @@ class RpcMethod {
void* const channel_tag_;
};
} // namespace internal
} // namespace grpc
#endif // GRPCXX_IMPL_CODEGEN_RPC_METHOD_H

3
include/grpc++/impl/codegen/rpc_service_method.h
Unescape View File

@ -35,8 +35,8 @@ struct grpc_byte_buffer;
namespace grpc {
class ServerContext;
class StreamContextInterface;
namespace internal {
/// Base class for running an RPC handler.
class MethodHandler {
public:
@ -71,7 +71,6 @@ class RpcServiceMethod : public RpcMethod {
void* server_tag_;
std::unique_ptr<MethodHandler> handler_;
};
} // namespace internal
} // namespace grpc

27
include/grpc++/impl/codegen/server_context.h
Unescape View File

@ -55,6 +55,7 @@ class ServerWriter;
namespace internal {
template <class W, class R>
class ServerReaderWriterBody;
}
template <class ServiceType, class RequestType, class ResponseType>
class RpcMethodHandler;
template <class ServiceType, class RequestType, class ResponseType>
@ -64,11 +65,9 @@ class ServerStreamingHandler;
template <class ServiceType, class RequestType, class ResponseType>
class BidiStreamingHandler;
class UnknownMethodHandler;
template <class Streamer, bool WriteNeeded>
class TemplatedBidiStreamingHandler;
class Call;
} // namespace internal
class Call;
class CallOpBuffer;
class CompletionQueue;
class Server;
class ServerInterface;
@ -248,14 +247,14 @@ class ServerContext {
template <class W, class R>
friend class ::grpc::internal::ServerReaderWriterBody;
template <class ServiceType, class RequestType, class ResponseType>
friend class ::grpc::internal::RpcMethodHandler;
friend class RpcMethodHandler;
template <class ServiceType, class RequestType, class ResponseType>
friend class ::grpc::internal::ClientStreamingHandler;
friend class ClientStreamingHandler;
template <class ServiceType, class RequestType, class ResponseType>
friend class ::grpc::internal::ServerStreamingHandler;
friend class ServerStreamingHandler;
template <class Streamer, bool WriteNeeded>
friend class ::grpc::internal::TemplatedBidiStreamingHandler;
friend class ::grpc::internal::UnknownMethodHandler;
friend class TemplatedBidiStreamingHandler;
friend class UnknownMethodHandler;
friend class ::grpc::ClientContext;
/// Prevent copying.
@ -264,9 +263,9 @@ class ServerContext {
class CompletionOp;
void BeginCompletionOp(internal::Call* call);
void BeginCompletionOp(Call* call);
/// Return the tag queued by BeginCompletionOp()
internal::CompletionQueueTag* GetCompletionOpTag();
CompletionQueueTag* GetCompletionOpTag();
ServerContext(gpr_timespec deadline, grpc_metadata_array* arr);
@ -283,7 +282,7 @@ class ServerContext {
CompletionQueue* cq_;
bool sent_initial_metadata_;
mutable std::shared_ptr<const AuthContext> auth_context_;
internal::MetadataMap client_metadata_;
MetadataMap client_metadata_;
std::multimap<grpc::string, grpc::string> initial_metadata_;
std::multimap<grpc::string, grpc::string> trailing_metadata_;
@ -291,9 +290,7 @@ class ServerContext {
grpc_compression_level compression_level_;
grpc_compression_algorithm compression_algorithm_;
internal::CallOpSet<internal::CallOpSendInitialMetadata,
internal::CallOpSendMessage>
pending_ops_;
CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage> pending_ops_;
bool has_pending_ops_;
};

42
include/grpc++/impl/codegen/server_interface.h
Unescape View File

@ -30,21 +30,20 @@ namespace grpc {
class AsyncGenericService;
class Channel;
class GenericServerContext;
class RpcService;
class ServerAsyncStreamingInterface;
class ServerCompletionQueue;
class ServerContext;
class ServerCredentials;
class Service;
class ThreadPoolInterface;
extern CoreCodegenInterface* g_core_codegen_interface;
/// Models a gRPC server.
///
/// Servers are configured and started via \a grpc::ServerBuilder.
namespace internal {
class ServerAsyncStreamingInterface;
} // namespace internal
class ServerInterface : public internal::CallHook {
class ServerInterface : public CallHook {
public:
virtual ~ServerInterface() {}
@ -79,7 +78,7 @@ class ServerInterface : public internal::CallHook {
virtual void Wait() = 0;
protected:
friend class ::grpc::Service;
friend class Service;
/// Register a service. This call does not take ownership of the service.
/// The service must exist for the lifetime of the Server instance.
@ -117,13 +116,12 @@ class ServerInterface : public internal::CallHook {
virtual grpc_server* server() = 0;
virtual void PerformOpsOnCall(internal::CallOpSetInterface* ops,
internal::Call* call) = 0;
virtual void PerformOpsOnCall(CallOpSetInterface* ops, Call* call) = 0;
class BaseAsyncRequest : public internal::CompletionQueueTag {
class BaseAsyncRequest : public CompletionQueueTag {
public:
BaseAsyncRequest(ServerInterface* server, ServerContext* context,
internal::ServerAsyncStreamingInterface* stream,
ServerAsyncStreamingInterface* stream,
CompletionQueue* call_cq, void* tag,
bool delete_on_finalize);
virtual ~BaseAsyncRequest();
@ -133,7 +131,7 @@ class ServerInterface : public internal::CallHook {
protected:
ServerInterface* const server_;
ServerContext* const context_;
internal::ServerAsyncStreamingInterface* const stream_;
ServerAsyncStreamingInterface* const stream_;
CompletionQueue* const call_cq_;
void* const tag_;
const bool delete_on_finalize_;
@ -143,7 +141,7 @@ class ServerInterface : public internal::CallHook {
class RegisteredAsyncRequest : public BaseAsyncRequest {
public:
RegisteredAsyncRequest(ServerInterface* server, ServerContext* context,
internal::ServerAsyncStreamingInterface* stream,
ServerAsyncStreamingInterface* stream,
CompletionQueue* call_cq, void* tag);
// uses BaseAsyncRequest::FinalizeResult
@ -157,7 +155,7 @@ class ServerInterface : public internal::CallHook {
public:
NoPayloadAsyncRequest(void* registered_method, ServerInterface* server,
ServerContext* context,
internal::ServerAsyncStreamingInterface* stream,
ServerAsyncStreamingInterface* stream,
CompletionQueue* call_cq,
ServerCompletionQueue* notification_cq, void* tag)
: RegisteredAsyncRequest(server, context, stream, call_cq, tag) {
@ -172,7 +170,7 @@ class ServerInterface : public internal::CallHook {
public:
PayloadAsyncRequest(void* registered_method, ServerInterface* server,
ServerContext* context,
internal::ServerAsyncStreamingInterface* stream,
ServerAsyncStreamingInterface* stream,
CompletionQueue* call_cq,
ServerCompletionQueue* notification_cq, void* tag,
Message* request)
@ -214,7 +212,7 @@ class ServerInterface : public internal::CallHook {
void* const registered_method_;
ServerInterface* const server_;
ServerContext* const context_;
internal::ServerAsyncStreamingInterface* const stream_;
ServerAsyncStreamingInterface* const stream_;
CompletionQueue* const call_cq_;
ServerCompletionQueue* const notification_cq_;
void* const tag_;
@ -225,7 +223,7 @@ class ServerInterface : public internal::CallHook {
class GenericAsyncRequest : public BaseAsyncRequest {
public:
GenericAsyncRequest(ServerInterface* server, GenericServerContext* context,
internal::ServerAsyncStreamingInterface* stream,
ServerAsyncStreamingInterface* stream,
CompletionQueue* call_cq,
ServerCompletionQueue* notification_cq, void* tag,
bool delete_on_finalize);
@ -237,9 +235,8 @@ class ServerInterface : public internal::CallHook {
};
template <class Message>
void RequestAsyncCall(internal::RpcServiceMethod* method,
ServerContext* context,
internal::ServerAsyncStreamingInterface* stream,
void RequestAsyncCall(RpcServiceMethod* method, ServerContext* context,
ServerAsyncStreamingInterface* stream,
CompletionQueue* call_cq,
ServerCompletionQueue* notification_cq, void* tag,
Message* message) {
@ -249,9 +246,8 @@ class ServerInterface : public internal::CallHook {
message);
}
void RequestAsyncCall(internal::RpcServiceMethod* method,
ServerContext* context,
internal::ServerAsyncStreamingInterface* stream,
void RequestAsyncCall(RpcServiceMethod* method, ServerContext* context,
ServerAsyncStreamingInterface* stream,
CompletionQueue* call_cq,
ServerCompletionQueue* notification_cq, void* tag) {
GPR_CODEGEN_ASSERT(method);
@ -260,7 +256,7 @@ class ServerInterface : public internal::CallHook {
}
void RequestAsyncGenericCall(GenericServerContext* context,
internal::ServerAsyncStreamingInterface* stream,
ServerAsyncStreamingInterface* stream,
CompletionQueue* call_cq,
ServerCompletionQueue* notification_cq,
void* tag) {

46
include/grpc++/impl/codegen/service_type.h
Unescape View File

@ -28,14 +28,13 @@
namespace grpc {
class Call;
class CompletionQueue;
class Server;
class ServerInterface;
class ServerCompletionQueue;
class ServerContext;
namespace internal {
class Call;
class ServerAsyncStreamingInterface {
public:
virtual ~ServerAsyncStreamingInterface() {}
@ -49,10 +48,9 @@ class ServerAsyncStreamingInterface {
virtual void SendInitialMetadata(void* tag) = 0;
private:
friend class ::grpc::ServerInterface;
friend class ServerInterface;
virtual void BindCall(Call* call) = 0;
};
} // namespace internal
/// Desriptor of an RPC service and its various RPC methods
class Service {
@ -90,38 +88,40 @@ class Service {
protected:
template <class Message>
void RequestAsyncUnary(int index, ServerContext* context, Message* request,
internal::ServerAsyncStreamingInterface* stream,
ServerAsyncStreamingInterface* stream,
CompletionQueue* call_cq,
ServerCompletionQueue* notification_cq, void* tag) {
server_->RequestAsyncCall(methods_[index].get(), context, stream, call_cq,
notification_cq, tag, request);
}
void RequestAsyncClientStreaming(
int index, ServerContext* context,
internal::ServerAsyncStreamingInterface* stream, CompletionQueue* call_cq,
ServerCompletionQueue* notification_cq, void* tag) {
void RequestAsyncClientStreaming(int index, ServerContext* context,
ServerAsyncStreamingInterface* stream,
CompletionQueue* call_cq,
ServerCompletionQueue* notification_cq,
void* tag) {
server_->RequestAsyncCall(methods_[index].get(), context, stream, call_cq,
notification_cq, tag);
}
template <class Message>
void RequestAsyncServerStreaming(
int index, ServerContext* context, Message* request,
internal::ServerAsyncStreamingInterface* stream, CompletionQueue* call_cq,
ServerCompletionQueue* notification_cq, void* tag) {
void RequestAsyncServerStreaming(int index, ServerContext* context,
Message* request,
ServerAsyncStreamingInterface* stream,
CompletionQueue* call_cq,
ServerCompletionQueue* notification_cq,
void* tag) {
server_->RequestAsyncCall(methods_[index].get(), context, stream, call_cq,
notification_cq, tag, request);
}
void RequestAsyncBidiStreaming(
int index, ServerContext* context,
internal::ServerAsyncStreamingInterface* stream, CompletionQueue* call_cq,
ServerCompletionQueue* notification_cq, void* tag) {
void RequestAsyncBidiStreaming(int index, ServerContext* context,
ServerAsyncStreamingInterface* stream,
CompletionQueue* call_cq,
ServerCompletionQueue* notification_cq,
void* tag) {
server_->RequestAsyncCall(methods_[index].get(), context, stream, call_cq,
notification_cq, tag);
}
void AddMethod(internal::RpcServiceMethod* method) {
methods_.emplace_back(method);
}
void AddMethod(RpcServiceMethod* method) { methods_.emplace_back(method); }
void MarkMethodAsync(int index) {
GPR_CODEGEN_ASSERT(
@ -139,7 +139,7 @@ class Service {
methods_[index].reset();
}
void MarkMethodStreamed(int index, internal::MethodHandler* streamed_method) {
void MarkMethodStreamed(int index, MethodHandler* streamed_method) {
GPR_CODEGEN_ASSERT(methods_[index] && methods_[index]->handler() &&
"Cannot mark an async or generic method Streamed");
methods_[index]->SetHandler(streamed_method);
@ -148,14 +148,14 @@ class Service {
// case of BIDI_STREAMING that has 1 read and 1 write, in that order,
// and split server-side streaming is BIDI_STREAMING with 1 read and
// any number of writes, in that order.
methods_[index]->SetMethodType(internal::RpcMethod::BIDI_STREAMING);
methods_[index]->SetMethodType(::grpc::RpcMethod::BIDI_STREAMING);
}
private:
friend class Server;
friend class ServerInterface;
ServerInterface* server_;
std::vector<std::unique_ptr<internal::RpcServiceMethod>> methods_;
std::vector<std::unique_ptr<RpcServiceMethod>> methods_;
};
} // namespace grpc

398
include/grpc++/impl/codegen/sync_stream.h
Unescape View File

@ -30,7 +30,6 @@
namespace grpc {
namespace internal {
/// Common interface for all synchronous client side streaming.
class ClientStreamingInterface {
public:
@ -63,6 +62,20 @@ class ClientStreamingInterface {
virtual 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 {
@ -128,55 +141,16 @@ class WriterInterface {
}
};
} // 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;
};
/// 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 clinet 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;
};
/// 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> {
class ClientReaderInterface : public ClientStreamingInterface,
public 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 clinet 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;
};
/// Synchronous (blocking) client-side API for doing server-streaming RPCs,
@ -185,14 +159,28 @@ class ClientReaderWriterInterface : public internal::ClientStreamingInterface,
template <class R>
class ClientReader final : public ClientReaderInterface<R> {
public:
struct internal {
template <class W>
static ClientReader* Create(::grpc::ChannelInterface* channel,
const ::grpc::internal::RpcMethod& method,
ClientContext* context, const W& request) {
return new ClientReader(channel, method, context, request);
}
};
/// 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(ChannelInterface* channel, const RpcMethod& method,
ClientContext* context, const W& request)
: context_(context),
cq_(grpc_completion_queue_attributes{
GRPC_CQ_CURRENT_VERSION, GRPC_CQ_PLUCK,
GRPC_CQ_DEFAULT_POLLING}), // Pluckable cq
call_(channel->CreateCall(method, context, &cq_)) {
CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage,
CallOpClientSendClose>
ops;
ops.SendInitialMetadata(context->send_initial_metadata_,
context->initial_metadata_flags());
// TODO(ctiller): don't assert
GPR_CODEGEN_ASSERT(ops.SendMessage(request).ok());
ops.ClientSendClose();
call_.PerformOps(&ops);
cq_.Pluck(&ops);
}
/// See the \a ClientStreamingInterface.WaitForInitialMetadata method for
/// semantics.
@ -204,8 +192,7 @@ class ClientReader final : public ClientReaderInterface<R> {
void WaitForInitialMetadata() override {
GPR_CODEGEN_ASSERT(!context_->initial_metadata_received_);
::grpc::internal::CallOpSet<::grpc::internal::CallOpRecvInitialMetadata>
ops;
CallOpSet<CallOpRecvInitialMetadata> ops;
ops.RecvInitialMetadata(context_);
call_.PerformOps(&ops);
cq_.Pluck(&ops); /// status ignored
@ -222,9 +209,7 @@ class ClientReader final : public ClientReaderInterface<R> {
/// 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;
CallOpSet<CallOpRecvInitialMetadata, CallOpRecvMessage<R>> ops;
if (!context_->initial_metadata_received_) {
ops.RecvInitialMetadata(context_);
}
@ -239,7 +224,7 @@ class ClientReader final : public ClientReaderInterface<R> {
/// The \a ClientContext associated with this call is updated with
/// possible metadata received from the server.
Status Finish() override {
::grpc::internal::CallOpSet<::grpc::internal::CallOpClientRecvStatus> ops;
CallOpSet<CallOpClientRecvStatus> ops;
Status status;
ops.ClientRecvStatus(context_, &status);
call_.PerformOps(&ops);
@ -250,48 +235,53 @@ class ClientReader final : public ClientReaderInterface<R> {
private:
ClientContext* context_;
CompletionQueue cq_;
::grpc::internal::Call call_;
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,
ClientContext* context, const W& request)
: context_(context),
cq_(grpc_completion_queue_attributes{
GRPC_CQ_CURRENT_VERSION, GRPC_CQ_PLUCK,
GRPC_CQ_DEFAULT_POLLING}), // 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.SendMessage(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 ClientStreamingInterface,
public WriterInterface<W> {
public:
/// Half close writing from the client. (signal that the stream of messages
/// coming from the clinet 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;
};
/// 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 final : public ClientWriterInterface<W> {
class ClientWriter : public ClientWriterInterface<W> {
public:
struct internal {
template <class R>
static ClientWriter* Create(::grpc::ChannelInterface* channel,
const ::grpc::internal::RpcMethod& method,
ClientContext* context, R* response) {
return new ClientWriter(channel, method, context, response);
/// 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(ChannelInterface* channel, const RpcMethod& method,
ClientContext* context, R* response)
: context_(context),
cq_(grpc_completion_queue_attributes{
GRPC_CQ_CURRENT_VERSION, GRPC_CQ_PLUCK,
GRPC_CQ_DEFAULT_POLLING}), // Pluckable cq
call_(channel->CreateCall(method, context, &cq_)) {
finish_ops_.RecvMessage(response);
finish_ops_.AllowNoMessage();
if (!context_->initial_metadata_corked_) {
CallOpSet<CallOpSendInitialMetadata> ops;
ops.SendInitialMetadata(context->send_initial_metadata_,
context->initial_metadata_flags());
call_.PerformOps(&ops);
cq_.Pluck(&ops);
}
};
}
/// See the \a ClientStreamingInterface.WaitForInitialMetadata method for
/// semantics.
@ -302,8 +292,7 @@ class ClientWriter final : public ClientWriterInterface<W> {
void WaitForInitialMetadata() {
GPR_CODEGEN_ASSERT(!context_->initial_metadata_received_);
::grpc::internal::CallOpSet<::grpc::internal::CallOpRecvInitialMetadata>
ops;
CallOpSet<CallOpRecvInitialMetadata> ops;
ops.RecvInitialMetadata(context_);
call_.PerformOps(&ops);
cq_.Pluck(&ops); // status ignored
@ -315,11 +304,10 @@ class ClientWriter final : public ClientWriterInterface<W> {
/// Side effect:
/// Also sends initial metadata if not already sent (using the
/// \a ClientContext associated with this call).
using ::grpc::internal::WriterInterface<W>::Write;
using WriterInterface<W>::Write;
bool Write(const W& msg, WriteOptions options) override {
::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata,
::grpc::internal::CallOpSendMessage,
::grpc::internal::CallOpClientSendClose>
CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage,
CallOpClientSendClose>
ops;
if (options.is_last_message()) {
@ -340,7 +328,7 @@ class ClientWriter final : public ClientWriterInterface<W> {
}
bool WritesDone() override {
::grpc::internal::CallOpSet<::grpc::internal::CallOpClientSendClose> ops;
CallOpSet<CallOpClientSendClose> ops;
ops.ClientSendClose();
call_.PerformOps(&ops);
return cq_.Pluck(&ops);
@ -365,55 +353,61 @@ class ClientWriter final : public ClientWriterInterface<W> {
private:
ClientContext* context_;
::grpc::internal::CallOpSet<::grpc::internal::CallOpRecvInitialMetadata,
::grpc::internal::CallOpGenericRecvMessage,
::grpc::internal::CallOpClientRecvStatus>
CallOpSet<CallOpRecvInitialMetadata, CallOpGenericRecvMessage,
CallOpClientRecvStatus>
finish_ops_;
CompletionQueue cq_;
::grpc::internal::Call call_;
Call call_;
};
/// 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,
ClientContext* context, R* response)
/// 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 ClientStreamingInterface,
public WriterInterface<W>,
public 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 clinet 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;
};
/// 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 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(ChannelInterface* channel, const RpcMethod& method,
ClientContext* context)
: context_(context),
cq_(grpc_completion_queue_attributes{
GRPC_CQ_CURRENT_VERSION, GRPC_CQ_PLUCK,
GRPC_CQ_DEFAULT_POLLING}), // 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;
CallOpSet<CallOpSendInitialMetadata> ops;
ops.SendInitialMetadata(context->send_initial_metadata_,
context->initial_metadata_flags());
call_.PerformOps(&ops);
cq_.Pluck(&ops);
}
}
};
/// 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:
struct internal {
static ClientReaderWriter* Create(::grpc::ChannelInterface* channel,
const ::grpc::internal::RpcMethod& method,
ClientContext* context) {
return new ClientReaderWriter(channel, method, context);
}
};
/// Block waiting to read initial metadata from the server.
/// This call is optional, but if it is used, it cannot be used concurrently
@ -424,8 +418,7 @@ class ClientReaderWriter final : public ClientReaderWriterInterface<W, R> {
void WaitForInitialMetadata() override {
GPR_CODEGEN_ASSERT(!context_->initial_metadata_received_);
::grpc::internal::CallOpSet<::grpc::internal::CallOpRecvInitialMetadata>
ops;
CallOpSet<CallOpRecvInitialMetadata> ops;
ops.RecvInitialMetadata(context_);
call_.PerformOps(&ops);
cq_.Pluck(&ops); // status ignored
@ -441,9 +434,7 @@ class ClientReaderWriter final : public ClientReaderWriterInterface<W, R> {
/// 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;
CallOpSet<CallOpRecvInitialMetadata, CallOpRecvMessage<R>> ops;
if (!context_->initial_metadata_received_) {
ops.RecvInitialMetadata(context_);
}
@ -457,11 +448,10 @@ class ClientReaderWriter final : public ClientReaderWriterInterface<W, R> {
/// Side effect:
/// Also sends initial metadata if not already sent (using the
/// \a ClientContext associated with this call to fill in values).
using ::grpc::internal::WriterInterface<W>::Write;
using WriterInterface<W>::Write;
bool Write(const W& msg, WriteOptions options) override {
::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata,
::grpc::internal::CallOpSendMessage,
::grpc::internal::CallOpClientSendClose>
CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage,
CallOpClientSendClose>
ops;
if (options.is_last_message()) {
@ -482,7 +472,7 @@ class ClientReaderWriter final : public ClientReaderWriterInterface<W, R> {
}
bool WritesDone() override {
::grpc::internal::CallOpSet<::grpc::internal::CallOpClientSendClose> ops;
CallOpSet<CallOpClientSendClose> ops;
ops.ClientSendClose();
call_.PerformOps(&ops);
return cq_.Pluck(&ops);
@ -494,9 +484,7 @@ class ClientReaderWriter final : public ClientReaderWriterInterface<W, R> {
/// - the \a ClientContext associated with this call is updated with
/// possible trailing metadata sent from the server.
Status Finish() override {
::grpc::internal::CallOpSet<::grpc::internal::CallOpRecvInitialMetadata,
::grpc::internal::CallOpClientRecvStatus>
ops;
CallOpSet<CallOpRecvInitialMetadata, CallOpClientRecvStatus> ops;
if (!context_->initial_metadata_received_) {
ops.RecvInitialMetadata(context_);
}
@ -510,61 +498,13 @@ class ClientReaderWriter final : public ClientReaderWriterInterface<W, R> {
private:
ClientContext* context_;
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,
ClientContext* context)
: context_(context),
cq_(grpc_completion_queue_attributes{
GRPC_CQ_CURRENT_VERSION, GRPC_CQ_PLUCK,
GRPC_CQ_DEFAULT_POLLING}), // 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);
}
}
Call call_;
};
namespace internal {
/// 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;
};
} // namespace internal
/// Server-side interface for streaming reads of message of type \a R.
template <class R>
class ServerReaderInterface : public internal::ServerStreamingInterface,
public internal::ReaderInterface<R> {};
/// Server-side interface for streaming writes of message of type \a W.
template <class W>
class ServerWriterInterface : public internal::ServerStreamingInterface,
public internal::WriterInterface<W> {};
/// 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> {};
class ServerReaderInterface : public ServerStreamingInterface,
public ReaderInterface<R> {};
/// Synchronous (blocking) server-side API for doing client-streaming RPCs,
/// where the incoming message stream coming from the client has messages of
@ -572,13 +512,15 @@ class ServerReaderWriterInterface : public internal::ServerStreamingInterface,
template <class R>
class ServerReader final : public ServerReaderInterface<R> {
public:
ServerReader(Call* call, ServerContext* ctx) : call_(call), ctx_(ctx) {}
/// 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_);
internal::CallOpSet<internal::CallOpSendInitialMetadata> ops;
CallOpSet<CallOpSendInitialMetadata> ops;
ops.SendInitialMetadata(ctx_->initial_metadata_,
ctx_->initial_metadata_flags());
if (ctx_->compression_level_set()) {
@ -595,29 +537,30 @@ class ServerReader final : public ServerReaderInterface<R> {
}
bool Read(R* msg) override {
internal::CallOpSet<internal::CallOpRecvMessage<R>> ops;
CallOpSet<CallOpRecvMessage<R>> ops;
ops.RecvMessage(msg);
call_->PerformOps(&ops);
return call_->cq()->Pluck(&ops) && ops.got_message;
}
private:
internal::Call* const call_;
Call* const call_;
ServerContext* const ctx_;
template <class ServiceType, class RequestType, class ResponseType>
friend class internal::ClientStreamingHandler;
ServerReader(internal::Call* call, ServerContext* ctx)
: call_(call), ctx_(ctx) {}
};
/// Server-side interface for streaming writes of message of type \a W.
template <class W>
class ServerWriterInterface : public ServerStreamingInterface,
public 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:
ServerWriter(Call* call, ServerContext* ctx) : call_(call), ctx_(ctx) {}
/// See the \a ServerStreamingInterface.SendInitialMetadata method
/// for semantics.
/// Note that initial metadata will be affected by the
@ -625,7 +568,7 @@ class ServerWriter final : public ServerWriterInterface<W> {
void SendInitialMetadata() override {
GPR_CODEGEN_ASSERT(!ctx_->sent_initial_metadata_);
internal::CallOpSet<internal::CallOpSendInitialMetadata> ops;
CallOpSet<CallOpSendInitialMetadata> ops;
ops.SendInitialMetadata(ctx_->initial_metadata_,
ctx_->initial_metadata_flags());
if (ctx_->compression_level_set()) {
@ -641,12 +584,11 @@ class ServerWriter final : public ServerWriterInterface<W> {
/// 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;
using WriterInterface<W>::Write;
bool Write(const W& msg, WriteOptions options) override {
if (options.is_last_message()) {
options.set_buffer_hint();
}
if (!ctx_->pending_ops_.SendMessage(msg, options).ok()) {
return false;
}
@ -671,16 +613,16 @@ class ServerWriter final : public ServerWriterInterface<W> {
}
private:
internal::Call* const call_;
Call* const call_;
ServerContext* const ctx_;
template <class ServiceType, class RequestType, class ResponseType>
friend class internal::ServerStreamingHandler;
ServerWriter(internal::Call* call, ServerContext* ctx)
: call_(call), ctx_(ctx) {}
};
/// Server-side interface for bi-directional streaming.
template <class W, class R>
class ServerReaderWriterInterface : public ServerStreamingInterface,
public WriterInterface<W>,
public ReaderInterface<R> {};
/// Actual implementation of bi-directional streaming
namespace internal {
template <class W, class R>
@ -746,7 +688,6 @@ class ServerReaderWriterBody final {
Call* const call_;
ServerContext* const ctx_;
};
} // namespace internal
/// Synchronous (blocking) server-side API for a bidirectional
@ -756,6 +697,8 @@ class ServerReaderWriterBody final {
template <class W, class R>
class ServerReaderWriter final : public ServerReaderWriterInterface<W, R> {
public:
ServerReaderWriter(Call* call, ServerContext* ctx) : body_(call, ctx) {}
/// See the \a ServerStreamingInterface.SendInitialMetadata method
/// for semantics. Note that initial metadata will be affected by the
/// \a ServerContext associated with this call.
@ -772,18 +715,13 @@ class ServerReaderWriter final : public ServerReaderWriterInterface<W, R> {
/// Side effect:
/// Also sends initial metadata if not already sent (using the \a
/// ServerContext associated with this call).
using internal::WriterInterface<W>::Write;
using WriterInterface<W>::Write;
bool Write(const W& msg, WriteOptions options) override {
return body_.Write(msg, options);
}
private:
internal::ServerReaderWriterBody<W, R> body_;
friend class internal::TemplatedBidiStreamingHandler<ServerReaderWriter<W, R>,
false>;
ServerReaderWriter(internal::Call* call, ServerContext* ctx)
: body_(call, ctx) {}
};
/// A class to represent a flow-controlled unary call. This is something
@ -798,6 +736,9 @@ template <class RequestType, class ResponseType>
class ServerUnaryStreamer final
: public ServerReaderWriterInterface<ResponseType, RequestType> {
public:
ServerUnaryStreamer(Call* call, ServerContext* ctx)
: body_(call, ctx), read_done_(false), write_done_(false) {}
/// Block to send initial metadata to client.
/// Implicit input parameter:
/// - the \a ServerContext associated with this call will be used for
@ -834,7 +775,7 @@ class ServerUnaryStreamer final
/// \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;
using WriterInterface<ResponseType>::Write;
bool Write(const ResponseType& response, WriteOptions options) override {
if (write_done_ || !read_done_) {
return false;
@ -847,11 +788,6 @@ class ServerUnaryStreamer final
internal::ServerReaderWriterBody<ResponseType, RequestType> body_;
bool read_done_;
bool write_done_;
friend class internal::TemplatedBidiStreamingHandler<
ServerUnaryStreamer<RequestType, ResponseType>, true>;
ServerUnaryStreamer(internal::Call* call, ServerContext* ctx)
: body_(call, ctx), read_done_(false), write_done_(false) {}
};
/// A class to represent a flow-controlled server-side streaming call.
@ -863,6 +799,9 @@ template <class RequestType, class ResponseType>
class ServerSplitStreamer final
: public ServerReaderWriterInterface<ResponseType, RequestType> {
public:
ServerSplitStreamer(Call* call, ServerContext* ctx)
: body_(call, ctx), read_done_(false) {}
/// Block to send initial metadata to client.
/// Implicit input parameter:
/// - the \a ServerContext associated with this call will be used for
@ -899,7 +838,7 @@ class ServerSplitStreamer final
/// \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;
using WriterInterface<ResponseType>::Write;
bool Write(const ResponseType& response, WriteOptions options) override {
return read_done_ && body_.Write(response, options);
}
@ -907,11 +846,6 @@ class ServerSplitStreamer final
private:
internal::ServerReaderWriterBody<ResponseType, RequestType> body_;
bool read_done_;
friend class internal::TemplatedBidiStreamingHandler<
ServerSplitStreamer<RequestType, ResponseType>, false>;
ServerSplitStreamer(internal::Call* call, ServerContext* ctx)
: body_(call, ctx), read_done_(false) {}
};
} // namespace grpc

6
include/grpc++/impl/codegen/time.h
Unescape View File

@ -19,8 +19,6 @@
#ifndef GRPCXX_IMPL_CODEGEN_TIME_H
#define GRPCXX_IMPL_CODEGEN_TIME_H
#include <chrono>
#include <grpc++/impl/codegen/config.h>
#include <grpc/impl/codegen/grpc_types.h>
@ -61,6 +59,10 @@ class TimePoint<gpr_timespec> {
} // namespace grpc
#include <chrono>
#include <grpc/impl/codegen/grpc_types.h>
namespace grpc {
// from and to should be absolute time.

3
include/grpc++/server.h
Unescape View File

@ -175,8 +175,7 @@ class Server final : public ServerInterface, private GrpcLibraryCodegen {
/// \param num_cqs How many completion queues does \a cqs hold.
void Start(ServerCompletionQueue** cqs, size_t num_cqs) override;
void PerformOpsOnCall(internal::CallOpSetInterface* ops,
internal::Call* call) override;
void PerformOpsOnCall(CallOpSetInterface* ops, Call* call) override;
void ShutdownInternal(gpr_timespec deadline) override;

1
include/grpc++/server_builder.h
Unescape View File

@ -40,6 +40,7 @@ namespace grpc {
class AsyncGenericService;
class ResourceQuota;
class CompletionQueue;
class RpcService;
class Server;
class ServerCompletionQueue;
class ServerCredentials;

199
src/compiler/cpp_generator.cc
Unescape View File

@ -140,6 +140,7 @@ grpc::string GetHeaderIncludes(grpc_generator::File *file,
printer->Print(vars, "namespace grpc {\n");
printer->Print(vars, "class CompletionQueue;\n");
printer->Print(vars, "class Channel;\n");
printer->Print(vars, "class RpcService;\n");
printer->Print(vars, "class ServerCompletionQueue;\n");
printer->Print(vars, "class ServerContext;\n");
printer->Print(vars, "} // namespace grpc\n\n");
@ -186,21 +187,19 @@ void PrintHeaderClientMethodInterfaces(
} else if (ClientOnlyStreaming(method)) {
printer->Print(
*vars,
"std::unique_ptr< ::grpc::ClientWriterInterface< "
"$Request$>>"
"std::unique_ptr< ::grpc::ClientWriterInterface< $Request$>>"
" $Method$("
"::grpc::ClientContext* context, $Response$* response) {\n");
printer->Indent();
printer->Print(*vars,
"return std::unique_ptr< "
"::grpc::ClientWriterInterface< $Request$>>"
"($Method$Raw(context, response));\n");
printer->Print(
*vars,
"return std::unique_ptr< ::grpc::ClientWriterInterface< $Request$>>"
"($Method$Raw(context, response));\n");
printer->Outdent();
printer->Print("}\n");
printer->Print(
*vars,
"std::unique_ptr< ::grpc::ClientAsyncWriterInterface< "
"$Request$>>"
"std::unique_ptr< ::grpc::ClientAsyncWriterInterface< $Request$>>"
" Async$Method$(::grpc::ClientContext* context, $Response$* "
"response, "
"::grpc::CompletionQueue* cq, void* tag) {\n");
@ -214,21 +213,19 @@ void PrintHeaderClientMethodInterfaces(
} else if (ServerOnlyStreaming(method)) {
printer->Print(
*vars,
"std::unique_ptr< ::grpc::ClientReaderInterface< "
"$Response$>>"
"std::unique_ptr< ::grpc::ClientReaderInterface< $Response$>>"
" $Method$(::grpc::ClientContext* context, const $Request$& request)"
" {\n");
printer->Indent();
printer->Print(*vars,
"return std::unique_ptr< "
"::grpc::ClientReaderInterface< $Response$>>"
"($Method$Raw(context, request));\n");
printer->Print(
*vars,
"return std::unique_ptr< ::grpc::ClientReaderInterface< $Response$>>"
"($Method$Raw(context, request));\n");
printer->Outdent();
printer->Print("}\n");
printer->Print(
*vars,
"std::unique_ptr< ::grpc::ClientAsyncReaderInterface< "
"$Response$>> "
"std::unique_ptr< ::grpc::ClientAsyncReaderInterface< $Response$>> "
"Async$Method$("
"::grpc::ClientContext* context, const $Request$& request, "
"::grpc::CompletionQueue* cq, void* tag) {\n");
@ -245,37 +242,36 @@ void PrintHeaderClientMethodInterfaces(
"$Request$, $Response$>> "
"$Method$(::grpc::ClientContext* context) {\n");
printer->Indent();
printer->Print(*vars,
"return std::unique_ptr< "
"::grpc::ClientReaderWriterInterface< "
"$Request$, $Response$>>("
"$Method$Raw(context));\n");
printer->Print(
*vars,
"return std::unique_ptr< "
"::grpc::ClientReaderWriterInterface< $Request$, $Response$>>("
"$Method$Raw(context));\n");
printer->Outdent();
printer->Print("}\n");
printer->Print(*vars,
"std::unique_ptr< "
"::grpc::ClientAsyncReaderWriterInterface< "
"$Request$, $Response$>> "
"Async$Method$(::grpc::ClientContext* context, "
"::grpc::CompletionQueue* cq, void* tag) {\n");
printer->Print(
*vars,
"std::unique_ptr< "
"::grpc::ClientAsyncReaderWriterInterface< $Request$, $Response$>> "
"Async$Method$(::grpc::ClientContext* context, "
"::grpc::CompletionQueue* cq, void* tag) {\n");
printer->Indent();
printer->Print(*vars,
"return std::unique_ptr< "
"::grpc::ClientAsyncReaderWriterInterface< "
"$Request$, $Response$>>("
"Async$Method$Raw(context, cq, tag));\n");
printer->Print(
*vars,
"return std::unique_ptr< "
"::grpc::ClientAsyncReaderWriterInterface< $Request$, $Response$>>("
"Async$Method$Raw(context, cq, tag));\n");
printer->Outdent();
printer->Print("}\n");
}
} else {
if (method->NoStreaming()) {
printer->Print(*vars,
"virtual "
"::grpc::ClientAsyncResponseReaderInterface< "
"$Response$>* "
"Async$Method$Raw(::grpc::ClientContext* context, "
"const $Request$& request, "
"::grpc::CompletionQueue* cq) = 0;\n");
printer->Print(
*vars,
"virtual ::grpc::ClientAsyncResponseReaderInterface< $Response$>* "
"Async$Method$Raw(::grpc::ClientContext* context, "
"const $Request$& request, "
"::grpc::CompletionQueue* cq) = 0;\n");
} else if (ClientOnlyStreaming(method)) {
printer->Print(
*vars,
@ -290,8 +286,7 @@ void PrintHeaderClientMethodInterfaces(
} else if (ServerOnlyStreaming(method)) {
printer->Print(
*vars,
"virtual ::grpc::ClientReaderInterface< $Response$>* "
"$Method$Raw("
"virtual ::grpc::ClientReaderInterface< $Response$>* $Method$Raw("
"::grpc::ClientContext* context, const $Request$& request) = 0;\n");
printer->Print(
*vars,
@ -456,8 +451,7 @@ void PrintHeaderClientMethodData(grpc_generator::Printer *printer,
const grpc_generator::Method *method,
std::map<grpc::string, grpc::string> *vars) {
(*vars)["Method"] = method->name();
printer->Print(*vars,
"const ::grpc::internal::RpcMethod rpcmethod_$Method$_;\n");
printer->Print(*vars, "const ::grpc::RpcMethod rpcmethod_$Method$_;\n");
}
void PrintHeaderServerMethodSync(grpc_generator::Printer *printer,
@ -629,7 +623,7 @@ void PrintHeaderServerMethodStreamedUnary(
printer->Print(*vars,
"WithStreamedUnaryMethod_$Method$() {\n"
" ::grpc::Service::MarkMethodStreamed($Idx$,\n"
" new ::grpc::internal::StreamedUnaryHandler< $Request$, "
" new ::grpc::StreamedUnaryHandler< $Request$, "
"$Response$>(std::bind"
"(&WithStreamedUnaryMethod_$Method$<BaseClass>::"
"Streamed$Method$, this, std::placeholders::_1, "
@ -677,16 +671,15 @@ void PrintHeaderServerMethodSplitStreaming(
"{}\n");
printer->Print(" public:\n");
printer->Indent();
printer->Print(
*vars,
"WithSplitStreamingMethod_$Method$() {\n"
" ::grpc::Service::MarkMethodStreamed($Idx$,\n"
" new ::grpc::internal::SplitServerStreamingHandler< $Request$, "
"$Response$>(std::bind"
"(&WithSplitStreamingMethod_$Method$<BaseClass>::"
"Streamed$Method$, this, std::placeholders::_1, "
"std::placeholders::_2)));\n"
"}\n");
printer->Print(*vars,
"WithSplitStreamingMethod_$Method$() {\n"
" ::grpc::Service::MarkMethodStreamed($Idx$,\n"
" new ::grpc::SplitServerStreamingHandler< $Request$, "
"$Response$>(std::bind"
"(&WithSplitStreamingMethod_$Method$<BaseClass>::"
"Streamed$Method$, this, std::placeholders::_1, "
"std::placeholders::_2)));\n"
"}\n");
printer->Print(*vars,
"~WithSplitStreamingMethod_$Method$() override {\n"
" BaseClassMustBeDerivedFromService(this);\n"
@ -826,8 +819,7 @@ void PrintHeaderService(grpc_generator::Printer *printer,
" {\n public:\n");
printer->Indent();
printer->Print(
"Stub(const std::shared_ptr< ::grpc::ChannelInterface>& "
"channel);\n");
"Stub(const std::shared_ptr< ::grpc::ChannelInterface>& channel);\n");
for (int i = 0; i < service->method_count(); ++i) {
PrintHeaderClientMethod(printer, service->method(i).get(), vars, true);
}
@ -1090,12 +1082,11 @@ void PrintSourceClientMethod(grpc_generator::Printer *printer,
"::grpc::Status $ns$$Service$::Stub::$Method$("
"::grpc::ClientContext* context, "
"const $Request$& request, $Response$* response) {\n");
printer->Print(
*vars,
" return ::grpc::internal::BlockingUnaryCall(channel_.get(), "
"rpcmethod_$Method$_, "
"context, request, response);\n"
"}\n\n");
printer->Print(*vars,
" return ::grpc::BlockingUnaryCall(channel_.get(), "
"rpcmethod_$Method$_, "
"context, request, response);\n"
"}\n\n");
printer->Print(
*vars,
"::grpc::ClientAsyncResponseReader< $Response$>* "
@ -1104,8 +1095,7 @@ void PrintSourceClientMethod(grpc_generator::Printer *printer,
"::grpc::CompletionQueue* cq) {\n");
printer->Print(*vars,
" return "
"::grpc::ClientAsyncResponseReader< $Response$>::"
"internal::Create("
"::grpc::ClientAsyncResponseReader< $Response$>::Create("
"channel_.get(), cq, "
"rpcmethod_$Method$_, "
"context, request);\n"
@ -1115,21 +1105,19 @@ void PrintSourceClientMethod(grpc_generator::Printer *printer,
"::grpc::ClientWriter< $Request$>* "
"$ns$$Service$::Stub::$Method$Raw("
"::grpc::ClientContext* context, $Response$* response) {\n");
printer->Print(
*vars,
" return ::grpc::ClientWriter< $Request$>::internal::Create("
"channel_.get(), "
"rpcmethod_$Method$_, "
"context, response);\n"
"}\n\n");
printer->Print(*vars,
" return new ::grpc::ClientWriter< $Request$>("
"channel_.get(), "
"rpcmethod_$Method$_, "
"context, response);\n"
"}\n\n");
printer->Print(*vars,
"::grpc::ClientAsyncWriter< $Request$>* "
"$ns$$Service$::Stub::Async$Method$Raw("
"::grpc::ClientContext* context, $Response$* response, "
"::grpc::CompletionQueue* cq, void* tag) {\n");
printer->Print(*vars,
" return ::grpc::ClientAsyncWriter< $Request$>::"
"internal::Create("
" return ::grpc::ClientAsyncWriter< $Request$>::Create("
"channel_.get(), cq, "
"rpcmethod_$Method$_, "
"context, response, tag);\n"
@ -1140,21 +1128,19 @@ void PrintSourceClientMethod(grpc_generator::Printer *printer,
"::grpc::ClientReader< $Response$>* "
"$ns$$Service$::Stub::$Method$Raw("
"::grpc::ClientContext* context, const $Request$& request) {\n");
printer->Print(
*vars,
" return ::grpc::ClientReader< $Response$>::internal::Create("
"channel_.get(), "
"rpcmethod_$Method$_, "
"context, request);\n"
"}\n\n");
printer->Print(*vars,
" return new ::grpc::ClientReader< $Response$>("
"channel_.get(), "
"rpcmethod_$Method$_, "
"context, request);\n"
"}\n\n");
printer->Print(*vars,
"::grpc::ClientAsyncReader< $Response$>* "
"$ns$$Service$::Stub::Async$Method$Raw("
"::grpc::ClientContext* context, const $Request$& request, "
"::grpc::CompletionQueue* cq, void* tag) {\n");
printer->Print(*vars,
" return ::grpc::ClientAsyncReader< $Response$>::"
"internal::Create("
" return ::grpc::ClientAsyncReader< $Response$>::Create("
"channel_.get(), cq, "
"rpcmethod_$Method$_, "
"context, request, tag);\n"
@ -1165,8 +1151,8 @@ void PrintSourceClientMethod(grpc_generator::Printer *printer,
"::grpc::ClientReaderWriter< $Request$, $Response$>* "
"$ns$$Service$::Stub::$Method$Raw(::grpc::ClientContext* context) {\n");
printer->Print(*vars,
" return ::grpc::ClientReaderWriter< "
"$Request$, $Response$>::internal::Create("
" return new ::grpc::ClientReaderWriter< "
"$Request$, $Response$>("
"channel_.get(), "
"rpcmethod_$Method$_, "
"context);\n"
@ -1176,14 +1162,14 @@ void PrintSourceClientMethod(grpc_generator::Printer *printer,
"::grpc::ClientAsyncReaderWriter< $Request$, $Response$>* "
"$ns$$Service$::Stub::Async$Method$Raw(::grpc::ClientContext* context, "
"::grpc::CompletionQueue* cq, void* tag) {\n");
printer->Print(*vars,
" return "
"::grpc::ClientAsyncReaderWriter< $Request$, $Response$>::"
"internal::Create("
"channel_.get(), cq, "
"rpcmethod_$Method$_, "
"context, tag);\n"
"}\n\n");
printer->Print(
*vars,
" return "
"::grpc::ClientAsyncReaderWriter< $Request$, $Response$>::Create("
"channel_.get(), cq, "
"rpcmethod_$Method$_, "
"context, tag);\n"
"}\n\n");
}
}
@ -1293,7 +1279,7 @@ void PrintSourceService(grpc_generator::Printer *printer,
printer->Print(*vars,
", rpcmethod_$Method$_("
"$prefix$$Service$_method_names[$Idx$], "
"::grpc::internal::RpcMethod::$StreamingType$, "
"::grpc::RpcMethod::$StreamingType$, "
"channel"
")\n");
}
@ -1316,38 +1302,38 @@ void PrintSourceService(grpc_generator::Printer *printer,
if (method->NoStreaming()) {
printer->Print(
*vars,
"AddMethod(new ::grpc::internal::RpcServiceMethod(\n"
"AddMethod(new ::grpc::RpcServiceMethod(\n"
" $prefix$$Service$_method_names[$Idx$],\n"
" ::grpc::internal::RpcMethod::NORMAL_RPC,\n"
" new ::grpc::internal::RpcMethodHandler< $ns$$Service$::Service, "
" ::grpc::RpcMethod::NORMAL_RPC,\n"
" new ::grpc::RpcMethodHandler< $ns$$Service$::Service, "
"$Request$, "
"$Response$>(\n"
" std::mem_fn(&$ns$$Service$::Service::$Method$), this)));\n");
} else if (ClientOnlyStreaming(method.get())) {
printer->Print(
*vars,
"AddMethod(new ::grpc::internal::RpcServiceMethod(\n"
"AddMethod(new ::grpc::RpcServiceMethod(\n"
" $prefix$$Service$_method_names[$Idx$],\n"
" ::grpc::internal::RpcMethod::CLIENT_STREAMING,\n"
" new ::grpc::internal::ClientStreamingHandler< "
" ::grpc::RpcMethod::CLIENT_STREAMING,\n"
" new ::grpc::ClientStreamingHandler< "
"$ns$$Service$::Service, $Request$, $Response$>(\n"
" std::mem_fn(&$ns$$Service$::Service::$Method$), this)));\n");
} else if (ServerOnlyStreaming(method.get())) {
printer->Print(
*vars,
"AddMethod(new ::grpc::internal::RpcServiceMethod(\n"
"AddMethod(new ::grpc::RpcServiceMethod(\n"
" $prefix$$Service$_method_names[$Idx$],\n"
" ::grpc::internal::RpcMethod::SERVER_STREAMING,\n"
" new ::grpc::internal::ServerStreamingHandler< "
" ::grpc::RpcMethod::SERVER_STREAMING,\n"
" new ::grpc::ServerStreamingHandler< "
"$ns$$Service$::Service, $Request$, $Response$>(\n"
" std::mem_fn(&$ns$$Service$::Service::$Method$), this)));\n");
} else if (method->BidiStreaming()) {
printer->Print(
*vars,
"AddMethod(new ::grpc::internal::RpcServiceMethod(\n"
"AddMethod(new ::grpc::RpcServiceMethod(\n"
" $prefix$$Service$_method_names[$Idx$],\n"
" ::grpc::internal::RpcMethod::BIDI_STREAMING,\n"
" new ::grpc::internal::BidiStreamingHandler< "
" ::grpc::RpcMethod::BIDI_STREAMING,\n"
" new ::grpc::BidiStreamingHandler< "
"$ns$$Service$::Service, $Request$, $Response$>(\n"
" std::mem_fn(&$ns$$Service$::Service::$Method$), this)));\n");
}
@ -1515,8 +1501,7 @@ void PrintMockClientMethods(grpc_generator::Printer *printer,
printer->Print(
*vars,
"MOCK_METHOD3(Async$Method$Raw, "
"::grpc::ClientAsyncReaderWriterInterface<$Request$, "
"$Response$>*"
"::grpc::ClientAsyncReaderWriterInterface<$Request$, $Response$>*"
"(::grpc::ClientContext* context, ::grpc::CompletionQueue* cq, "
"void* tag));\n");
}

12
src/cpp/client/channel_cc.cc
Unescape View File

@ -76,9 +76,8 @@ grpc::string Channel::GetServiceConfigJSON() const {
&channel_info.service_config_json);
}
internal::Call Channel::CreateCall(const internal::RpcMethod& method,
ClientContext* context,
CompletionQueue* cq) {
Call Channel::CreateCall(const RpcMethod& method, ClientContext* context,
CompletionQueue* cq) {
const bool kRegistered = method.channel_tag() && context->authority().empty();
grpc_call* c_call = NULL;
if (kRegistered) {
@ -110,11 +109,10 @@ internal::Call Channel::CreateCall(const internal::RpcMethod& method,
}
grpc_census_call_set_context(c_call, context->census_context());
context->set_call(c_call, shared_from_this());
return internal::Call(c_call, this, cq);
return Call(c_call, this, cq);
}
void Channel::PerformOpsOnCall(internal::CallOpSetInterface* ops,
internal::Call* call) {
void Channel::PerformOpsOnCall(CallOpSetInterface* ops, Call* call) {
static const size_t MAX_OPS = 8;
size_t nops = 0;
grpc_op cops[MAX_OPS];
@ -133,7 +131,7 @@ grpc_connectivity_state Channel::GetState(bool try_to_connect) {
}
namespace {
class TagSaver final : public internal::CompletionQueueTag {
class TagSaver final : public CompletionQueueTag {
public:
explicit TagSaver(void* tag) : tag_(tag) {}
~TagSaver() override {}

6
src/cpp/client/generic_stub.cc
Unescape View File

@ -27,11 +27,9 @@ std::unique_ptr<GenericClientAsyncReaderWriter> GenericStub::Call(
ClientContext* context, const grpc::string& method, CompletionQueue* cq,
void* tag) {
return std::unique_ptr<GenericClientAsyncReaderWriter>(
GenericClientAsyncReaderWriter::internal::Create(
GenericClientAsyncReaderWriter::Create(
channel_.get(), cq,
internal::RpcMethod(method.c_str(),
internal::RpcMethod::BIDI_STREAMING),
context, tag));
RpcMethod(method.c_str(), RpcMethod::BIDI_STREAMING), context, tag));
}
} // namespace grpc

2
src/cpp/common/completion_queue_cc.cc
Unescape View File

@ -60,7 +60,7 @@ CompletionQueue::NextStatus CompletionQueue::AsyncNextInternal(
case GRPC_QUEUE_SHUTDOWN:
return SHUTDOWN;
case GRPC_OP_COMPLETE:
auto cq_tag = static_cast<internal::CompletionQueueTag*>(ev.tag);
auto cq_tag = static_cast<CompletionQueueTag*>(ev.tag);
*ok = ev.success != 0;
*tag = cq_tag;
if (cq_tag->FinalizeResult(tag, ok)) {

9
src/cpp/server/health/default_health_check_service.cc
Unescape View File

@ -36,12 +36,11 @@ const char kHealthCheckMethodName[] = "/grpc.health.v1.Health/Check";
DefaultHealthCheckService::HealthCheckServiceImpl::HealthCheckServiceImpl(
DefaultHealthCheckService* service)
: service_(service), method_(nullptr) {
internal::MethodHandler* handler =
new internal::RpcMethodHandler<HealthCheckServiceImpl, ByteBuffer,
ByteBuffer>(
MethodHandler* handler =
new RpcMethodHandler<HealthCheckServiceImpl, ByteBuffer, ByteBuffer>(
std::mem_fn(&HealthCheckServiceImpl::Check), this);
method_ = new internal::RpcServiceMethod(
kHealthCheckMethodName, internal::RpcMethod::NORMAL_RPC, handler);
method_ = new RpcServiceMethod(kHealthCheckMethodName, RpcMethod::NORMAL_RPC,
handler);
AddMethod(method_);
}

2
src/cpp/server/health/default_health_check_service.h
Unescape View File

@ -41,7 +41,7 @@ class DefaultHealthCheckService final : public HealthCheckServiceInterface {
private:
const DefaultHealthCheckService* const service_;
internal::RpcServiceMethod* method_;
RpcServiceMethod* method_;
};
DefaultHealthCheckService();

69
src/cpp/server/server_cc.cc
Unescape View File

@ -88,8 +88,7 @@ class Server::UnimplementedAsyncRequest final
ServerCompletionQueue* const cq_;
};
typedef internal::SneakyCallOpSet<internal::CallOpSendInitialMetadata,
internal::CallOpServerSendStatus>
typedef SneakyCallOpSet<CallOpSendInitialMetadata, CallOpServerSendStatus>
UnimplementedAsyncResponseOp;
class Server::UnimplementedAsyncResponse final
: public UnimplementedAsyncResponseOp {
@ -107,12 +106,12 @@ class Server::UnimplementedAsyncResponse final
UnimplementedAsyncRequest* const request_;
};
class ShutdownTag : public internal::CompletionQueueTag {
class ShutdownTag : public CompletionQueueTag {
public:
bool FinalizeResult(void** tag, bool* status) { return false; }
};
class DummyTag : public internal::CompletionQueueTag {
class DummyTag : public CompletionQueueTag {
public:
bool FinalizeResult(void** tag, bool* status) {
*status = true;
@ -120,15 +119,15 @@ class DummyTag : public internal::CompletionQueueTag {
}
};
class Server::SyncRequest final : public internal::CompletionQueueTag {
class Server::SyncRequest final : public CompletionQueueTag {
public:
SyncRequest(internal::RpcServiceMethod* method, void* tag)
SyncRequest(RpcServiceMethod* method, void* tag)
: method_(method),
tag_(tag),
in_flight_(false),
has_request_payload_(
method->method_type() == internal::RpcMethod::NORMAL_RPC ||
method->method_type() == internal::RpcMethod::SERVER_STREAMING),
has_request_payload_(method->method_type() == RpcMethod::NORMAL_RPC ||
method->method_type() ==
RpcMethod::SERVER_STREAMING),
call_details_(nullptr),
cq_(nullptr) {
grpc_metadata_array_init(&request_metadata_);
@ -205,14 +204,14 @@ class Server::SyncRequest final : public internal::CompletionQueueTag {
void Run(std::shared_ptr<GlobalCallbacks> global_callbacks) {
ctx_.BeginCompletionOp(&call_);
global_callbacks->PreSynchronousRequest(&ctx_);
method_->handler()->RunHandler(internal::MethodHandler::HandlerParameter(
&call_, &ctx_, request_payload_));
method_->handler()->RunHandler(
MethodHandler::HandlerParameter(&call_, &ctx_, request_payload_));
global_callbacks->PostSynchronousRequest(&ctx_);
request_payload_ = nullptr;
cq_.Shutdown();
internal::CompletionQueueTag* op_tag = ctx_.GetCompletionOpTag();
CompletionQueueTag* op_tag = ctx_.GetCompletionOpTag();
cq_.TryPluck(op_tag, gpr_inf_future(GPR_CLOCK_REALTIME));
/* Ensure the cq_ is shutdown */
@ -222,15 +221,15 @@ class Server::SyncRequest final : public internal::CompletionQueueTag {
private:
CompletionQueue cq_;
internal::Call call_;
Call call_;
ServerContext ctx_;
const bool has_request_payload_;
grpc_byte_buffer* request_payload_;
internal::RpcServiceMethod* const method_;
RpcServiceMethod* const method_;
};
private:
internal::RpcServiceMethod* const method_;
RpcServiceMethod* const method_;
void* const tag_;
bool in_flight_;
const bool has_request_payload_;
@ -303,15 +302,14 @@ class Server::SyncRequestThreadManager : public ThreadManager {
// object
}
void AddSyncMethod(internal::RpcServiceMethod* method, void* tag) {
void AddSyncMethod(RpcServiceMethod* method, void* tag) {
sync_requests_.emplace_back(new SyncRequest(method, tag));
}
void AddUnknownSyncMethod() {
if (!sync_requests_.empty()) {
unknown_method_.reset(new internal::RpcServiceMethod(
"unknown", internal::RpcMethod::BIDI_STREAMING,
new internal::UnknownMethodHandler));
unknown_method_.reset(new RpcServiceMethod(
"unknown", RpcMethod::BIDI_STREAMING, new UnknownMethodHandler));
sync_requests_.emplace_back(
new SyncRequest(unknown_method_.get(), nullptr));
}
@ -348,8 +346,8 @@ class Server::SyncRequestThreadManager : public ThreadManager {
CompletionQueue* server_cq_;
int cq_timeout_msec_;
std::vector<std::unique_ptr<SyncRequest>> sync_requests_;
std::unique_ptr<internal::RpcServiceMethod> unknown_method_;
std::unique_ptr<internal::RpcServiceMethod> health_check_;
std::unique_ptr<RpcServiceMethod> unknown_method_;
std::unique_ptr<RpcServiceMethod> health_check_;
std::shared_ptr<Server::GlobalCallbacks> global_callbacks_;
};
@ -432,13 +430,13 @@ std::shared_ptr<Channel> Server::InProcessChannel(
}
static grpc_server_register_method_payload_handling PayloadHandlingForMethod(
internal::RpcServiceMethod* method) {
RpcServiceMethod* method) {
switch (method->method_type()) {
case internal::RpcMethod::NORMAL_RPC:
case internal::RpcMethod::SERVER_STREAMING:
case RpcMethod::NORMAL_RPC:
case RpcMethod::SERVER_STREAMING:
return GRPC_SRM_PAYLOAD_READ_INITIAL_BYTE_BUFFER;
case internal::RpcMethod::CLIENT_STREAMING:
case internal::RpcMethod::BIDI_STREAMING:
case RpcMethod::CLIENT_STREAMING:
case RpcMethod::BIDI_STREAMING:
return GRPC_SRM_PAYLOAD_NONE;
}
GPR_UNREACHABLE_CODE(return GRPC_SRM_PAYLOAD_NONE;);
@ -459,7 +457,7 @@ bool Server::RegisterService(const grpc::string* host, Service* service) {
continue;
}
internal::RpcServiceMethod* method = it->get();
RpcServiceMethod* method = it->get();
void* tag = grpc_server_register_method(
server_, method->name(), host ? host->c_str() : nullptr,
PayloadHandlingForMethod(method), 0);
@ -599,8 +597,7 @@ void Server::Wait() {
}
}
void Server::PerformOpsOnCall(internal::CallOpSetInterface* ops,
internal::Call* call) {
void Server::PerformOpsOnCall(CallOpSetInterface* ops, Call* call) {
static const size_t MAX_OPS = 8;
size_t nops = 0;
grpc_op cops[MAX_OPS];
@ -611,8 +608,8 @@ void Server::PerformOpsOnCall(internal::CallOpSetInterface* ops,
ServerInterface::BaseAsyncRequest::BaseAsyncRequest(
ServerInterface* server, ServerContext* context,
internal::ServerAsyncStreamingInterface* stream, CompletionQueue* call_cq,
void* tag, bool delete_on_finalize)
ServerAsyncStreamingInterface* stream, CompletionQueue* call_cq, void* tag,
bool delete_on_finalize)
: server_(server),
context_(context),
stream_(stream),
@ -634,8 +631,7 @@ bool ServerInterface::BaseAsyncRequest::FinalizeResult(void** tag,
}
context_->set_call(call_);
context_->cq_ = call_cq_;
internal::Call call(call_, server_, call_cq_,
server_->max_receive_message_size());
Call call(call_, server_, call_cq_, server_->max_receive_message_size());
if (*status && call_) {
context_->BeginCompletionOp(&call);
}
@ -650,8 +646,7 @@ bool ServerInterface::BaseAsyncRequest::FinalizeResult(void** tag,
ServerInterface::RegisteredAsyncRequest::RegisteredAsyncRequest(
ServerInterface* server, ServerContext* context,
internal::ServerAsyncStreamingInterface* stream, CompletionQueue* call_cq,
void* tag)
ServerAsyncStreamingInterface* stream, CompletionQueue* call_cq, void* tag)
: BaseAsyncRequest(server, context, stream, call_cq, tag, true) {}
void ServerInterface::RegisteredAsyncRequest::IssueRequest(
@ -665,7 +660,7 @@ void ServerInterface::RegisteredAsyncRequest::IssueRequest(
ServerInterface::GenericAsyncRequest::GenericAsyncRequest(
ServerInterface* server, GenericServerContext* context,
internal::ServerAsyncStreamingInterface* stream, CompletionQueue* call_cq,
ServerAsyncStreamingInterface* stream, CompletionQueue* call_cq,
ServerCompletionQueue* notification_cq, void* tag, bool delete_on_finalize)
: BaseAsyncRequest(server, context, stream, call_cq, tag,
delete_on_finalize) {
@ -707,7 +702,7 @@ Server::UnimplementedAsyncResponse::UnimplementedAsyncResponse(
UnimplementedAsyncRequest* request)
: request_(request) {
Status status(StatusCode::UNIMPLEMENTED, "");
internal::UnknownMethodHandler::FillOps(request_->context(), this);
UnknownMethodHandler::FillOps(request_->context(), this);
request_->stream()->call_.PerformOps(this);
}

8
src/cpp/server/server_context.cc
Unescape View File

@ -37,7 +37,7 @@ namespace grpc {
// CompletionOp
class ServerContext::CompletionOp final : public internal::CallOpSetInterface {
class ServerContext::CompletionOp final : public CallOpSetInterface {
public:
// initial refs: one in the server context, one in the cq
CompletionOp()
@ -146,7 +146,7 @@ ServerContext::~ServerContext() {
}
}
void ServerContext::BeginCompletionOp(internal::Call* call) {
void ServerContext::BeginCompletionOp(Call* call) {
GPR_ASSERT(!completion_op_);
completion_op_ = new CompletionOp();
if (has_notify_when_done_tag_) {
@ -155,8 +155,8 @@ void ServerContext::BeginCompletionOp(internal::Call* call) {
call->PerformOps(completion_op_);
}
internal::CompletionQueueTag* ServerContext::GetCompletionOpTag() {
return static_cast<internal::CompletionQueueTag*>(completion_op_);
CompletionQueueTag* ServerContext::GetCompletionOpTag() {
return static_cast<CompletionQueueTag*>(completion_op_);
}
void ServerContext::AddInitialMetadata(const grpc::string& key,

17
test/cpp/codegen/compiler_test_golden
Unescape View File

@ -39,6 +39,7 @@
namespace grpc {
class CompletionQueue;
class Channel;
class RpcService;
class ServerCompletionQueue;
class ServerContext;
} // namespace grpc
@ -136,10 +137,10 @@ class ServiceA final {
::grpc::ClientAsyncReader< ::grpc::testing::Response>* AsyncMethodA3Raw(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq, void* tag) override;
::grpc::ClientReaderWriter< ::grpc::testing::Request, ::grpc::testing::Response>* MethodA4Raw(::grpc::ClientContext* context) override;
::grpc::ClientAsyncReaderWriter< ::grpc::testing::Request, ::grpc::testing::Response>* AsyncMethodA4Raw(::grpc::ClientContext* context, ::grpc::CompletionQueue* cq, void* tag) override;
const ::grpc::internal::RpcMethod rpcmethod_MethodA1_;
const ::grpc::internal::RpcMethod rpcmethod_MethodA2_;
const ::grpc::internal::RpcMethod rpcmethod_MethodA3_;
const ::grpc::internal::RpcMethod rpcmethod_MethodA4_;
const ::grpc::RpcMethod rpcmethod_MethodA1_;
const ::grpc::RpcMethod rpcmethod_MethodA2_;
const ::grpc::RpcMethod rpcmethod_MethodA3_;
const ::grpc::RpcMethod rpcmethod_MethodA4_;
};
static std::unique_ptr<Stub> NewStub(const std::shared_ptr< ::grpc::ChannelInterface>& channel, const ::grpc::StubOptions& options = ::grpc::StubOptions());
@ -319,7 +320,7 @@ class ServiceA final {
public:
WithStreamedUnaryMethod_MethodA1() {
::grpc::Service::MarkMethodStreamed(0,
new ::grpc::internal::StreamedUnaryHandler< ::grpc::testing::Request, ::grpc::testing::Response>(std::bind(&WithStreamedUnaryMethod_MethodA1<BaseClass>::StreamedMethodA1, this, std::placeholders::_1, std::placeholders::_2)));
new ::grpc::StreamedUnaryHandler< ::grpc::testing::Request, ::grpc::testing::Response>(std::bind(&WithStreamedUnaryMethod_MethodA1<BaseClass>::StreamedMethodA1, this, std::placeholders::_1, std::placeholders::_2)));
}
~WithStreamedUnaryMethod_MethodA1() override {
BaseClassMustBeDerivedFromService(this);
@ -340,7 +341,7 @@ class ServiceA final {
public:
WithSplitStreamingMethod_MethodA3() {
::grpc::Service::MarkMethodStreamed(2,
new ::grpc::internal::SplitServerStreamingHandler< ::grpc::testing::Request, ::grpc::testing::Response>(std::bind(&WithSplitStreamingMethod_MethodA3<BaseClass>::StreamedMethodA3, this, std::placeholders::_1, std::placeholders::_2)));
new ::grpc::SplitServerStreamingHandler< ::grpc::testing::Request, ::grpc::testing::Response>(std::bind(&WithSplitStreamingMethod_MethodA3<BaseClass>::StreamedMethodA3, this, std::placeholders::_1, std::placeholders::_2)));
}
~WithSplitStreamingMethod_MethodA3() override {
BaseClassMustBeDerivedFromService(this);
@ -386,7 +387,7 @@ class ServiceB final {
private:
std::shared_ptr< ::grpc::ChannelInterface> channel_;
::grpc::ClientAsyncResponseReader< ::grpc::testing::Response>* AsyncMethodB1Raw(::grpc::ClientContext* context, const ::grpc::testing::Request& request, ::grpc::CompletionQueue* cq) override;
const ::grpc::internal::RpcMethod rpcmethod_MethodB1_;
const ::grpc::RpcMethod rpcmethod_MethodB1_;
};
static std::unique_ptr<Stub> NewStub(const std::shared_ptr< ::grpc::ChannelInterface>& channel, const ::grpc::StubOptions& options = ::grpc::StubOptions());
@ -443,7 +444,7 @@ class ServiceB final {
public:
WithStreamedUnaryMethod_MethodB1() {
::grpc::Service::MarkMethodStreamed(0,
new ::grpc::internal::StreamedUnaryHandler< ::grpc::testing::Request, ::grpc::testing::Response>(std::bind(&WithStreamedUnaryMethod_MethodB1<BaseClass>::StreamedMethodB1, this, std::placeholders::_1, std::placeholders::_2)));
new ::grpc::StreamedUnaryHandler< ::grpc::testing::Request, ::grpc::testing::Response>(std::bind(&WithStreamedUnaryMethod_MethodB1<BaseClass>::StreamedMethodB1, this, std::placeholders::_1, std::placeholders::_2)));
}
~WithStreamedUnaryMethod_MethodB1() override {
BaseClassMustBeDerivedFromService(this);

2
test/cpp/microbenchmarks/bm_cq.cc
Unescape View File

@ -69,7 +69,7 @@ BENCHMARK(BM_CreateDestroyCore);
static void DoneWithCompletionOnStack(grpc_exec_ctx* exec_ctx, void* arg,
grpc_cq_completion* completion) {}
class DummyTag final : public internal::CompletionQueueTag {
class DummyTag final : public CompletionQueueTag {
public:
bool FinalizeResult(void** tag, bool* status) override { return true; }
};

Write Preview
Loading…
Cancel
Save