legacy-libs/grpc-cloned/ext/completion_queue.cc

   1 /*
   2  *
   3  * Copyright 2016 gRPC authors.
   4  *
   5  * Licensed under the Apache License, Version 2.0 (the "License");
   6  * you may not use this file except in compliance with the License.
   7  * You may obtain a copy of the License at
   8  *
   9  *     http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  * Unless required by applicable law or agreed to in writing, software
  12  * distributed under the License is distributed on an "AS IS" BASIS,
  13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14  * See the License for the specific language governing permissions and
  15  * limitations under the License.
  16  *
  17  */
  18
  19 #include <grpc/grpc.h>
  20 #include <node.h>
  21 #include <uv.h>
  22 #include <v8.h>
  23
  24 #include "call.h"
  25 #include "completion_queue.h"
  26
  27 namespace grpc {
  28 namespace node {
  29
  30 using v8::Local;
  31 using v8::Object;
  32 using v8::Value;
  33
  34 grpc_completion_queue *queue;
  35 uv_prepare_t prepare;
  36 int pending_batches;
  37
  38 static void drain_completion_queue(uv_prepare_t *handle) {
  39   Nan::HandleScope scope;
  40   grpc_event event;
  41   (void)handle;
  42   do {
  43     event = grpc_completion_queue_next(queue, gpr_inf_past(GPR_CLOCK_MONOTONIC),
  44                                        NULL);
  45
  46     if (event.type == GRPC_OP_COMPLETE) {
  47       const char *error_message;
  48       if (event.success) {
  49         error_message = NULL;
  50       } else {
  51         error_message = "The async function encountered an error";
  52       }
  53       CompleteTag(event.tag, error_message);
  54       grpc::node::DestroyTag(event.tag);
  55       pending_batches--;
  56     }
  57     if (pending_batches == 0) {
  58       uv_prepare_stop(&prepare);
  59     }
  60   } while (event.type != GRPC_QUEUE_TIMEOUT);
  61 }
  62
  63 grpc_completion_queue *GetCompletionQueue() { return queue; }
  64
  65 void CompletionQueueNext() {
  66   if (pending_batches == 0) {
  67     uv_prepare_start(&prepare, drain_completion_queue);
  68   }
  69   pending_batches++;
  70 }
  71
  72 void CompletionQueueInit(Local<Object> exports) {
  73   queue = grpc_completion_queue_create_for_next(NULL);
  74   uv_prepare_init(uv_default_loop(), &prepare);
  75   pending_batches = 0;
  76 }
  77
  78 void CompletionQueueForcePoll() {
  79   /* This sets the prepare object to poll on the completion queue the next time
  80    * Node polls for IO. But it doesn't increment the number of pending batches,
  81    * so it will immediately stop polling after that unless there is an
  82    * intervening CompletionQueueNext call */
  83   if (pending_batches == 0) {
  84     uv_prepare_start(&prepare, drain_completion_queue);
  85   }
  86 }
  87
  88 }  // namespace node
  89 }  // namespace grpc