1 files changed, 436 insertions, 0 deletions
diff --git a/include/grpc++/support/async_stream.h b/include/grpc++/support/async_stream.h
new file mode 100644
index 0000000000..4c12fda12f
--- /dev/null
+++ b/include/grpc++/support/async_stream.h
@@ -0,0 +1,436 @@
+/*
+ *
+ * Copyright 2015, Google Inc.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ *     * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *     * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ *     * Neither the name of Google Inc. nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+
+#ifndef GRPCXX_SUPPORT_ASYNC_STREAM_H
+#define GRPCXX_SUPPORT_ASYNC_STREAM_H
+
+#include <grpc/support/log.h>
+#include <grpc++/channel.h>
+#include <grpc++/client_context.h>
+#include <grpc++/completion_queue.h>
+#include <grpc++/impl/call.h>
+#include <grpc++/impl/service_type.h>
+#include <grpc++/server_context.h>
+#include <grpc++/support/status.h>
+
+namespace grpc {
+
+// Async interfaces
+// Common interface for all client side streaming.
+class ClientAsyncStreamingInterface {
+ public:
+  virtual ~ClientAsyncStreamingInterface() {}
+
+  virtual void ReadInitialMetadata(void* tag) = 0;
+
+  virtual void Finish(Status* status, void* tag) = 0;
+};
+
+// An interface that yields a sequence of R messages.
+template <class R>
+class AsyncReaderInterface {
+ public:
+  virtual ~AsyncReaderInterface() {}
+
+  virtual void Read(R* msg, void* tag) = 0;
+};
+
+// An interface that can be fed a sequence of W messages.
+template <class W>
+class AsyncWriterInterface {
+ public:
+  virtual ~AsyncWriterInterface() {}
+
+  virtual void Write(const W& msg, void* tag) = 0;
+};
+
+template <class 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.
+  template <class W>
+  ClientAsyncReader(Channel* channel, CompletionQueue* cq,
+                    const RpcMethod& method, ClientContext* context,
+                    const W& request, void* tag)
+      : context_(context), call_(channel->CreateCall(method, context, cq)) {
+    init_ops_.set_output_tag(tag);
+    init_ops_.SendInitialMetadata(context->send_initial_metadata_);
+    // TODO(ctiller): don't assert
+    GPR_ASSERT(init_ops_.SendMessage(request).ok());
+    init_ops_.ClientSendClose();
+    call_.PerformOps(&init_ops_);
+  }
+
+  void ReadInitialMetadata(void* tag) GRPC_OVERRIDE {
+    GPR_ASSERT(!context_->initial_metadata_received_);
+
+    meta_ops_.set_output_tag(tag);
+    meta_ops_.RecvInitialMetadata(context_);
+    call_.PerformOps(&meta_ops_);
+  }
+
+  void Read(R* msg, void* tag) GRPC_OVERRIDE {
+    read_ops_.set_output_tag(tag);
+    if (!context_->initial_metadata_received_) {
+      read_ops_.RecvInitialMetadata(context_);
+    }
+    read_ops_.RecvMessage(msg);
+    call_.PerformOps(&read_ops_);
+  }
+
+  void Finish(Status* status, void* tag) GRPC_OVERRIDE {
+    finish_ops_.set_output_tag(tag);
+    if (!context_->initial_metadata_received_) {
+      finish_ops_.RecvInitialMetadata(context_);
+    }
+    finish_ops_.ClientRecvStatus(context_, status);
+    call_.PerformOps(&finish_ops_);
+  }
+
+ private:
+  ClientContext* context_;
+  Call call_;
+  CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage, CallOpClientSendClose>
+      init_ops_;
+  CallOpSet<CallOpRecvInitialMetadata> meta_ops_;
+  CallOpSet<CallOpRecvInitialMetadata, CallOpRecvMessage<R>> read_ops_;
+  CallOpSet<CallOpRecvInitialMetadata, CallOpClientRecvStatus> finish_ops_;
+};
+
+template <class 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:
+  template <class R>
+  ClientAsyncWriter(Channel* channel, CompletionQueue* cq,
+                    const RpcMethod& method, ClientContext* context,
+                    R* response, void* tag)
+      : context_(context), call_(channel->CreateCall(method, context, cq)) {
+    finish_ops_.RecvMessage(response);
+
+    init_ops_.set_output_tag(tag);
+    init_ops_.SendInitialMetadata(context->send_initial_metadata_);
+    call_.PerformOps(&init_ops_);
+  }
+
+  void ReadInitialMetadata(void* tag) GRPC_OVERRIDE {
+    GPR_ASSERT(!context_->initial_metadata_received_);
+
+    meta_ops_.set_output_tag(tag);
+    meta_ops_.RecvInitialMetadata(context_);
+    call_.PerformOps(&meta_ops_);
+  }
+
+  void Write(const W& msg, void* tag) GRPC_OVERRIDE {
+    write_ops_.set_output_tag(tag);
+    // TODO(ctiller): don't assert
+    GPR_ASSERT(write_ops_.SendMessage(msg).ok());
+    call_.PerformOps(&write_ops_);
+  }
+
+  void WritesDone(void* tag) GRPC_OVERRIDE {
+    writes_done_ops_.set_output_tag(tag);
+    writes_done_ops_.ClientSendClose();
+    call_.PerformOps(&writes_done_ops_);
+  }
+
+  void Finish(Status* status, void* tag) GRPC_OVERRIDE {
+    finish_ops_.set_output_tag(tag);
+    if (!context_->initial_metadata_received_) {
+      finish_ops_.RecvInitialMetadata(context_);
+    }
+    finish_ops_.ClientRecvStatus(context_, status);
+    call_.PerformOps(&finish_ops_);
+  }
+
+ private:
+  ClientContext* context_;
+  Call call_;
+  CallOpSet<CallOpSendInitialMetadata> init_ops_;
+  CallOpSet<CallOpRecvInitialMetadata> meta_ops_;
+  CallOpSet<CallOpSendMessage> write_ops_;
+  CallOpSet<CallOpClientSendClose> writes_done_ops_;
+  CallOpSet<CallOpRecvInitialMetadata, CallOpGenericRecvMessage,
+            CallOpClientRecvStatus> finish_ops_;
+};
+
+// Client-side interface for bi-directional streaming.
+template <class W, class 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(Channel* channel, CompletionQueue* cq,
+                          const RpcMethod& method, ClientContext* context,
+                          void* tag)
+      : context_(context), call_(channel->CreateCall(method, context, cq)) {
+    init_ops_.set_output_tag(tag);
+    init_ops_.SendInitialMetadata(context->send_initial_metadata_);
+    call_.PerformOps(&init_ops_);
+  }
+
+  void ReadInitialMetadata(void* tag) GRPC_OVERRIDE {
+    GPR_ASSERT(!context_->initial_metadata_received_);
+
+    meta_ops_.set_output_tag(tag);
+    meta_ops_.RecvInitialMetadata(context_);
+    call_.PerformOps(&meta_ops_);
+  }
+
+  void Read(R* msg, void* tag) GRPC_OVERRIDE {
+    read_ops_.set_output_tag(tag);
+    if (!context_->initial_metadata_received_) {
+      read_ops_.RecvInitialMetadata(context_);
+    }
+    read_ops_.RecvMessage(msg);
+    call_.PerformOps(&read_ops_);
+  }
+
+  void Write(const W& msg, void* tag) GRPC_OVERRIDE {
+    write_ops_.set_output_tag(tag);
+    // TODO(ctiller): don't assert
+    GPR_ASSERT(write_ops_.SendMessage(msg).ok());
+    call_.PerformOps(&write_ops_);
+  }
+
+  void WritesDone(void* tag) GRPC_OVERRIDE {
+    writes_done_ops_.set_output_tag(tag);
+    writes_done_ops_.ClientSendClose();
+    call_.PerformOps(&writes_done_ops_);
+  }
+
+  void Finish(Status* status, void* tag) GRPC_OVERRIDE {
+    finish_ops_.set_output_tag(tag);
+    if (!context_->initial_metadata_received_) {
+      finish_ops_.RecvInitialMetadata(context_);
+    }
+    finish_ops_.ClientRecvStatus(context_, status);
+    call_.PerformOps(&finish_ops_);
+  }
+
+ private:
+  ClientContext* context_;
+  Call call_;
+  CallOpSet<CallOpSendInitialMetadata> init_ops_;
+  CallOpSet<CallOpRecvInitialMetadata> meta_ops_;
+  CallOpSet<CallOpRecvInitialMetadata, CallOpRecvMessage<R>> read_ops_;
+  CallOpSet<CallOpSendMessage> write_ops_;
+  CallOpSet<CallOpClientSendClose> writes_done_ops_;
+  CallOpSet<CallOpRecvInitialMetadata, CallOpClientRecvStatus> finish_ops_;
+};
+
+template <class W, class R>
+class ServerAsyncReader GRPC_FINAL : public ServerAsyncStreamingInterface,
+                                     public AsyncReaderInterface<R> {
+ public:
+  explicit ServerAsyncReader(ServerContext* ctx)
+      : call_(nullptr, nullptr, nullptr), ctx_(ctx) {}
+
+  void SendInitialMetadata(void* tag) GRPC_OVERRIDE {
+    GPR_ASSERT(!ctx_->sent_initial_metadata_);
+
+    meta_ops_.set_output_tag(tag);
+    meta_ops_.SendInitialMetadata(ctx_->initial_metadata_);
+    ctx_->sent_initial_metadata_ = true;
+    call_.PerformOps(&meta_ops_);
+  }
+
+  void Read(R* msg, void* tag) GRPC_OVERRIDE {
+    read_ops_.set_output_tag(tag);
+    read_ops_.RecvMessage(msg);
+    call_.PerformOps(&read_ops_);
+  }
+
+  void Finish(const W& msg, const Status& status, void* tag) {
+    finish_ops_.set_output_tag(tag);
+    if (!ctx_->sent_initial_metadata_) {
+      finish_ops_.SendInitialMetadata(ctx_->initial_metadata_);
+      ctx_->sent_initial_metadata_ = true;
+    }
+    // The response is dropped if the status is not OK.
+    if (status.ok()) {
+      finish_ops_.ServerSendStatus(ctx_->trailing_metadata_,
+                                   finish_ops_.SendMessage(msg));
+    } else {
+      finish_ops_.ServerSendStatus(ctx_->trailing_metadata_, status);
+    }
+    call_.PerformOps(&finish_ops_);
+  }
+
+  void FinishWithError(const Status& status, void* tag) {
+    GPR_ASSERT(!status.ok());
+    finish_ops_.set_output_tag(tag);
+    if (!ctx_->sent_initial_metadata_) {
+      finish_ops_.SendInitialMetadata(ctx_->initial_metadata_);
+      ctx_->sent_initial_metadata_ = true;
+    }
+    finish_ops_.ServerSendStatus(ctx_->trailing_metadata_, status);
+    call_.PerformOps(&finish_ops_);
+  }
+
+ private:
+  void BindCall(Call* call) GRPC_OVERRIDE { call_ = *call; }
+
+  Call call_;
+  ServerContext* ctx_;
+  CallOpSet<CallOpSendInitialMetadata> meta_ops_;
+  CallOpSet<CallOpRecvMessage<R>> read_ops_;
+  CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage,
+            CallOpServerSendStatus> finish_ops_;
+};
+
+template <class W>
+class ServerAsyncWriter GRPC_FINAL : public ServerAsyncStreamingInterface,
+                                     public AsyncWriterInterface<W> {
+ public:
+  explicit ServerAsyncWriter(ServerContext* ctx)
+      : call_(nullptr, nullptr, nullptr), ctx_(ctx) {}
+
+  void SendInitialMetadata(void* tag) GRPC_OVERRIDE {
+    GPR_ASSERT(!ctx_->sent_initial_metadata_);
+
+    meta_ops_.set_output_tag(tag);
+    meta_ops_.SendInitialMetadata(ctx_->initial_metadata_);
+    ctx_->sent_initial_metadata_ = true;
+    call_.PerformOps(&meta_ops_);
+  }
+
+  void Write(const W& msg, void* tag) GRPC_OVERRIDE {
+    write_ops_.set_output_tag(tag);
+    if (!ctx_->sent_initial_metadata_) {
+      write_ops_.SendInitialMetadata(ctx_->initial_metadata_);
+      ctx_->sent_initial_metadata_ = true;
+    }
+    // TODO(ctiller): don't assert
+    GPR_ASSERT(write_ops_.SendMessage(msg).ok());
+    call_.PerformOps(&write_ops_);
+  }
+
+  void Finish(const Status& status, void* tag) {
+    finish_ops_.set_output_tag(tag);
+    if (!ctx_->sent_initial_metadata_) {
+      finish_ops_.SendInitialMetadata(ctx_->initial_metadata_);
+      ctx_->sent_initial_metadata_ = true;
+    }
+    finish_ops_.ServerSendStatus(ctx_->trailing_metadata_, status);
+    call_.PerformOps(&finish_ops_);
+  }
+
+ private:
+  void BindCall(Call* call) GRPC_OVERRIDE { call_ = *call; }
+
+  Call call_;
+  ServerContext* ctx_;
+  CallOpSet<CallOpSendInitialMetadata> meta_ops_;
+  CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage> write_ops_;
+  CallOpSet<CallOpSendInitialMetadata, CallOpServerSendStatus> finish_ops_;
+};
+
+// Server-side interface for bi-directional streaming.
+template <class W, class R>
+class ServerAsyncReaderWriter GRPC_FINAL : public ServerAsyncStreamingInterface,
+                                           public AsyncWriterInterface<W>,
+                                           public AsyncReaderInterface<R> {
+ public:
+  explicit ServerAsyncReaderWriter(ServerContext* ctx)
+      : call_(nullptr, nullptr, nullptr), ctx_(ctx) {}
+
+  void SendInitialMetadata(void* tag) GRPC_OVERRIDE {
+    GPR_ASSERT(!ctx_->sent_initial_metadata_);
+
+    meta_ops_.set_output_tag(tag);
+    meta_ops_.SendInitialMetadata(ctx_->initial_metadata_);
+    ctx_->sent_initial_metadata_ = true;
+    call_.PerformOps(&meta_ops_);
+  }
+
+  void Read(R* msg, void* tag) GRPC_OVERRIDE {
+    read_ops_.set_output_tag(tag);
+    read_ops_.RecvMessage(msg);
+    call_.PerformOps(&read_ops_);
+  }
+
+  void Write(const W& msg, void* tag) GRPC_OVERRIDE {
+    write_ops_.set_output_tag(tag);
+    if (!ctx_->sent_initial_metadata_) {
+      write_ops_.SendInitialMetadata(ctx_->initial_metadata_);
+      ctx_->sent_initial_metadata_ = true;
+    }
+    // TODO(ctiller): don't assert
+    GPR_ASSERT(write_ops_.SendMessage(msg).ok());
+    call_.PerformOps(&write_ops_);
+  }
+
+  void Finish(const Status& status, void* tag) {
+    finish_ops_.set_output_tag(tag);
+    if (!ctx_->sent_initial_metadata_) {
+      finish_ops_.SendInitialMetadata(ctx_->initial_metadata_);
+      ctx_->sent_initial_metadata_ = true;
+    }
+    finish_ops_.ServerSendStatus(ctx_->trailing_metadata_, status);
+    call_.PerformOps(&finish_ops_);
+  }
+
+ private:
+  friend class ::grpc::Server;
+
+  void BindCall(Call* call) GRPC_OVERRIDE { call_ = *call; }
+
+  Call call_;
+  ServerContext* ctx_;
+  CallOpSet<CallOpSendInitialMetadata> meta_ops_;
+  CallOpSet<CallOpRecvMessage<R>> read_ops_;
+  CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage> write_ops_;
+  CallOpSet<CallOpSendInitialMetadata, CallOpServerSendStatus> finish_ops_;
+};
+
+}  // namespace grpc
+
+#endif  // GRPCXX_SUPPORT_ASYNC_STREAM_H