Emit proper status when failing to parse server stream

src/node/ext/call.cc

@@ -461,6 +461,9 @@ void Call::Init(Handle<Object> exports) {
                           NanNew<FunctionTemplate>(StartBatch)->GetFunction());
   NanSetPrototypeTemplate(tpl, "cancel",
                           NanNew<FunctionTemplate>(Cancel)->GetFunction());
+  NanSetPrototypeTemplate(
+      tpl, "cancelWithStatus",
+      NanNew<FunctionTemplate>(CancelWithStatus)->GetFunction());
   NanSetPrototypeTemplate(tpl, "getPeer",
                           NanNew<FunctionTemplate>(GetPeer)->GetFunction());
   NanAssignPersistent(fun_tpl, tpl);
@@ -643,6 +646,26 @@ NAN_METHOD(Call::Cancel) {
   NanReturnUndefined();
 }
 
+NAN_METHOD(Call::CancelWithStatus) {
+  NanScope();
+  if (!HasInstance(args.This())) {
+    return NanThrowTypeError("cancel can only be called on Call objects");
+  }
+  if (!args[0]->IsUint32()) {
+    return NanThrowTypeError(
+        "cancelWithStatus's first argument must be a status code");
+  }
+  if (!args[1]->IsString()) {
+    return NanThrowTypeError(
+        "cancelWithStatus's second argument must be a string");
+  }
+  Call *call = ObjectWrap::Unwrap<Call>(args.This());
+  grpc_status_code code = static_cast<grpc_status_code>(args[0]->Uint32Value());
+  NanUtf8String details(args[0]);
+  grpc_call_cancel_with_status(call->wrapped_call, code, *details, NULL);
+  NanReturnUndefined();
+}
+
 NAN_METHOD(Call::GetPeer) {
   NanScope();
   if (!HasInstance(args.This())) {

src/node/ext/call.h

@@ -133,6 +133,7 @@ class Call : public ::node::ObjectWrap {
   static NAN_METHOD(New);
   static NAN_METHOD(StartBatch);
   static NAN_METHOD(Cancel);
+  static NAN_METHOD(CancelWithStatus);
   static NAN_METHOD(GetPeer);
   static NanCallback *constructor;
   // Used for typechecking instances of this javascript class

src/node/src/client.js

@@ -146,7 +146,8 @@ function _read(size) {
     try {
       deserialized = self.deserialize(data);
     } catch (e) {
-      self.cancel();
+      self.call.cancelWithStatus(grpc.status.INTERNAL,
+                                 'Failed to parse server response');
     }
     if (self.push(deserialized) && data !== null) {
       var read_batch = {};