grpc
/
grpc-node
mirror of https://github.com/grpc/grpc-node.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge github.com:grpc/grpc into cpparena

This commit is contained in:
Craig Tiller 2017-04-14 13:14:10 -07:00
parent 857a8fd02b c5cb93c6dd
commit 991f0b18c2
11 changed files with 144 additions and 73 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

69
ext/call.cc
View File

@ -209,6 +209,7 @@ class SendMetadataOp : public Op {
return true;
}
bool IsFinalOp() { return false; }
void OnComplete(bool success) {}
protected:
std::string GetTypeString() const { return "send_metadata"; }
@ -247,8 +248,14 @@ class SendMessageOp : public Op {
out->data.send_message.send_message = send_message;
return true;
}
<<<<<<< HEAD
bool IsFinalOp() { return false; }
=======
bool IsFinalOp() { return false; }
void OnComplete(bool success) {}
>>>>>>> e412a180602753972ac496560322e224a5db987f
protected:
std::string GetTypeString() const { return "send_message"; }
@ -262,9 +269,16 @@ class SendClientCloseOp : public Op {
EscapableHandleScope scope;
return scope.Escape(Nan::True());
}
<<<<<<< HEAD
bool ParseOp(Local<Value> value, grpc_op *out) { return true; }
bool IsFinalOp() { return false; }
=======
bool ParseOp(Local<Value> value, grpc_op *out) { return true; }
bool IsFinalOp() { return false; }
void OnComplete(bool success) {}
>>>>>>> e412a180602753972ac496560322e224a5db987f
protected:
std::string GetTypeString() const { return "client_close"; }
};
@ -327,8 +341,14 @@ class SendServerStatusOp : public Op {
out->data.send_status_from_server.status_details = &this->details;
return true;
}
<<<<<<< HEAD
bool IsFinalOp() { return true; }
=======
bool IsFinalOp() { return true; }
void OnComplete(bool success) {}
>>>>>>> e412a180602753972ac496560322e224a5db987f
protected:
std::string GetTypeString() const { return "send_status"; }
@ -352,7 +372,12 @@ class GetMetadataOp : public Op {
out->data.recv_initial_metadata.recv_initial_metadata = &recv_metadata;
return true;
}
<<<<<<< HEAD
bool IsFinalOp() { return false; }
=======
bool IsFinalOp() { return false; }
void OnComplete(bool success) {}
>>>>>>> e412a180602753972ac496560322e224a5db987f
protected:
std::string GetTypeString() const { return "metadata"; }
@ -378,7 +403,12 @@ class ReadMessageOp : public Op {
out->data.recv_message.recv_message = &recv_message;
return true;
}
<<<<<<< HEAD
bool IsFinalOp() { return false; }
=======
bool IsFinalOp() { return false; }
void OnComplete(bool success) {}
>>>>>>> e412a180602753972ac496560322e224a5db987f
protected:
std::string GetTypeString() const { return "read"; }
@ -411,8 +441,14 @@ class ClientStatusOp : public Op {
ParseMetadata(&metadata_array));
return scope.Escape(status_obj);
}
<<<<<<< HEAD
bool IsFinalOp() { return true; }
=======
bool IsFinalOp() { return true; }
void OnComplete(bool success) {}
>>>>>>> e412a180602753972ac496560322e224a5db987f
protected:
std::string GetTypeString() const { return "status"; }
@ -433,7 +469,12 @@ class ServerCloseResponseOp : public Op {
out->data.recv_close_on_server.cancelled = &cancelled;
return true;
}
<<<<<<< HEAD
bool IsFinalOp() { return false; }
=======
bool IsFinalOp() { return false; }
void OnComplete(bool success) {}
>>>>>>> e412a180602753972ac496560322e224a5db987f
protected:
std::string GetTypeString() const { return "cancelled"; }
@ -453,36 +494,36 @@ tag::~tag() {
delete ops;
}
Local<Value> GetTagNodeValue(void *tag) {
EscapableHandleScope scope;
void CompleteTag(void *tag, const char *error_message) {
HandleScope scope;
struct tag *tag_struct = reinterpret_cast<struct tag *>(tag);
Callback *callback = tag_struct->callback;
if (error_message == NULL) {
Local<Object> tag_obj = Nan::New<Object>();
for (vector<unique_ptr<Op> >::iterator it = tag_struct->ops->begin();
it != tag_struct->ops->end(); ++it) {
Op *op_ptr = it->get();
Nan::Set(tag_obj, op_ptr->GetOpType(), op_ptr->GetNodeValue());
}
return scope.Escape(tag_obj);
Local<Value> argv[] = {Nan::Null(), tag_obj};
callback->Call(2, argv);
} else {
Local<Value> argv[] = {Nan::Error(error_message)};
callback->Call(1, argv);
}
Callback *GetTagCallback(void *tag) {
struct tag *tag_struct = reinterpret_cast<struct tag *>(tag);
return tag_struct->callback;
}
void CompleteTag(void *tag) {
struct tag *tag_struct = reinterpret_cast<struct tag *>(tag);
bool success = (error_message == NULL);
bool is_final_op = false;
if (tag_struct->call == NULL) {
return;
}
for (vector<unique_ptr<Op> >::iterator it = tag_struct->ops->begin();
it != tag_struct->ops->end(); ++it) {
Op *op_ptr = it->get();
op_ptr->OnComplete(success);
if (op_ptr->IsFinalOp()) {
is_final_op = true;
}
}
if (tag_struct->call == NULL) {
return;
}
tag_struct->call->CompleteBatch(is_final_op);
}

7
ext/call.h
View File

@ -106,6 +106,7 @@ class Op {
virtual ~Op();
v8::Local<v8::Value> GetOpType() const;
virtual bool IsFinalOp() = 0;
virtual void OnComplete(bool success) = 0;
protected:
virtual std::string GetTypeString() const = 0;
@ -123,13 +124,9 @@ struct tag {
call_persist;
};
v8::Local<v8::Value> GetTagNodeValue(void *tag);
Nan::Callback *GetTagCallback(void *tag);
void DestroyTag(void *tag);
void CompleteTag(void *tag);
void CompleteTag(void *tag, const char *error_message);
} // namespace node
} // namespace grpc

25
ext/completion_queue_threadpool.cc
View File

@ -34,14 +34,14 @@
/* I don't like using #ifndef, but I don't see a better way to do this */
#ifndef GRPC_UV
#include <node.h>
#include <nan.h>
#include <node.h>
#include "call.h"
#include "completion_queue.h"
#include "grpc/grpc.h"
#include "grpc/support/log.h"
#include "grpc/support/time.h"
#include "completion_queue.h"
#include "call.h"
namespace grpc {
namespace node {
@ -111,8 +111,8 @@ CompletionQueueAsyncWorker::CompletionQueueAsyncWorker()
CompletionQueueAsyncWorker::~CompletionQueueAsyncWorker() {}
void CompletionQueueAsyncWorker::Execute() {
result =
grpc_completion_queue_next(queue, gpr_inf_future(GPR_CLOCK_REALTIME), NULL);
result = grpc_completion_queue_next(queue, gpr_inf_future(GPR_CLOCK_REALTIME),
NULL);
if (!result.success) {
SetErrorMessage("The async function encountered an error");
}
@ -141,16 +141,14 @@ void CompletionQueueAsyncWorker::Init(Local<Object> exports) {
Nan::HandleScope scope;
current_threads = 0;
waiting_next_calls = 0;
queue = grpc_completion_queue_create(NULL);
queue = grpc_completion_queue_create_for_next(NULL);
}
void CompletionQueueAsyncWorker::HandleOKCallback() {
Nan::HandleScope scope;
current_threads -= 1;
TryAddWorker();
Nan::Callback *callback = GetTagCallback(result.tag);
Local<Value> argv[] = {Nan::Null(), GetTagNodeValue(result.tag)};
callback->Call(2, argv);
CompleteTag(result.tag, NULL);
DestroyTag(result.tag);
}
@ -159,10 +157,7 @@ void CompletionQueueAsyncWorker::HandleErrorCallback() {
Nan::HandleScope scope;
current_threads -= 1;
TryAddWorker();
Nan::Callback *callback = GetTagCallback(result.tag);
Local<Value> argv[] = {Nan::Error(ErrorMessage())};
callback->Call(1, argv);
CompleteTag(result.tag, ErrorMessage());
DestroyTag(result.tag);
}
@ -173,9 +168,7 @@ grpc_completion_queue *GetCompletionQueue() {
return CompletionQueueAsyncWorker::GetQueue();
}
void CompletionQueueNext() {
CompletionQueueAsyncWorker::Next();
}
void CompletionQueueNext() { CompletionQueueAsyncWorker::Next(); }
void CompletionQueueInit(Local<Object> exports) {
CompletionQueueAsyncWorker::Init(exports);

28
ext/completion_queue_uv.cc
View File

@ -33,10 +33,10 @@
#ifdef GRPC_UV
#include <uv.h>
#include <node.h>
#include <v8.h>
#include <grpc/grpc.h>
#include <node.h>
#include <uv.h>
#include <v8.h>
#include "call.h"
#include "completion_queue.h"
@ -57,21 +57,17 @@ void drain_completion_queue(uv_prepare_t *handle) {
grpc_event event;
(void)handle;
do {
event = grpc_completion_queue_next(
queue, gpr_inf_past(GPR_CLOCK_MONOTONIC), NULL);
event = grpc_completion_queue_next(queue, gpr_inf_past(GPR_CLOCK_MONOTONIC),
NULL);
if (event.type == GRPC_OP_COMPLETE) {
Nan::Callback *callback = grpc::node::GetTagCallback(event.tag);
const char *error_message;
if (event.success) {
Local<Value> argv[] = {Nan::Null(),
grpc::node::GetTagNodeValue(event.tag)};
callback->Call(2, argv);
error_message = NULL;
} else {
Local<Value> argv[] = {Nan::Error(
"The async function encountered an error")};
callback->Call(1, argv);
error_message = "The async function encountered an error";
}
grpc::node::CompleteTag(event.tag);
CompleteTag(event.tag, error_message);
grpc::node::DestroyTag(event.tag);
pending_batches--;
if (pending_batches == 0) {
@ -81,9 +77,7 @@ void drain_completion_queue(uv_prepare_t *handle) {
} while (event.type != GRPC_QUEUE_TIMEOUT);
}
grpc_completion_queue *GetCompletionQueue() {
return queue;
}
grpc_completion_queue *GetCompletionQueue() { return queue; }
void CompletionQueueNext() {
if (pending_batches == 0) {
@ -94,7 +88,7 @@ void CompletionQueueNext() {
}
void CompletionQueueInit(Local<Object> exports) {
queue = grpc_completion_queue_create(NULL);
queue = grpc_completion_queue_create_for_next(NULL);
uv_prepare_init(uv_default_loop(), &prepare);
pending_batches = 0;
}

45
ext/server.cc
View File

@ -117,6 +117,8 @@ class NewCallOp : public Op {
bool IsFinalOp() {
return false;
}
void OnComplete(bool success) {
}
grpc_call *call;
grpc_call_details details;
@ -126,6 +128,34 @@ class NewCallOp : public Op {
std::string GetTypeString() const { return "new_call"; }
};
class TryShutdownOp: public Op {
public:
TryShutdownOp(Server *server, Local<Value> server_value) : server(server) {
server_persist.Reset(server_value);
}
Local<Value> GetNodeValue() const {
EscapableHandleScope scope;
return scope.Escape(Nan::New(server_persist));
}
bool ParseOp(Local<Value> value, grpc_op *out) {
return true;
}
bool IsFinalOp() {
return false;
}
void OnComplete(bool success) {
if (success) {
server->DestroyWrappedServer();
}
}
protected:
std::string GetTypeString() const { return "try_shutdown"; }
private:
Server *server;
Nan::Persistent<v8::Value, Nan::CopyablePersistentTraits<v8::Value>>
server_persist;
};
void Server::Init(Local<Object> exports) {
HandleScope scope;
Local<FunctionTemplate> tpl = Nan::New<FunctionTemplate>(New);
@ -147,6 +177,13 @@ bool Server::HasInstance(Local<Value> val) {
return Nan::New(fun_tpl)->HasInstance(val);
}
void Server::DestroyWrappedServer() {
if (this->wrapped_server != NULL) {
grpc_server_destroy(this->wrapped_server);
this->wrapped_server = NULL;
}
}
NAN_METHOD(Server::New) {
/* If this is not a constructor call, make a constructor call and return
the result */
@ -242,7 +279,15 @@ NAN_METHOD(Server::TryShutdown) {
return Nan::ThrowTypeError("tryShutdown can only be called on a Server");
}
Server *server = ObjectWrap::Unwrap<Server>(info.This());
if (server->wrapped_server == NULL) {
// Server is already shut down. Call callback immediately.
Nan::Callback callback(info[0].As<Function>());
callback.Call(0, {});
return;
}
TryShutdownOp *op = new TryShutdownOp(server, info.This());
unique_ptr<OpVec> ops(new OpVec());
ops->push_back(unique_ptr<Op>(op));
grpc_server_shutdown_and_notify(
server->wrapped_server, GetCompletionQueue(),
new struct tag(new Nan::Callback(info[0].As<Function>()), ops.release(),

2
ext/server.h
View File

@ -53,6 +53,8 @@ class Server : public Nan::ObjectWrap {
JavaScript constructor */
static bool HasInstance(v8::Local<v8::Value> val);
void DestroyWrappedServer();
private:
explicit Server(grpc_server *server);
~Server();

4
ext/server_generic.cc
View File

@ -35,8 +35,8 @@
#include "server.h"
#include <node.h>
#include <nan.h>
#include <node.h>
#include "grpc/grpc.h"
#include "grpc/support/time.h"
@ -44,7 +44,7 @@ namespace grpc {
namespace node {
Server::Server(grpc_server *server) : wrapped_server(server) {
shutdown_queue = grpc_completion_queue_create(NULL);
shutdown_queue = grpc_completion_queue_create_for_pluck(NULL);
grpc_server_register_non_listening_completion_queue(server, shutdown_queue,
NULL);
}

15
ext/server_uv.cc
View File

@ -67,7 +67,7 @@ class ServerShutdownOp : public Op {
}
Local<Value> GetNodeValue() const {
return Nan::New<External>(reinterpret_cast<void *>(server));
return Nan::Null();
}
bool ParseOp(Local<Value> value, grpc_op *out) {
@ -76,6 +76,11 @@ class ServerShutdownOp : public Op {
bool IsFinalOp() {
return false;
}
void OnComplete(bool success) {
/* Because cancel_all_calls was called, we assume that shutdown_and_notify
completes successfully */
grpc_server_destroy(server);
}
grpc_server *server;
@ -94,16 +99,10 @@ NAN_METHOD(ServerShutdownCallback) {
if (!info[0]->IsNull()) {
return Nan::ThrowError("forceShutdown failed somehow");
}
MaybeLocal<Object> maybe_result = Nan::To<Object>(info[1]);
Local<Object> result = maybe_result.ToLocalChecked();
Local<Value> server_val = Nan::Get(
result, Nan::New("shutdown").ToLocalChecked()).ToLocalChecked();
Local<External> server_extern = server_val.As<External>();
grpc_server *server = reinterpret_cast<grpc_server *>(server_extern->Value());
grpc_server_destroy(server);
}
void Server::ShutdownServer() {
Nan::HandleScope scope;
if (this->wrapped_server != NULL) {
if (shutdown_callback == NULL) {
Local<FunctionTemplate>callback_tpl =

4
health_check/package.json
View File

@ -1,6 +1,6 @@
{
"name": "grpc-health-check",
"version": "1.3.0-dev",
"version": "1.4.0-dev",
"author": "Google Inc.",
"description": "Health check service for use with gRPC",
"repository": {
@ -15,7 +15,7 @@
}
],
"dependencies": {
"grpc": "^1.3.0-dev",
"grpc": "^1.4.0-dev",
"lodash": "^3.9.3",
"google-protobuf": "^3.0.0"
},

2
tools/package.json
View File

@ -1,6 +1,6 @@
{
"name": "grpc-tools",
"version": "1.3.0-dev",
"version": "1.4.0-dev",
"author": "Google Inc.",
"description": "Tools for developing with gRPC on Node.js",
"homepage": "http://www.grpc.io/",