Cleanup some cruft

include/grpc++/async_server.h

-/*
- *
- * Copyright 2014, 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 __GRPCPP_ASYNC_SERVER_H__
-#define __GRPCPP_ASYNC_SERVER_H__
-
-#include <mutex>
-
-#include <grpc++/config.h>
-
-struct grpc_server;
-
-namespace grpc {
-class CompletionQueue;
-
-class AsyncServer {
- public:
-  explicit AsyncServer(CompletionQueue* cc);
-  ~AsyncServer();
-
-  void AddPort(const grpc::string& addr);
-
-  void Start();
-
-  // The user has to call this to get one new rpc on the completion
-  // queue.
-  void RequestOneRpc();
-
-  void Shutdown();
-
- private:
-  bool started_;
-  std::mutex shutdown_mu_;
-  bool shutdown_;
-  grpc_server* server_;
-};
-
-}  // namespace grpc
-
-#endif  // __GRPCPP_ASYNC_SERVER_H__

include/grpc++/async_server_context.h

-/*
- *
- * Copyright 2014, 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 __GRPCPP_ASYNC_SERVER_CONTEXT_H__
-#define __GRPCPP_ASYNC_SERVER_CONTEXT_H__
-
-#include <chrono>
-
-#include <grpc++/config.h>
-
-struct grpc_byte_buffer;
-struct grpc_call;
-struct grpc_completion_queue;
-
-namespace google {
-namespace protobuf {
-class Message;
-}
-}
-
-using std::chrono::system_clock;
-
-namespace grpc {
-class Status;
-
-// TODO(rocking): wrap grpc c structures.
-class AsyncServerContext {
- public:
-  AsyncServerContext(grpc_call* call, const grpc::string& method,
-                     const grpc::string& host,
-                     system_clock::time_point absolute_deadline);
-  ~AsyncServerContext();
-
-  // Accept this rpc, bind it to a completion queue.
-  void Accept(grpc_completion_queue* cq);
-
-  // Read and write calls, all async. Return true for success.
-  bool StartRead(google::protobuf::Message* request);
-  bool StartWrite(const google::protobuf::Message& response, int flags);
-  bool StartWriteStatus(const Status& status);
-
-  bool ParseRead(grpc_byte_buffer* read_buffer);
-
-  grpc::string method() const { return method_; }
-  grpc::string host() const { return host_; }
-  system_clock::time_point absolute_deadline() { return absolute_deadline_; }
-
-  grpc_call* call() { return call_; }
-
- private:
-  AsyncServerContext(const AsyncServerContext&);
-  AsyncServerContext& operator=(const AsyncServerContext&);
-
-  // These properties may be moved to a ServerContext class.
-  const grpc::string method_;
-  const grpc::string host_;
-  system_clock::time_point absolute_deadline_;
-
-  google::protobuf::Message* request_;  // not owned
-  grpc_call* call_;                     // owned
-};
-
-}  // namespace grpc
-
-#endif  // __GRPCPP_ASYNC_SERVER_CONTEXT_H__

include/grpc++/call.h

@@ -67,7 +67,7 @@ class CallOpBuffer final : public CompletionQueueTag {
   void FillOps(grpc_op *ops, size_t *nops);
 
   // Called by completion queue just prior to returning from Next() or Pluck()
-  void FinalizeResult() override;
+  FinalizeResultOutput FinalizeResult(bool status) override;
 };
 
 class CCallDeleter {

include/grpc++/completion_queue.h

@@ -42,7 +42,13 @@ class CompletionQueue;
 
 class CompletionQueueTag {
  public:
-  virtual void FinalizeResult() = 0;
+  enum FinalizeResultOutput {
+    SUCCEED,
+    FAIL,
+    SWALLOW,
+  };
+
+  virtual FinalizeResultOutput FinalizeResult(bool status) = 0;
 
  private:
   friend class CompletionQueue;

include/grpc++/stream_context_interface.h

-/*
- *
- * Copyright 2014, 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 __GRPCPP_STREAM_CONTEXT_INTERFACE_H__
-#define __GRPCPP_STREAM_CONTEXT_INTERFACE_H__
-
-namespace google {
-namespace protobuf {
-class Message;
-}
-}
-
-namespace grpc {
-class Status;
-
-// An interface to avoid dependency on internal implementation.
-class StreamContextInterface {
- public:
-  virtual ~StreamContextInterface() {}
-
-  virtual void Start(bool buffered) = 0;
-
-  virtual bool Read(google::protobuf::Message* msg) = 0;
-  virtual bool Write(const google::protobuf::Message* msg, bool is_last) = 0;
-  virtual const Status& Wait() = 0;
-  virtual void Cancel() = 0;
-
-  virtual google::protobuf::Message* request() = 0;
-  virtual google::protobuf::Message* response() = 0;
-};
-
-}  // namespace grpc
-
-#endif  // __GRPCPP_STREAM_CONTEXT_INTERFACE_H__

include/grpc/grpc.h

@@ -542,14 +542,6 @@ grpc_call_error grpc_server_request_call(
     grpc_metadata_array *request_metadata,
     grpc_completion_queue *completion_queue, void *tag_new);
 
-void *grpc_server_register_call(grpc_server *server, const char *method, const char *host);
-
-grpc_call_error grpc_server_request_specific_call(
-    grpc_server *server, grpc_call **call, 
-    void *registered_call_tag, 
-    grpc_metadata_array *request_metadata,
-    grpc_completion_queue *completion_queue, void *tag_new);
-
 /* Create a server */
 grpc_server *grpc_server_create(grpc_completion_queue *cq,
                                 const grpc_channel_args *args);

