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

   1 /*
   2  *
   3  * Copyright 2015 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 <map>
  20 #include <memory>
  21 #include <vector>
  22
  23 #include <node.h>
  24
  25 #include "byte_buffer.h"
  26 #include "call.h"
  27 #include "call_credentials.h"
  28 #include "channel.h"
  29 #include "completion_queue.h"
  30 #include "grpc/grpc.h"
  31 #include "grpc/grpc_security.h"
  32 #include "grpc/support/alloc.h"
  33 #include "grpc/support/log.h"
  34 #include "grpc/support/time.h"
  35 #include "slice.h"
  36 #include "timeval.h"
  37
  38 using std::unique_ptr;
  39 using std::shared_ptr;
  40 using std::vector;
  41
  42 namespace grpc {
  43 namespace node {
  44
  45 using Nan::Callback;
  46 using Nan::EscapableHandleScope;
  47 using Nan::HandleScope;
  48 using Nan::Maybe;
  49 using Nan::MaybeLocal;
  50 using Nan::ObjectWrap;
  51 using Nan::Persistent;
  52 using Nan::Utf8String;
  53
  54 using v8::Array;
  55 using v8::Boolean;
  56 using v8::Exception;
  57 using v8::External;
  58 using v8::Function;
  59 using v8::FunctionTemplate;
  60 using v8::Integer;
  61 using v8::Local;
  62 using v8::Number;
  63 using v8::Object;
  64 using v8::ObjectTemplate;
  65 using v8::Uint32;
  66 using v8::String;
  67 using v8::Value;
  68
  69 Callback *Call::constructor;
  70 Persistent<FunctionTemplate> Call::fun_tpl;
  71
  72 /**
  73  * Helper function for throwing errors with a grpc_call_error value.
  74  * Modified from the answer by Gus Goose to
  75  * http://stackoverflow.com/questions/31794200.
  76  */
  77 Local<Value> nanErrorWithCode(const char *msg, grpc_call_error code) {
  78   EscapableHandleScope scope;
  79   Local<Object> err = Nan::Error(msg).As<Object>();
  80   Nan::Set(err, Nan::New("code").ToLocalChecked(), Nan::New<Uint32>(code));
  81   return scope.Escape(err);
  82 }
  83
  84 bool CreateMetadataArray(Local<Object> metadata_obj, grpc_metadata_array *array) {
  85   HandleScope scope;
  86   Local<Value> metadata_value = (Nan::Get(metadata_obj, Nan::New("metadata").ToLocalChecked())).ToLocalChecked();
  87   if (!metadata_value->IsObject()) {
  88     return false;
  89   }
  90   Local<Object> metadata = Nan::To<Object>(metadata_value).ToLocalChecked();
  91   Local<Array> keys = Nan::GetOwnPropertyNames(metadata).ToLocalChecked();
  92   for (unsigned int i = 0; i < keys->Length(); i++) {
  93     Local<String> current_key =
  94         Nan::To<String>(Nan::Get(keys, i).ToLocalChecked()).ToLocalChecked();
  95     Local<Value> value_array = Nan::Get(metadata, current_key).ToLocalChecked();
  96     if (!value_array->IsArray()) {
  97       return false;
  98     }
  99     array->capacity += Local<Array>::Cast(value_array)->Length();
 100   }
 101   array->metadata = reinterpret_cast<grpc_metadata *>(
 102       gpr_zalloc(array->capacity * sizeof(grpc_metadata)));
 103   for (unsigned int i = 0; i < keys->Length(); i++) {
 104     Local<String> current_key(Nan::To<String>(Nan::Get(keys, i).ToLocalChecked()).ToLocalChecked());
 105     Local<Array> values =
 106         Local<Array>::Cast(Nan::Get(metadata, current_key).ToLocalChecked());
 107     grpc_slice key_slice = CreateSliceFromString(current_key);
 108     grpc_slice key_intern_slice = grpc_slice_intern(key_slice);
 109     grpc_slice_unref(key_slice);
 110     for (unsigned int j = 0; j < values->Length(); j++) {
 111       Local<Value> value = Nan::Get(values, j).ToLocalChecked();
 112       grpc_metadata *current = &array->metadata[array->count];
 113       current->key = key_intern_slice;
 114       // Only allow binary headers for "-bin" keys
 115       if (grpc_is_binary_header(key_intern_slice)) {
 116         if (::node::Buffer::HasInstance(value)) {
 117           current->value = CreateSliceFromBuffer(value);
 118         } else {
 119           return false;
 120         }
 121       } else {
 122         if (value->IsString()) {
 123           Local<String> string_value = Nan::To<String>(value).ToLocalChecked();
 124           current->value = CreateSliceFromString(string_value);
 125         } else {
 126           return false;
 127         }
 128       }
 129       array->count += 1;
 130     }
 131   }
 132   return true;
 133 }
 134
 135 void DestroyMetadataArray(grpc_metadata_array *array) {
 136   for (size_t i = 0; i < array->count; i++) {
 137     // Don't unref keys because they are interned
 138     grpc_slice_unref(array->metadata[i].value);
 139   }
 140   grpc_metadata_array_destroy(array);
 141 }
 142
 143 Local<Value> ParseMetadata(const grpc_metadata_array *metadata_array) {
 144   EscapableHandleScope scope;
 145   grpc_metadata *metadata_elements = metadata_array->metadata;
 146   size_t length = metadata_array->count;
 147   Local<Object> metadata_object = Nan::New<Object>();
 148   for (unsigned int i = 0; i < length; i++) {
 149     grpc_metadata *elem = &metadata_elements[i];
 150     // TODO(murgatroid99): Use zero-copy string construction instead
 151     Local<String> key_string = CopyStringFromSlice(elem->key);
 152     Local<Array> array;
 153     MaybeLocal<Value> maybe_array = Nan::Get(metadata_object, key_string);
 154     if (maybe_array.IsEmpty() || !maybe_array.ToLocalChecked()->IsArray()) {
 155       array = Nan::New<Array>(0);
 156       Nan::Set(metadata_object, key_string, array);
 157     } else {
 158       array = Local<Array>::Cast(maybe_array.ToLocalChecked());
 159     }
 160     if (grpc_is_binary_header(elem->key)) {
 161       Nan::Set(array, array->Length(), CopyBufferFromSlice(elem->value));
 162     } else {
 163       // TODO(murgatroid99): Use zero-copy string construction instead
 164       Nan::Set(array, array->Length(), CopyStringFromSlice(elem->value));
 165     }
 166   }
 167   Local<Object> result = Nan::New<Object>();
 168   Nan::Set(result, Nan::New("metadata").ToLocalChecked(), metadata_object);
 169   Nan::Set(result, Nan::New("flags").ToLocalChecked(), Nan::New<v8::Uint32>(0));
 170   return scope.Escape(result);
 171 }
 172
 173 Local<Value> Op::GetOpType() const {
 174   EscapableHandleScope scope;
 175   return scope.Escape(Nan::New(GetTypeString()).ToLocalChecked());
 176 }
 177
 178 Op::~Op() {}
 179
 180 class SendMetadataOp : public Op {
 181  public:
 182   SendMetadataOp() { grpc_metadata_array_init(&send_metadata); }
 183   ~SendMetadataOp() { DestroyMetadataArray(&send_metadata); }
 184   Local<Value> GetNodeValue() const {
 185     EscapableHandleScope scope;
 186     return scope.Escape(Nan::True());
 187   }
 188   bool ParseOp(Local<Value> value, grpc_op *out) {
 189     if (!value->IsObject()) {
 190       return false;
 191     }
 192     MaybeLocal<Object> maybe_metadata = Nan::To<Object>(value);
 193     if (maybe_metadata.IsEmpty()) {
 194       return false;
 195     }
 196     Local<Object> metadata_object = maybe_metadata.ToLocalChecked();
 197     MaybeLocal<Value> maybe_flag_value =
 198         Nan::Get(metadata_object, Nan::New("flags").ToLocalChecked());
 199     if (!maybe_flag_value.IsEmpty()) {
 200       Local<Value> flag_value = maybe_flag_value.ToLocalChecked();
 201       if (flag_value->IsUint32()) {
 202         Maybe<uint32_t> maybe_flag = Nan::To<uint32_t>(flag_value);
 203         out->flags |= maybe_flag.FromMaybe(0) & GRPC_INITIAL_METADATA_USED_MASK;
 204       }
 205     }
 206     if (!CreateMetadataArray(metadata_object, &send_metadata)) {
 207       return false;
 208     }
 209     out->data.send_initial_metadata.count = send_metadata.count;
 210     out->data.send_initial_metadata.metadata = send_metadata.metadata;
 211     return true;
 212   }
 213   bool IsFinalOp() { return false; }
 214   void OnComplete(bool success) {}
 215
 216  protected:
 217   std::string GetTypeString() const { return "send_metadata"; }
 218
 219  private:
 220   grpc_metadata_array send_metadata;
 221 };
 222
 223 class SendMessageOp : public Op {
 224  public:
 225   SendMessageOp() { send_message = NULL; }
 226   ~SendMessageOp() {
 227     if (send_message != NULL) {
 228       grpc_byte_buffer_destroy(send_message);
 229     }
 230   }
 231   Local<Value> GetNodeValue() const {
 232     EscapableHandleScope scope;
 233     return scope.Escape(Nan::True());
 234   }
 235   bool ParseOp(Local<Value> value, grpc_op *out) {
 236     if (!::node::Buffer::HasInstance(value)) {
 237       return false;
 238     }
 239     Local<Object> object_value = Nan::To<Object>(value).ToLocalChecked();
 240     MaybeLocal<Value> maybe_flag_value =
 241         Nan::Get(object_value, Nan::New("grpcWriteFlags").ToLocalChecked());
 242     if (!maybe_flag_value.IsEmpty()) {
 243       Local<Value> flag_value = maybe_flag_value.ToLocalChecked();
 244       if (flag_value->IsUint32()) {
 245         Maybe<uint32_t> maybe_flag = Nan::To<uint32_t>(flag_value);
 246         out->flags |= maybe_flag.FromMaybe(0) & GRPC_WRITE_USED_MASK;
 247       }
 248     }
 249     send_message = BufferToByteBuffer(value);
 250     out->data.send_message.send_message = send_message;
 251     return true;
 252   }
 253
 254   bool IsFinalOp() { return false; }
 255   void OnComplete(bool success) {}
 256
 257  protected:
 258   std::string GetTypeString() const { return "send_message"; }
 259
 260  private:
 261   grpc_byte_buffer *send_message;
 262 };
 263
 264 class SendClientCloseOp : public Op {
 265  public:
 266   Local<Value> GetNodeValue() const {
 267     EscapableHandleScope scope;
 268     return scope.Escape(Nan::True());
 269   }
 270
 271   bool ParseOp(Local<Value> value, grpc_op *out) { return true; }
 272   bool IsFinalOp() { return false; }
 273   void OnComplete(bool success) {}
 274
 275  protected:
 276   std::string GetTypeString() const { return "client_close"; }
 277 };
 278
 279 class SendServerStatusOp : public Op {
 280  public:
 281   SendServerStatusOp() {
 282     details = grpc_empty_slice();
 283     grpc_metadata_array_init(&status_metadata);
 284   }
 285   ~SendServerStatusOp() {
 286     grpc_slice_unref(details);
 287     DestroyMetadataArray(&status_metadata);
 288   }
 289   Local<Value> GetNodeValue() const {
 290     EscapableHandleScope scope;
 291     return scope.Escape(Nan::True());
 292   }
 293   bool ParseOp(Local<Value> value, grpc_op *out) {
 294     if (!value->IsObject()) {
 295       return false;
 296     }
 297     Local<Object> server_status = Nan::To<Object>(value).ToLocalChecked();
 298     MaybeLocal<Value> maybe_metadata =
 299         Nan::Get(server_status, Nan::New("metadata").ToLocalChecked());
 300     if (maybe_metadata.IsEmpty()) {
 301       return false;
 302     }
 303     if (!maybe_metadata.ToLocalChecked()->IsObject()) {
 304       return false;
 305     }
 306     Local<Object> metadata =
 307         Nan::To<Object>(maybe_metadata.ToLocalChecked()).ToLocalChecked();
 308     MaybeLocal<Value> maybe_code =
 309         Nan::Get(server_status, Nan::New("code").ToLocalChecked());
 310     if (maybe_code.IsEmpty()) {
 311       return false;
 312     }
 313     if (!maybe_code.ToLocalChecked()->IsUint32()) {
 314       return false;
 315     }
 316     uint32_t code = Nan::To<uint32_t>(maybe_code.ToLocalChecked()).FromJust();
 317     MaybeLocal<Value> maybe_details =
 318         Nan::Get(server_status, Nan::New("details").ToLocalChecked());
 319     if (maybe_details.IsEmpty()) {
 320       return false;
 321     }
 322     if (!maybe_details.ToLocalChecked()->IsString()) {
 323       return false;
 324     }
 325     Local<String> details =
 326         Nan::To<String>(maybe_details.ToLocalChecked()).ToLocalChecked();
 327     if (!CreateMetadataArray(metadata, &status_metadata)) {
 328       return false;
 329     }
 330     out->data.send_status_from_server.trailing_metadata_count =
 331         status_metadata.count;
 332     out->data.send_status_from_server.trailing_metadata =
 333         status_metadata.metadata;
 334     out->data.send_status_from_server.status =
 335         static_cast<grpc_status_code>(code);
 336     this->details = CreateSliceFromString(details);
 337     out->data.send_status_from_server.status_details = &this->details;
 338     return true;
 339   }
 340   bool IsFinalOp() { return true; }
 341   void OnComplete(bool success) {}
 342
 343  protected:
 344   std::string GetTypeString() const { return "send_status"; }
 345
 346  private:
 347   grpc_slice details;
 348   grpc_metadata_array status_metadata;
 349 };
 350
 351 class GetMetadataOp : public Op {
 352  public:
 353   GetMetadataOp() { grpc_metadata_array_init(&recv_metadata); }
 354
 355   ~GetMetadataOp() { grpc_metadata_array_destroy(&recv_metadata); }
 356
 357   Local<Value> GetNodeValue() const {
 358     EscapableHandleScope scope;
 359     return scope.Escape(ParseMetadata(&recv_metadata));
 360   }
 361
 362   bool ParseOp(Local<Value> value, grpc_op *out) {
 363     out->data.recv_initial_metadata.recv_initial_metadata = &recv_metadata;
 364     return true;
 365   }
 366   bool IsFinalOp() { return false; }
 367   void OnComplete(bool success) {}
 368
 369  protected:
 370   std::string GetTypeString() const { return "metadata"; }
 371
 372  private:
 373   grpc_metadata_array recv_metadata;
 374 };
 375
 376 class ReadMessageOp : public Op {
 377  public:
 378   ReadMessageOp() { recv_message = NULL; }
 379   ~ReadMessageOp() {
 380     if (recv_message != NULL) {
 381       grpc_byte_buffer_destroy(recv_message);
 382     }
 383   }
 384   Local<Value> GetNodeValue() const {
 385     EscapableHandleScope scope;
 386     return scope.Escape(ByteBufferToBuffer(recv_message));
 387   }
 388
 389   bool ParseOp(Local<Value> value, grpc_op *out) {
 390     out->data.recv_message.recv_message = &recv_message;
 391     return true;
 392   }
 393   bool IsFinalOp() { return false; }
 394   void OnComplete(bool success) {}
 395
 396  protected:
 397   std::string GetTypeString() const { return "read"; }
 398
 399  private:
 400   grpc_byte_buffer *recv_message;
 401 };
 402
 403 class ClientStatusOp : public Op {
 404  public:
 405   ClientStatusOp() {
 406     grpc_metadata_array_init(&metadata_array);
 407     status_details = grpc_empty_slice();
 408   }
 409
 410   ~ClientStatusOp() {
 411     grpc_metadata_array_destroy(&metadata_array);
 412     grpc_slice_unref(status_details);
 413   }
 414
 415   bool ParseOp(Local<Value> value, grpc_op *out) {
 416     out->data.recv_status_on_client.trailing_metadata = &metadata_array;
 417     out->data.recv_status_on_client.status = &status;
 418     out->data.recv_status_on_client.status_details = &status_details;
 419     return true;
 420   }
 421
 422   Local<Value> GetNodeValue() const {
 423     EscapableHandleScope scope;
 424     Local<Object> status_obj = Nan::New<Object>();
 425     Nan::Set(status_obj, Nan::New("code").ToLocalChecked(),
 426              Nan::New<Number>(status));
 427     Nan::Set(status_obj, Nan::New("details").ToLocalChecked(),
 428              CopyStringFromSlice(status_details));
 429     Nan::Set(status_obj, Nan::New("metadata").ToLocalChecked(),
 430              ParseMetadata(&metadata_array));
 431     return scope.Escape(status_obj);
 432   }
 433   bool IsFinalOp() { return true; }
 434   void OnComplete(bool success) {}
 435
 436  protected:
 437   std::string GetTypeString() const { return "status"; }
 438
 439  private:
 440   grpc_metadata_array metadata_array;
 441   grpc_status_code status;
 442   grpc_slice status_details;
 443 };
 444
 445 class ServerCloseResponseOp : public Op {
 446  public:
 447   Local<Value> GetNodeValue() const {
 448     EscapableHandleScope scope;
 449     return scope.Escape(Nan::New<Boolean>(cancelled));
 450   }
 451
 452   bool ParseOp(Local<Value> value, grpc_op *out) {
 453     out->data.recv_close_on_server.cancelled = &cancelled;
 454     return true;
 455   }
 456   bool IsFinalOp() { return false; }
 457   void OnComplete(bool success) {}
 458
 459  protected:
 460   std::string GetTypeString() const { return "cancelled"; }
 461
 462  private:
 463   int cancelled;
 464 };
 465
 466 tag::tag(Callback *callback, OpVec *ops, Call *call, Local<Value> call_value)
 467     : callback(callback),
 468       async_resource(NULL),
 469       ops(ops),
 470       call(call) {
 471   HandleScope scope;
 472   async_resource = new Nan::AsyncResource("grpc:tag");  // Needs handle scope.
 473   call_persist.Reset(call_value);
 474 }
 475
 476 tag::~tag() {
 477   delete callback;
 478   delete async_resource;
 479   delete ops;
 480 }
 481
 482 void CompleteTag(void *tag, const char *error_message) {
 483   HandleScope scope;
 484   struct tag *tag_struct = reinterpret_cast<struct tag *>(tag);
 485   Callback *callback = tag_struct->callback;
 486   if (error_message == NULL) {
 487     Local<Object> tag_obj = Nan::New<Object>();
 488     for (vector<unique_ptr<Op> >::iterator it = tag_struct->ops->begin();
 489          it != tag_struct->ops->end(); ++it) {
 490       Op *op_ptr = it->get();
 491       Nan::Set(tag_obj, op_ptr->GetOpType(), op_ptr->GetNodeValue());
 492     }
 493     Local<Value> argv[] = {Nan::Null(), tag_obj};
 494     callback->Call(2, argv, tag_struct->async_resource);
 495   } else {
 496     Local<Value> argv[] = {Nan::Error(error_message)};
 497     callback->Call(1, argv, tag_struct->async_resource);
 498   }
 499   bool success = (error_message == NULL);
 500   bool is_final_op = false;
 501   for (vector<unique_ptr<Op> >::iterator it = tag_struct->ops->begin();
 502        it != tag_struct->ops->end(); ++it) {
 503     Op *op_ptr = it->get();
 504     op_ptr->OnComplete(success);
 505     if (op_ptr->IsFinalOp()) {
 506       is_final_op = true;
 507     }
 508   }
 509   if (tag_struct->call == NULL) {
 510     return;
 511   }
 512   tag_struct->call->CompleteBatch(is_final_op);
 513 }
 514
 515 void DestroyTag(void *tag) {
 516   struct tag *tag_struct = reinterpret_cast<struct tag *>(tag);
 517   delete tag_struct;
 518 }
 519
 520 void Call::DestroyCall() {
 521   if (this->wrapped_call != NULL) {
 522     grpc_call_unref(this->wrapped_call);
 523     this->wrapped_call = NULL;
 524   }
 525 }
 526
 527 Call::Call(grpc_call *call)
 528     : wrapped_call(call), pending_batches(0), has_final_op_completed(false) {
 529   peer = grpc_call_get_peer(call);
 530 }
 531
 532 Call::~Call() {
 533   DestroyCall();
 534   gpr_free(peer);
 535 }
 536
 537 void Call::Init(Local<Object> exports) {
 538   HandleScope scope;
 539   Local<FunctionTemplate> tpl = Nan::New<FunctionTemplate>(New);
 540   tpl->SetClassName(Nan::New("Call").ToLocalChecked());
 541   tpl->InstanceTemplate()->SetInternalFieldCount(1);
 542   Nan::SetPrototypeMethod(tpl, "startBatch", StartBatch);
 543   Nan::SetPrototypeMethod(tpl, "cancel", Cancel);
 544   Nan::SetPrototypeMethod(tpl, "cancelWithStatus", CancelWithStatus);
 545   Nan::SetPrototypeMethod(tpl, "getPeer", GetPeer);
 546   Nan::SetPrototypeMethod(tpl, "setCredentials", SetCredentials);
 547   fun_tpl.Reset(tpl);
 548   Local<Function> ctr = Nan::GetFunction(tpl).ToLocalChecked();
 549   Nan::Set(exports, Nan::New("Call").ToLocalChecked(), ctr);
 550   constructor = new Callback(ctr);
 551 }
 552
 553 bool Call::HasInstance(Local<Value> val) {
 554   HandleScope scope;
 555   return Nan::New(fun_tpl)->HasInstance(val);
 556 }
 557
 558 grpc_call *Call::GetWrappedCall() { return this->wrapped_call; }
 559
 560 Local<Value> Call::WrapStruct(grpc_call *call) {
 561   EscapableHandleScope scope;
 562   if (call == NULL) {
 563     return scope.Escape(Nan::Null());
 564   }
 565   const int argc = 1;
 566   Local<Value> argv[argc] = {
 567       Nan::New<External>(reinterpret_cast<void *>(call))};
 568   MaybeLocal<Object> maybe_instance =
 569       Nan::NewInstance(constructor->GetFunction(), argc, argv);
 570   if (maybe_instance.IsEmpty()) {
 571     return scope.Escape(Nan::Null());
 572   } else {
 573     return scope.Escape(maybe_instance.ToLocalChecked());
 574   }
 575 }
 576
 577 void Call::CompleteBatch(bool is_final_op) {
 578   if (is_final_op) {
 579     this->has_final_op_completed = true;
 580   }
 581   this->pending_batches--;
 582   if (this->has_final_op_completed && this->pending_batches == 0) {
 583     this->DestroyCall();
 584   }
 585 }
 586
 587 NAN_METHOD(Call::New) {
 588   /* Arguments:
 589    * 0: Channel to make the call on
 590    * 1: Method
 591    * 2: Deadline
 592    * 3: host
 593    * 4: parent Call
 594    * 5: propagation flags
 595    */
 596   if (info.IsConstructCall()) {
 597     Call *call;
 598     if (!info[0]->IsExternal()) {
 599       return Nan::ThrowTypeError(
 600         "Call can only be created with Channel.createCall");
 601     }
 602     Local<External> ext = info[0].As<External>();
 603     // This option is used for wrapping an existing call
 604     grpc_call *call_value = reinterpret_cast<grpc_call *>(ext->Value());
 605     call = new Call(call_value);
 606     call->Wrap(info.This());
 607     info.GetReturnValue().Set(info.This());
 608     return;
 609   } else {
 610     return Nan::ThrowTypeError(
 611       "Call can only be created with Channel.createCall");
 612   }
 613 }
 614
 615 NAN_METHOD(Call::StartBatch) {
 616   if (!Call::HasInstance(info.This())) {
 617     return Nan::ThrowTypeError("startBatch can only be called on Call objects");
 618   }
 619   if (!info[0]->IsObject()) {
 620     return Nan::ThrowError("startBatch's first argument must be an object");
 621   }
 622   if (!info[1]->IsFunction()) {
 623     return Nan::ThrowError("startBatch's second argument must be a callback");
 624   }
 625   Local<Function> callback_func = info[1].As<Function>();
 626   Call *call = ObjectWrap::Unwrap<Call>(info.This());
 627   if (call->wrapped_call == NULL) {
 628     /* This implies that the call has completed and has been destroyed. To
 629      * emulate
 630      * previous behavior, we should call the callback immediately with an error,
 631      * as though the batch had failed in core */
 632     Local<Value> argv[] = {
 633         Nan::Error("The async function failed because the call has completed")};
 634     Nan::Call(callback_func, Nan::New<Object>(), 1, argv);
 635     return;
 636   }
 637   Local<Object> obj = Nan::To<Object>(info[0]).ToLocalChecked();
 638   Local<Array> keys = Nan::GetOwnPropertyNames(obj).ToLocalChecked();
 639   size_t nops = keys->Length();
 640   vector<grpc_op> ops(nops);
 641   unique_ptr<OpVec> op_vector(new OpVec());
 642   for (unsigned int i = 0; i < nops; i++) {
 643     unique_ptr<Op> op;
 644     MaybeLocal<Value> maybe_key = Nan::Get(keys, i);
 645     if (maybe_key.IsEmpty() || (!maybe_key.ToLocalChecked()->IsUint32())) {
 646       return Nan::ThrowError(
 647           "startBatch's first argument's keys must be integers");
 648     }
 649     uint32_t type = Nan::To<uint32_t>(maybe_key.ToLocalChecked()).FromJust();
 650     ops[i].op = static_cast<grpc_op_type>(type);
 651     ops[i].flags = 0;
 652     ops[i].reserved = NULL;
 653     switch (type) {
 654       case GRPC_OP_SEND_INITIAL_METADATA:
 655         op.reset(new SendMetadataOp());
 656         break;
 657       case GRPC_OP_SEND_MESSAGE:
 658         op.reset(new SendMessageOp());
 659         break;
 660       case GRPC_OP_SEND_CLOSE_FROM_CLIENT:
 661         op.reset(new SendClientCloseOp());
 662         break;
 663       case GRPC_OP_SEND_STATUS_FROM_SERVER:
 664         op.reset(new SendServerStatusOp());
 665         break;
 666       case GRPC_OP_RECV_INITIAL_METADATA:
 667         op.reset(new GetMetadataOp());
 668         break;
 669       case GRPC_OP_RECV_MESSAGE:
 670         op.reset(new ReadMessageOp());
 671         break;
 672       case GRPC_OP_RECV_STATUS_ON_CLIENT:
 673         op.reset(new ClientStatusOp());
 674         break;
 675       case GRPC_OP_RECV_CLOSE_ON_SERVER:
 676         op.reset(new ServerCloseResponseOp());
 677         break;
 678       default:
 679         return Nan::ThrowError("Argument object had an unrecognized key");
 680     }
 681     if (!op->ParseOp(Nan::Get(obj, type).ToLocalChecked(), &ops[i])) {
 682       return Nan::ThrowTypeError("Incorrectly typed arguments to startBatch");
 683     }
 684     op_vector->push_back(std::move(op));
 685   }
 686   Callback *callback = new Callback(callback_func);
 687   grpc_call_error error = grpc_call_start_batch(
 688       call->wrapped_call, &ops[0], nops,
 689       new struct tag(callback, op_vector.release(), call, info.This()), NULL);
 690   if (error != GRPC_CALL_OK) {
 691     return Nan::ThrowError(nanErrorWithCode("startBatch failed", error));
 692   }
 693   call->pending_batches++;
 694   CompletionQueueNext();
 695 }
 696
 697 NAN_METHOD(Call::Cancel) {
 698   if (!Call::HasInstance(info.This())) {
 699     return Nan::ThrowTypeError("cancel can only be called on Call objects");
 700   }
 701   Call *call = ObjectWrap::Unwrap<Call>(info.This());
 702   if (call->wrapped_call == NULL) {
 703     /* Cancel is supposed to be idempotent. If the call has already finished,
 704      * cancel should just complete silently */
 705     return;
 706   }
 707   grpc_call_error error = grpc_call_cancel(call->wrapped_call, NULL);
 708   if (error != GRPC_CALL_OK) {
 709     return Nan::ThrowError(nanErrorWithCode("cancel failed", error));
 710   }
 711 }
 712
 713 NAN_METHOD(Call::CancelWithStatus) {
 714   Nan::HandleScope scope;
 715   if (!HasInstance(info.This())) {
 716     return Nan::ThrowTypeError("cancel can only be called on Call objects");
 717   }
 718   if (!info[0]->IsUint32()) {
 719     return Nan::ThrowTypeError(
 720         "cancelWithStatus's first argument must be a status code");
 721   }
 722   if (!info[1]->IsString()) {
 723     return Nan::ThrowTypeError(
 724         "cancelWithStatus's second argument must be a string");
 725   }
 726   Call *call = ObjectWrap::Unwrap<Call>(info.This());
 727   if (call->wrapped_call == NULL) {
 728     /* Cancel is supposed to be idempotent. If the call has already finished,
 729      * cancel should just complete silently */
 730     return;
 731   }
 732   grpc_status_code code =
 733       static_cast<grpc_status_code>(Nan::To<uint32_t>(info[0]).FromJust());
 734   if (code == GRPC_STATUS_OK) {
 735     return Nan::ThrowRangeError(
 736         "cancelWithStatus cannot be called with OK status");
 737   }
 738   Utf8String details(info[1]);
 739   grpc_call_cancel_with_status(call->wrapped_call, code, *details, NULL);
 740 }
 741
 742 NAN_METHOD(Call::GetPeer) {
 743   Nan::HandleScope scope;
 744   if (!HasInstance(info.This())) {
 745     return Nan::ThrowTypeError("getPeer can only be called on Call objects");
 746   }
 747   Call *call = ObjectWrap::Unwrap<Call>(info.This());
 748   Local<Value> peer_value = Nan::New(call->peer).ToLocalChecked();
 749   info.GetReturnValue().Set(peer_value);
 750 }
 751
 752 NAN_METHOD(Call::SetCredentials) {
 753   Nan::HandleScope scope;
 754   if (!HasInstance(info.This())) {
 755     return Nan::ThrowTypeError(
 756         "setCredentials can only be called on Call objects");
 757   }
 758   if (!CallCredentials::HasInstance(info[0])) {
 759     return Nan::ThrowTypeError(
 760         "setCredentials' first argument must be a CallCredentials");
 761   }
 762   Call *call = ObjectWrap::Unwrap<Call>(info.This());
 763   if (call->wrapped_call == NULL) {
 764     return Nan::ThrowError(
 765         "Cannot set credentials on a call that has already started");
 766   }
 767   CallCredentials *creds_object = ObjectWrap::Unwrap<CallCredentials>(
 768       Nan::To<Object>(info[0]).ToLocalChecked());
 769   grpc_call_credentials *creds = creds_object->GetWrappedCredentials();
 770   grpc_call_error error = GRPC_CALL_ERROR;
 771   if (creds) {
 772     error = grpc_call_set_credentials(call->wrapped_call, creds);
 773   }
 774   info.GetReturnValue().Set(Nan::New<Uint32>(error));
 775 }
 776
 777 }  // namespace node
 778 }  // namespace grpc