Skip to content

Commit b9ba5a4

Browse files
committed
async_hooks integration -- address comments
1 parent 617c167 commit b9ba5a4

File tree

3 files changed

+33
-19
lines changed

3 files changed

+33
-19
lines changed

packages/grpc-native-core/deps/grpc

Submodule grpc updated 996 files

packages/grpc-native-core/src/async-hooks-integration.js

+22-13
Original file line numberDiff line numberDiff line change
@@ -31,16 +31,23 @@ var useAsyncHooks = semver.satisfies(process.version, '>=8');
3131
*/
3232

3333
if (useAsyncHooks) {
34-
var asyncHooks = require('async_hooks');
35-
module.exports = function createAsyncResourceWrapper(name) {
36-
var resource = new asyncHooks.AsyncResource(name);
34+
const asyncHooks = require('async_hooks');
35+
class GrpcAsyncResource extends asyncHooks.AsyncResource {
36+
constructor(name, handle) {
37+
super(name);
38+
this.handle = handle;
39+
}
40+
}
41+
module.exports = function createAsyncResourceWrapper(name, handle) {
42+
let resource = new GrpcAsyncResource(name, handle);
3743
return {
38-
wrap: function(fn) {
44+
wrap: (fn) => {
3945
return function() {
4046
if (resource) {
4147
resource.emitBefore();
48+
let result;
4249
try {
43-
var result = fn.apply(this, arguments);
50+
result = fn.apply(this, arguments);
4451
} finally {
4552
resource.emitAfter();
4653
}
@@ -50,18 +57,20 @@ if (useAsyncHooks) {
5057
}
5158
}
5259
},
53-
destroy: function() {
54-
if (resource) {
55-
resource.emitDestroy();
56-
resource = null;
57-
}
60+
destroy: () => {
61+
setImmediate(() => {
62+
if (resource) {
63+
resource.emitDestroy();
64+
resource = null;
65+
}
66+
});
5867
}
5968
};
6069
}
6170
} else {
62-
var noImpl = {
63-
wrap: function(fn) { return fn; },
64-
destroy: function() {}
71+
const noImpl = {
72+
wrap: fn => fn,
73+
destroy: () => {}
6574
};
6675
module.exports = function createAsyncResourceWrapper() { return noImpl; }
6776
}

packages/grpc-native-core/src/server.js

+10-5
Original file line numberDiff line numberDiff line change
@@ -576,27 +576,31 @@ ServerDuplexStream.prototype.waitForCancel = waitForCancel;
576576
* @param {grpc.Metadata} metadata Metadata from the client
577577
*/
578578
function handleUnary(call, handler, metadata) {
579-
var asyncResource = createAsyncResourceWrapper('GrpcServerRequest');
580579
var emitter = new ServerUnaryCall(call, metadata);
580+
var asyncResource = createAsyncResourceWrapper('grpc.ServerRequest', emitter);
581581
emitter.on('error', function(error) {
582582
handleError(call, error);
583+
asyncResource.destroy();
583584
});
584585
emitter.waitForCancel();
585586
var batch = {};
586587
batch[grpc.opType.RECV_MESSAGE] = true;
587588
call.startBatch(batch, asyncResource.wrap(function(err, result) {
588589
if (err) {
589590
handleError(call, err);
591+
asyncResource.destroy();
590592
return;
591593
}
592594
try {
593595
emitter.request = handler.deserialize(result.read);
594596
} catch (e) {
595597
e.code = constants.status.INTERNAL;
596598
handleError(call, e);
599+
asyncResource.destroy();
597600
return;
598601
}
599602
if (emitter.cancelled) {
603+
asyncResource.destroy();
600604
return;
601605
}
602606
handler.func(emitter, function sendUnaryData(err, value, trailer, flags) {
@@ -633,8 +637,8 @@ function handleUnary(call, handler, metadata) {
633637
* @param {grpc.Metadata} metadata Metadata from the client
634638
*/
635639
function handleServerStreaming(call, handler, metadata) {
636-
var asyncResource = createAsyncResourceWrapper('GrpcServerRequest');
637640
var stream = new ServerWritableStream(call, metadata, handler.serialize);
641+
var asyncResource = createAsyncResourceWrapper('grpc.ServerRequest', stream);
638642
stream.on('error', asyncResource.destroy);
639643
stream.on('finish', asyncResource.destroy);
640644
stream.waitForCancel();
@@ -677,10 +681,11 @@ function handleServerStreaming(call, handler, metadata) {
677681
* @param {grpc.Metadata} metadata Metadata from the client
678682
*/
679683
function handleClientStreaming(call, handler, metadata) {
680-
var asyncResource = createAsyncResourceWrapper('GrpcServerRequest');
681684
var stream = new ServerReadableStream(call, metadata, handler.deserialize);
685+
var asyncResource = createAsyncResourceWrapper('grpc.ServerRequest', stream);
682686
stream.on('error', function(error) {
683687
handleError(call, error);
688+
asyncResource.destroy();
684689
});
685690
stream.waitForCancel();
686691
asyncResource.wrap(function() {
@@ -719,9 +724,9 @@ function handleClientStreaming(call, handler, metadata) {
719724
* @param {Metadata} metadata Metadata from the client
720725
*/
721726
function handleBidiStreaming(call, handler, metadata) {
722-
var asyncResource = createAsyncResourceWrapper('GrpcServerRequest');
723727
var stream = new ServerDuplexStream(call, metadata, handler.serialize,
724728
handler.deserialize);
729+
var asyncResource = createAsyncResourceWrapper('grpc.ServerRequest', stream);
725730
stream.on('error', asyncResource.destroy);
726731
stream.on('finish', asyncResource.destroy);
727732
stream.waitForCancel();
@@ -766,7 +771,7 @@ Server.prototype.start = function() {
766771
}
767772
var self = this;
768773
this.started = true;
769-
this.asyncResourceWrap = createAsyncResourceWrapper('GrpcServer');
774+
this.asyncResourceWrap = createAsyncResourceWrapper('grpc.Server', this);
770775
this._server.start();
771776
/**
772777
* Handles the SERVER_RPC_NEW event. If there is a handler associated with

0 commit comments

Comments
 (0)