src/cpp/common/completion_queue.cc

@@ -57,15 +57,19 @@ class EventDeleter {
 bool CompletionQueue::Next(void **tag, bool *ok) {
   std::unique_ptr<grpc_event, EventDeleter> ev;
 
-  ev.reset(grpc_completion_queue_next(cq_, gpr_inf_future));
-  if (ev->type == GRPC_QUEUE_SHUTDOWN) {
-    return false;
+  while (true) {
+    ev.reset(grpc_completion_queue_next(cq_, gpr_inf_future));
+    if (ev->type == GRPC_QUEUE_SHUTDOWN) {
+      return false;
+    }
+    auto cq_tag = static_cast<CompletionQueueTag *>(ev->tag);
+    switch (cq_tag->FinalizeResult(ev->data.op_complete == GRPC_OP_OK)) {
+      case CompletionQueueTag::SUCCEED: *ok = true; break;
+      case CompletionQueueTag::FAIL: *ok = false; break;
+      case CompletionQueueTag::SWALLOW: continue;
+    }
+    return true;
   }
-  auto cq_tag = static_cast<CompletionQueueTag *>(ev->tag);
-  cq_tag->FinalizeResult();
-  *tag = cq_tag->user_tag_;
-  *ok = ev->data.op_complete == GRPC_OP_OK;
-  return true;
 }
 
 }  // namespace grpc

src/cpp/server/async_server.cc

-/*
- *
- * Copyright 2014, 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.
- *
- */
-
-#include <grpc++/async_server.h>
-
-#include <grpc/grpc.h>
-#include <grpc/support/log.h>
-#include <grpc++/completion_queue.h>
-
-namespace grpc {
-
-AsyncServer::AsyncServer(CompletionQueue *cc)
-    : started_(false), shutdown_(false) {
-  server_ = grpc_server_create(cc->cq(), nullptr);
-}
-
-AsyncServer::~AsyncServer() {
-  std::unique_lock<std::mutex> lock(shutdown_mu_);
-  if (started_ && !shutdown_) {
-    lock.unlock();
-    Shutdown();
-  }
-  grpc_server_destroy(server_);
-}
-
-void AsyncServer::AddPort(const grpc::string &addr) {
-  GPR_ASSERT(!started_);
-  int success = grpc_server_add_http2_port(server_, addr.c_str());
-  GPR_ASSERT(success);
-}
-
-void AsyncServer::Start() {
-  GPR_ASSERT(!started_);
-  started_ = true;
-  grpc_server_start(server_);
-}
-
-void AsyncServer::RequestOneRpc() {
-  GPR_ASSERT(started_);
-  std::unique_lock<std::mutex> lock(shutdown_mu_);
-  if (shutdown_) {
-    return;
-  }
-  lock.unlock();
-  grpc_call_error err = grpc_server_request_call_old(server_, nullptr);
-  GPR_ASSERT(err == GRPC_CALL_OK);
-}
-
-void AsyncServer::Shutdown() {
-  std::unique_lock<std::mutex> lock(shutdown_mu_);
-  if (started_ && !shutdown_) {
-    shutdown_ = true;
-    lock.unlock();
-    // TODO(yangg) should we shutdown without start?
-    grpc_server_shutdown(server_);
-  }
-}
-
-}  // namespace grpc

src/cpp/server/server.cc

@@ -71,6 +71,8 @@ Server::~Server() {
   if (started_ && !shutdown_) {
     lock.unlock();
     Shutdown();
+  } else {
+    lock.unlock();
   }
   grpc_server_destroy(server_);
   if (thread_pool_owned_) {