Merge pull request #557 from WeiranFang/call-invocation-transformer

Call invocation transformer
grpc · Oct 19, 2018 · 8df65a9 · 8df65a9
2 parents 1e10613 + aea253c
commit 8df65a9
Show file tree

Hide file tree

Showing 2 changed files with 147 additions and 35 deletions.
diff --git a/.gitignore b/.gitignore
@@ -15,3 +15,8 @@ packages/grpc-native-core/src/node/
 
 .nyc_output/
 reports/
+
+package-lock.json
+
+# Test generated files
+coverage
diff --git a/packages/grpc-native-core/src/client.js b/packages/grpc-native-core/src/client.js
@@ -383,12 +383,16 @@ function Client(address, credentials, options) {
       .resolveInterceptorProviders(self.$interceptor_providers, method_definition)
       .concat(self.$interceptors);
   });
+
+  this.$callInvocationTransformer = options.callInvocationTransformer;
+
   let channelOverride = options.channelOverride;
   let channelFactoryOverride = options.channelFactoryOverride;
   // Exclude channel options which have already been consumed
   var channel_options = _.omit(options,
      ['interceptors', 'interceptor_providers',
-      'channelOverride', 'channelFactoryOverride']);
+      'channelOverride', 'channelFactoryOverride',
+      'callInvocationTransformer']);
   /* Private fields use $ as a prefix instead of _ because it is an invalid
    * prefix of a method name */
   if (channelOverride) {
@@ -480,22 +484,50 @@ Client.prototype.makeUnaryRequest = function(path, serialize, deserialize,
 
   metadata = metadata.clone();
 
+  var callProperties = {
+    argument: argument,
+    metadata: metadata,
+    call: new ClientUnaryCall(),
+    channel: this.$channel,
+    methodDefinition: method_definition,
+    callOptions: options,
+    callback: callback
+  };
+
+  // Transform call properties if specified.
+  if (this.$callInvocationTransformer) {
+    callProperties = this.$callInvocationTransformer(callProperties);
+  }
+
+  var callOptions = callProperties.callOptions;
+  var methodDefinition = callProperties.methodDefinition;
+
+  var interceptors = Client.prototype.resolveCallInterceptors.call(
+    this,
+    methodDefinition,
+    callOptions.interceptors,
+    callOptions.interceptor_providers
+  );
+
   var intercepting_call = client_interceptors.getInterceptingCall(
-    method_definition,
-    options,
-    Client.prototype.resolveCallInterceptors.call(this, method_definition, options.interceptors, options.interceptor_providers),
-    this.$channel,
-    callback
+    methodDefinition,
+    callOptions,
+    interceptors,
+    callProperties.channel,
+    callProperties.callback
   );
-  var emitter = new ClientUnaryCall(intercepting_call);
+
+  var emitter = callProperties.call;
+  emitter.call = intercepting_call;
+
   var last_listener = client_interceptors.getLastListener(
-    method_definition,
+    methodDefinition,
     emitter,
-    callback
+    callProperties.callback
   );
 
-  intercepting_call.start(metadata, last_listener);
-  intercepting_call.sendMessage(argument);
+  intercepting_call.start(callProperties.metadata, last_listener);
+  intercepting_call.sendMessage(callProperties.argument);
   intercepting_call.halfClose();
 
   return emitter;
@@ -555,21 +587,48 @@ Client.prototype.makeClientStreamRequest = function(path, serialize,
 
   metadata = metadata.clone();
 
+  var callProperties = {
+    metadata: metadata,
+    call: new ClientWritableStream(),
+    channel: this.$channel,
+    methodDefinition: method_definition,
+    callOptions: options,
+    callback: callback
+  };
+
+  // Transform call properties if specified.
+  if (this.$callInvocationTransformer) {
+    callProperties = this.$callInvocationTransformer(callProperties);
+  }
+
+  var callOptions = callProperties.callOptions;
+  var methodDefinition = callProperties.methodDefinition;
+
+  var interceptors = Client.prototype.resolveCallInterceptors.call(
+    this,
+    methodDefinition,
+    callOptions.interceptors,
+    callOptions.interceptor_providers
+  );
+
   var intercepting_call = client_interceptors.getInterceptingCall(
-    method_definition,
-    options,
-    Client.prototype.resolveCallInterceptors.call(this, method_definition, options.interceptors, options.interceptor_providers),
-    this.$channel,
-    callback
+    methodDefinition,
+    callOptions,
+    interceptors,
+    callProperties.channel,
+    callProperties.callback
   );
-  var emitter = new ClientWritableStream(intercepting_call);
+
+  var emitter = callProperties.call;
+  emitter.call = intercepting_call;
+
   var last_listener = client_interceptors.getLastListener(
-    method_definition,
+    methodDefinition,
     emitter,
-    callback
+    callProperties.callback
   );
 
-  intercepting_call.start(metadata, last_listener);
+  intercepting_call.start(callProperties.metadata, last_listener);
 
   return emitter;
 };
@@ -613,22 +672,46 @@ Client.prototype.makeServerStreamRequest = function(path, serialize,
 
   metadata = metadata.clone();
 
-  var emitter = new ClientReadableStream();
+  var callProperties = {
+    argument: argument,
+    metadata: metadata,
+    call: new ClientReadableStream(),
+    channel: this.$channel,
+    methodDefinition: method_definition,
+    callOptions: options,
+  };
+
+  // Transform call properties if specified.
+  if (this.$callInvocationTransformer) {
+    callProperties = this.$callInvocationTransformer(callProperties);
+  }
+
+  var callOptions = callProperties.callOptions;
+  var methodDefinition = callProperties.methodDefinition;
+
+  var interceptors = Client.prototype.resolveCallInterceptors.call(
+    this,
+    methodDefinition,
+    callOptions.interceptors,
+    callOptions.interceptor_providers
+  );
+
+  var emitter = callProperties.call;
   var intercepting_call = client_interceptors.getInterceptingCall(
-    method_definition,
-    options,
-    Client.prototype.resolveCallInterceptors.call(this, method_definition, options.interceptors, options.interceptor_providers),
-    this.$channel,
+    methodDefinition,
+    callOptions,
+    interceptors,
+    callProperties.channel,
     emitter
   );
   emitter.call = intercepting_call;
   var last_listener = client_interceptors.getLastListener(
-    method_definition,
+    methodDefinition,
     emitter
   );
 
-  intercepting_call.start(metadata, last_listener);
-  intercepting_call.sendMessage(argument);
+  intercepting_call.start(callProperties.metadata, last_listener);
+  intercepting_call.sendMessage(callProperties.argument);
   intercepting_call.halfClose();
 
   return emitter;
@@ -669,21 +752,45 @@ Client.prototype.makeBidiStreamRequest = function(path, serialize,
 
   metadata = metadata.clone();
 
-  var emitter = new ClientDuplexStream();
+  var callProperties = {
+    metadata: metadata,
+    call: new ClientDuplexStream(),
+    channel: this.$channel,
+    methodDefinition: method_definition,
+    callOptions: options,
+  };
+
+  // Transform call properties if specified.
+  if (this.$callInvocationTransformer) {
+    callProperties = this.$callInvocationTransformer(callProperties);
+  }
+
+  var callOptions = callProperties.callOptions;
+  var methodDefinition = callProperties.methodDefinition;
+
+  var interceptors = Client.prototype.resolveCallInterceptors.call(
+    this,
+    methodDefinition,
+    callOptions.interceptors,
+    callOptions.interceptor_providers
+  );
+
+
+  var emitter = callProperties.call;
   var intercepting_call = client_interceptors.getInterceptingCall(
-    method_definition,
-    options,
-    Client.prototype.resolveCallInterceptors.call(this, method_definition, options.interceptors, options.interceptor_providers),
-    this.$channel,
+    methodDefinition,
+    callOptions,
+    interceptors,
+    callProperties.channel,
     emitter
   );
   emitter.call = intercepting_call;
   var last_listener = client_interceptors.getLastListener(
-    method_definition,
+    methodDefinition,
     emitter
   );
 
-  intercepting_call.start(metadata, last_listener);
+  intercepting_call.start(callProperties.metadata, last_listener);
 
   return emitter;
 };