diff options
author | 2015-05-04 14:51:27 -0700 | |
---|---|---|
committer | 2015-05-04 14:51:27 -0700 | |
commit | 4273f1e3bc411a1aff34fedad5eaf0c967801b93 (patch) | |
tree | 28f74fad05c2b6638a46f7677d0a9ee1f72506e1 | |
parent | 72e0fb83413bf1c38e07b8f783d551e5c6b7c89f (diff) |
add client side streaming interfaces
-rw-r--r-- | include/grpc++/async_unary_call.h | 13 | ||||
-rw-r--r-- | include/grpc++/stream.h | 98 |
2 files changed, 79 insertions, 32 deletions
diff --git a/include/grpc++/async_unary_call.h b/include/grpc++/async_unary_call.h index d1d5be5b50..97cec77955 100644 --- a/include/grpc++/async_unary_call.h +++ b/include/grpc++/async_unary_call.h @@ -44,8 +44,19 @@ #include <grpc/support/log.h> namespace grpc { + +template <class R> +class ClientAsyncResponseReaderInterface { + public: + virtual ~ClientAsyncResponseReaderInterface() {} + virtual void ReadInitialMetadata(void* tag) = 0; + virtual void Finish(R* msg, Status* status, void* tag) = 0; + +}; + template <class R> -class ClientAsyncResponseReader GRPC_FINAL { +class ClientAsyncResponseReader GRPC_FINAL + : public ClientAsyncResponseReaderInterface<R> { public: ClientAsyncResponseReader(ChannelInterface* channel, CompletionQueue* cq, const RpcMethod& method, ClientContext* context, diff --git a/include/grpc++/stream.h b/include/grpc++/stream.h index 6647e345c0..0efdbdaeca 100644 --- a/include/grpc++/stream.h +++ b/include/grpc++/stream.h @@ -83,8 +83,14 @@ class WriterInterface { }; template <class R> -class ClientReader GRPC_FINAL : public ClientStreamingInterface, - public ReaderInterface<R> { +class ClientReaderInterface : public ClientStreamingInterface, + public ReaderInterface<R> { + public: + virtual void WaitForInitialMetadata() = 0; +}; + +template <class R> +class ClientReader GRPC_FINAL : public ClientReaderInterface<R> { public: // Blocking create a stream and write the first request out. ClientReader(ChannelInterface* channel, const RpcMethod& method, @@ -111,7 +117,7 @@ class ClientReader GRPC_FINAL : public ClientStreamingInterface, GPR_ASSERT(cq_.Pluck(&buf)); } - virtual bool Read(R* msg) GRPC_OVERRIDE { + bool Read(R* msg) GRPC_OVERRIDE { CallOpBuffer buf; if (!context_->initial_metadata_received_) { buf.AddRecvInitialMetadata(context_); @@ -121,7 +127,7 @@ class ClientReader GRPC_FINAL : public ClientStreamingInterface, return cq_.Pluck(&buf) && buf.got_message; } - virtual Status Finish() GRPC_OVERRIDE { + Status Finish() GRPC_OVERRIDE { CallOpBuffer buf; Status status; buf.AddClientRecvStatus(context_, &status); @@ -137,8 +143,14 @@ class ClientReader GRPC_FINAL : public ClientStreamingInterface, }; template <class W> -class ClientWriter GRPC_FINAL : public ClientStreamingInterface, - public WriterInterface<W> { +class ClientWriterInterface : public ClientStreamingInterface, + public WriterInterface<W> { + public: + virtual bool WritesDone() = 0; +}; + +template <class W> +class ClientWriter GRPC_FINAL : public ClientWriterInterface<W> { public: // Blocking create a stream. ClientWriter(ChannelInterface* channel, const RpcMethod& method, @@ -152,14 +164,14 @@ class ClientWriter GRPC_FINAL : public ClientStreamingInterface, cq_.Pluck(&buf); } - virtual bool Write(const W& msg) GRPC_OVERRIDE { + bool Write(const W& msg) GRPC_OVERRIDE { CallOpBuffer buf; buf.AddSendMessage(msg); call_.PerformOps(&buf); return cq_.Pluck(&buf); } - virtual bool WritesDone() { + bool WritesDone() GRPC_OVERRIDE { CallOpBuffer buf; buf.AddClientSendClose(); call_.PerformOps(&buf); @@ -167,7 +179,7 @@ class ClientWriter GRPC_FINAL : public ClientStreamingInterface, } // Read the final response and wait for the final status. - virtual Status Finish() GRPC_OVERRIDE { + Status Finish() GRPC_OVERRIDE { CallOpBuffer buf; Status status; buf.AddRecvMessage(response_); @@ -186,9 +198,16 @@ class ClientWriter GRPC_FINAL : public ClientStreamingInterface, // Client-side interface for bi-directional streaming. template <class W, class R> -class ClientReaderWriter GRPC_FINAL : public ClientStreamingInterface, - public WriterInterface<W>, - public ReaderInterface<R> { +class ClientReaderWriterInterface : public ClientStreamingInterface, + public WriterInterface<W>, + public ReaderInterface<R> { + public: + virtual void WaitForInitialMetadata() = 0; + virtual bool WritesDone() = 0; +}; + +template <class W, class R> +class ClientReaderWriter GRPC_FINAL : public ClientReaderWriterInterface<W, R> { public: // Blocking create a stream. ClientReaderWriter(ChannelInterface* channel, const RpcMethod& method, @@ -213,7 +232,7 @@ class ClientReaderWriter GRPC_FINAL : public ClientStreamingInterface, GPR_ASSERT(cq_.Pluck(&buf)); } - virtual bool Read(R* msg) GRPC_OVERRIDE { + bool Read(R* msg) GRPC_OVERRIDE { CallOpBuffer buf; if (!context_->initial_metadata_received_) { buf.AddRecvInitialMetadata(context_); @@ -223,21 +242,21 @@ class ClientReaderWriter GRPC_FINAL : public ClientStreamingInterface, return cq_.Pluck(&buf) && buf.got_message; } - virtual bool Write(const W& msg) GRPC_OVERRIDE { + bool Write(const W& msg) GRPC_OVERRIDE { CallOpBuffer buf; buf.AddSendMessage(msg); call_.PerformOps(&buf); return cq_.Pluck(&buf); } - virtual bool WritesDone() { + bool WritesDone() GRPC_OVERRIDE { CallOpBuffer buf; buf.AddClientSendClose(); call_.PerformOps(&buf); return cq_.Pluck(&buf); } - virtual Status Finish() GRPC_OVERRIDE { + Status Finish() GRPC_OVERRIDE { CallOpBuffer buf; Status status; buf.AddClientRecvStatus(context_, &status); @@ -267,7 +286,7 @@ class ServerReader GRPC_FINAL : public ReaderInterface<R> { call_->cq()->Pluck(&buf); } - virtual bool Read(R* msg) GRPC_OVERRIDE { + bool Read(R* msg) GRPC_OVERRIDE { CallOpBuffer buf; buf.AddRecvMessage(msg); call_->PerformOps(&buf); @@ -294,7 +313,7 @@ class ServerWriter GRPC_FINAL : public WriterInterface<W> { call_->cq()->Pluck(&buf); } - virtual bool Write(const W& msg) GRPC_OVERRIDE { + bool Write(const W& msg) GRPC_OVERRIDE { CallOpBuffer buf; if (!ctx_->sent_initial_metadata_) { buf.AddSendInitialMetadata(&ctx_->initial_metadata_); @@ -327,14 +346,14 @@ class ServerReaderWriter GRPC_FINAL : public WriterInterface<W>, call_->cq()->Pluck(&buf); } - virtual bool Read(R* msg) GRPC_OVERRIDE { + bool Read(R* msg) GRPC_OVERRIDE { CallOpBuffer buf; buf.AddRecvMessage(msg); call_->PerformOps(&buf); return call_->cq()->Pluck(&buf) && buf.got_message; } - virtual bool Write(const W& msg) GRPC_OVERRIDE { + bool Write(const W& msg) GRPC_OVERRIDE { CallOpBuffer buf; if (!ctx_->sent_initial_metadata_) { buf.AddSendInitialMetadata(&ctx_->initial_metadata_); @@ -380,8 +399,12 @@ class AsyncWriterInterface { }; template <class R> -class ClientAsyncReader GRPC_FINAL : public ClientAsyncStreamingInterface, - public AsyncReaderInterface<R> { +class ClientAsyncReaderInterface : public ClientAsyncStreamingInterface, + public AsyncReaderInterface<R> { +}; + +template <class R> +class ClientAsyncReader GRPC_FINAL : public ClientAsyncReaderInterface<R> { public: // Create a stream and write the first request out. ClientAsyncReader(ChannelInterface* channel, CompletionQueue* cq, @@ -431,8 +454,14 @@ class ClientAsyncReader GRPC_FINAL : public ClientAsyncStreamingInterface, }; template <class W> -class ClientAsyncWriter GRPC_FINAL : public ClientAsyncStreamingInterface, - public AsyncWriterInterface<W> { +class ClientAsyncWriterInterface : public ClientAsyncStreamingInterface, + public AsyncWriterInterface<W> { + public: + virtual void WritesDone(void* tag) = 0; +}; + +template <class W> +class ClientAsyncWriter GRPC_FINAL : public ClientAsyncWriterInterface<W> { public: ClientAsyncWriter(ChannelInterface* channel, CompletionQueue* cq, const RpcMethod& method, ClientContext* context, @@ -459,7 +488,7 @@ class ClientAsyncWriter GRPC_FINAL : public ClientAsyncStreamingInterface, call_.PerformOps(&write_buf_); } - void WritesDone(void* tag) { + void WritesDone(void* tag) GRPC_OVERRIDE { writes_done_buf_.Reset(tag); writes_done_buf_.AddClientSendClose(); call_.PerformOps(&writes_done_buf_); @@ -488,9 +517,16 @@ class ClientAsyncWriter GRPC_FINAL : public ClientAsyncStreamingInterface, // Client-side interface for bi-directional streaming. template <class W, class R> -class ClientAsyncReaderWriter GRPC_FINAL : public ClientAsyncStreamingInterface, - public AsyncWriterInterface<W>, - public AsyncReaderInterface<R> { +class ClientAsyncReaderWriterInterface : public ClientAsyncStreamingInterface, + public AsyncWriterInterface<W>, + public AsyncReaderInterface<R> { + public: + virtual void WritesDone(void* tag) = 0; +}; + +template <class W, class R> +class ClientAsyncReaderWriter GRPC_FINAL + : public ClientAsyncReaderWriterInterface<W, R> { public: ClientAsyncReaderWriter(ChannelInterface* channel, CompletionQueue* cq, const RpcMethod& method, ClientContext* context, @@ -524,7 +560,7 @@ class ClientAsyncReaderWriter GRPC_FINAL : public ClientAsyncStreamingInterface, call_.PerformOps(&write_buf_); } - void WritesDone(void* tag) { + void WritesDone(void* tag) GRPC_OVERRIDE { writes_done_buf_.Reset(tag); writes_done_buf_.AddClientSendClose(); call_.PerformOps(&writes_done_buf_); @@ -671,13 +707,13 @@ class ServerAsyncReaderWriter GRPC_FINAL : public ServerAsyncStreamingInterface, call_.PerformOps(&meta_buf_); } - virtual void Read(R* msg, void* tag) GRPC_OVERRIDE { + void Read(R* msg, void* tag) GRPC_OVERRIDE { read_buf_.Reset(tag); read_buf_.AddRecvMessage(msg); call_.PerformOps(&read_buf_); } - virtual void Write(const W& msg, void* tag) GRPC_OVERRIDE { + void Write(const W& msg, void* tag) GRPC_OVERRIDE { write_buf_.Reset(tag); if (!ctx_->sent_initial_metadata_) { write_buf_.AddSendInitialMetadata(&ctx_->initial_metadata_); |