Merge branch 'node_qps_test' into node_limit_thread_usage

include/grpc++/credentials.h

@@ -50,8 +50,8 @@ class Credentials {
 
  protected:
   friend std::unique_ptr<Credentials> CompositeCredentials(
-    const std::unique_ptr<Credentials>& creds1,
-    const std::unique_ptr<Credentials>& creds2);
+      const std::unique_ptr<Credentials>& creds1,
+      const std::unique_ptr<Credentials>& creds2);
 
   virtual SecureCredentials* AsSecureCredentials() = 0;
 
@@ -113,6 +113,12 @@ std::unique_ptr<Credentials> ServiceAccountCredentials(
 std::unique_ptr<Credentials> JWTCredentials(
     const grpc::string& json_key, std::chrono::seconds token_lifetime);
 
+// Builds refresh token credentials.
+// json_refresh_token is the JSON string containing the refresh token along
+// with a client_id and client_secret.
+std::unique_ptr<Credentials> RefreshTokenCredentials(
+    const grpc::string& json_refresh_token);
+
 // Builds IAM credentials.
 std::unique_ptr<Credentials> IAMCredentials(
     const grpc::string& authorization_token,

src/cpp/client/secure_credentials.cc

@@ -55,7 +55,8 @@ class SecureCredentials GRPC_FINAL : public Credentials {
     args.SetChannelArgs(&channel_args);
     return std::shared_ptr<ChannelInterface>(new Channel(
         args.GetSslTargetNameOverride().empty()
-            ? target : args.GetSslTargetNameOverride(),
+            ? target
+            : args.GetSslTargetNameOverride(),
         grpc_secure_channel_create(c_creds_, target.c_str(), &channel_args)));
   }
 
@@ -111,7 +112,7 @@ std::unique_ptr<Credentials> ServiceAccountCredentials(
 
 // Builds JWT credentials.
 std::unique_ptr<Credentials> JWTCredentials(
-    const grpc::string &json_key, std::chrono::seconds token_lifetime) {
+    const grpc::string& json_key, std::chrono::seconds token_lifetime) {
   if (token_lifetime.count() <= 0) {
     gpr_log(GPR_ERROR,
             "Trying to create JWTCredentials with non-positive lifetime");
@@ -122,6 +123,13 @@ std::unique_ptr<Credentials> JWTCredentials(
       grpc_jwt_credentials_create(json_key.c_str(), lifetime));
 }
 
+// Builds refresh token credentials.
+std::unique_ptr<Credentials> RefreshTokenCredentials(
+    const grpc::string& json_refresh_token) {
+  return WrapCredentials(
+      grpc_refresh_token_credentials_create(json_refresh_token.c_str()));
+}
+
 // Builds IAM credentials.
 std::unique_ptr<Credentials> IAMCredentials(
     const grpc::string& authorization_token,

src/node/examples/qps_test.js

@@ -0,0 +1,102 @@
+/*
+ *
+ * Copyright 2015, Google Inc.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ *     * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *     * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ *     * Neither the name of Google Inc. nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+
+'use strict';
+
+var async = require('async');
+var parseArgs = require('minimist');
+
+var grpc = require('..');
+var testProto = grpc.load(__dirname + '/../interop/test.proto').grpc.testing;
+var interop_server = require('../interop/interop_server.js');
+
+function runTest(concurrent, seconds, callback) {
+  var testServer = interop_server.getServer(0, false);
+  testServer.server.listen();
+  var client = new testProto.TestService('localhost:' + testServer.port);
+
+  var warmup_num = 100;
+
+  async.waterfall([
+    function warmUp(callback) {
+      var pending = warmup_num;
+      function startCall() {
+        client.emptyCall({}, function(err, resp) {
+          pending--;
+          if (pending === 0) {
+            callback(null);
+          }
+        });
+      }
+      for (var i = 0; i < warmup_num; i++) {
+        startCall();
+      }
+    }, function run(callback) {
+      var running = 0;
+      var count = 0;
+      var start = process.hrtime();
+      function responseCallback(err, resp) {
+        if (process.hrtime(start)[0] < seconds) {
+          count += 1;
+          client.emptyCall({}, responseCallback);
+        } else {
+          running -= 1;
+          if (running <= 0) {
+            callback(null, count);
+          }
+        }
+      }
+      for (var i = 0; i < concurrent; i++) {
+        running += 1;
+        client.emptyCall({}, responseCallback);
+      }
+    }], function(err, count) {
+      testServer.server.shutdown();
+      callback(err, count);
+    });
+}
+
+if (require.main === module) {
+  var argv = parseArgs(process.argv.slice(2), {
+    default: {'concurrent': 100,
+              'time': 10}
+  });
+  runTest(argv.concurrent, argv.time, function(err, count) {
+    if (err) {
+      throw err;
+    }
+    console.log('Concurrent calls:', argv.concurrent);
+    console.log('Time:', argv.time, 'seconds');
+    console.log('QPS:', (count/argv.time));
+  });
+}

src/ruby/lib/grpc/generic/active_call.rb

@@ -505,12 +505,12 @@ module GRPC
 
     # SingleReqView limits access to an ActiveCall's methods for use in server
     # handlers that receive just one request.
-    SingleReqView = view_class(:cancelled, :deadline)
+    SingleReqView = view_class(:cancelled, :deadline, :metadata)
 
     # MultiReqView limits access to an ActiveCall's methods for use in
     # server client_streamer handlers.
     MultiReqView = view_class(:cancelled, :deadline, :each_queued_msg,
-                              :each_remote_read)
+                              :each_remote_read, :metadata)
 
     # Operation limits access to an ActiveCall's methods for use as
     # a Operation on the client.

src/ruby/lib/grpc/generic/rpc_desc.rb

@@ -81,6 +81,7 @@ module GRPC
         active_call.run_server_bidi(mth)
       end
       send_status(active_call, OK, 'OK')
+      active_call.finished
     rescue BadStatus => e
       # this is raised by handlers that want GRPC to send an application
       # error code and detail message.

src/ruby/spec/generic/active_call_spec.rb

@@ -68,7 +68,7 @@ describe GRPC::ActiveCall do
 
     describe '#multi_req_view' do
       xit 'exposes a fixed subset of the ActiveCall methods' do
-        want = %w(cancelled, deadline, each_remote_read, shutdown)
+        want = %w(cancelled, deadline, each_remote_read, metadata, shutdown)
         v = @client_call.multi_req_view
         want.each do |w|
           expect(v.methods.include?(w))
@@ -78,7 +78,7 @@ describe GRPC::ActiveCall do
 
     describe '#single_req_view' do
       xit 'exposes a fixed subset of the ActiveCall methods' do
-        want = %w(cancelled, deadline, shutdown)
+        want = %w(cancelled, deadline, metadata, shutdown)
         v = @client_call.single_req_view
         want.each do |w|
           expect(v.methods.include?(w))

src/ruby/spec/generic/client_stub_spec.rb

@@ -434,7 +434,7 @@ describe 'ClientStub' do
       end
       expect(c.remote_read).to eq(expected_input)
       replys.each { |r| c.remote_send(r) }
-      c.send_status(status, status == @pass ? 'OK' : 'NOK')
+      c.send_status(status, status == @pass ? 'OK' : 'NOK', true)
     end
   end
 
@@ -444,7 +444,7 @@ describe 'ClientStub' do
       c = expect_server_to_be_invoked(mtx, cnd)
       expected_inputs.each { |i| expect(c.remote_read).to eq(i) }
       replys.each { |r| c.remote_send(r) }
-      c.send_status(status, status == @pass ? 'OK' : 'NOK')
+      c.send_status(status, status == @pass ? 'OK' : 'NOK', true)
     end
   end
 
@@ -460,7 +460,7 @@ describe 'ClientStub' do
           expect(c.remote_read).to eq(i)
         end
       end
-      c.send_status(status, status == @pass ? 'OK' : 'NOK')
+      c.send_status(status, status == @pass ? 'OK' : 'NOK', true)
     end
   end
 
@@ -473,7 +473,7 @@ describe 'ClientStub' do
         expect(c.metadata[k.to_s]).to eq(v)
       end
       c.remote_send(resp)
-      c.send_status(status, status == @pass ? 'OK' : 'NOK')
+      c.send_status(status, status == @pass ? 'OK' : 'NOK', true)
     end
   end
 
@@ -486,7 +486,7 @@ describe 'ClientStub' do
         expect(c.metadata[k.to_s]).to eq(v)
       end
       c.remote_send(resp)
-      c.send_status(status, status == @pass ? 'OK' : 'NOK')
+      c.send_status(status, status == @pass ? 'OK' : 'NOK', true)
     end
   end
 

src/ruby/spec/generic/rpc_desc_spec.rb

@@ -94,6 +94,7 @@ describe GRPC::RpcDesc do
         expect(@call).to receive(:remote_read).once.and_return(req)
         expect(@call).to receive(:remote_send).once.with(@ok_response)
         expect(@call).to receive(:send_status).once.with(OK, 'OK')
+        expect(@call).to receive(:finished).once
         @request_response.run_server_method(@call, method(:fake_reqresp))
       end
     end
@@ -134,6 +135,7 @@ describe GRPC::RpcDesc do
       it 'sends a response and closes the stream if there no errors' do
         expect(@call).to receive(:remote_send).once.with(@ok_response)
         expect(@call).to receive(:send_status).once.with(OK, 'OK')
+        expect(@call).to receive(:finished).once
         @client_streamer.run_server_method(@call, method(:fake_clstream))
       end
     end
@@ -178,6 +180,7 @@ describe GRPC::RpcDesc do
         expect(@call).to receive(:remote_read).once.and_return(req)
         expect(@call).to receive(:remote_send).twice.with(@ok_response)
         expect(@call).to receive(:send_status).once.with(OK, 'OK')
+        expect(@call).to receive(:finished).once
         @server_streamer.run_server_method(@call, method(:fake_svstream))
       end
     end
@@ -207,6 +210,7 @@ describe GRPC::RpcDesc do
       it 'closes the stream if there no errors' do
         expect(@call).to receive(:run_server_bidi)
         expect(@call).to receive(:send_status).once.with(OK, 'OK')
+        expect(@call).to receive(:finished).once
         @bidi_streamer.run_server_method(@call, method(:fake_bidistream))
       end
     end

src/ruby/spec/generic/rpc_server_spec.rb

@@ -62,12 +62,15 @@ end
 class EchoService
   include GRPC::GenericService
   rpc :an_rpc, EchoMsg, EchoMsg
+  attr_reader :received_md
 
   def initialize(_default_var = 'ignored')
+    @received_md = []
   end
 
-  def an_rpc(req, _call)
+  def an_rpc(req, call)
     logger.info('echo service received a request')
+    @received_md << call.metadata unless call.metadata.nil?
     req
   end
 end
@@ -337,6 +340,38 @@ describe GRPC::RpcServer do
         t.join
       end
 
+      it 'should receive metadata sent as rpc keyword args', server: true do
+        service = EchoService.new
+        @srv.handle(service)
+        t = Thread.new { @srv.run }
+        @srv.wait_till_running
+        req = EchoMsg.new
+        stub = EchoStub.new(@host, **@client_opts)
+        expect(stub.an_rpc(req, k1: 'v1', k2: 'v2')).to be_a(EchoMsg)
+        wanted_md = [{ 'k1' => 'v1', 'k2' => 'v2' }]
+        expect(service.received_md).to eq(wanted_md)
+        @srv.stop
+        t.join
+      end
+
+      it 'should receive updated metadata', server: true do
+        service = EchoService.new
+        @srv.handle(service)
+        t = Thread.new { @srv.run }
+        @srv.wait_till_running
+        req = EchoMsg.new
+        @client_opts[:update_metadata] = proc do |md|
+          md[:k1] = 'updated-v1'
+          md
+        end
+        stub = EchoStub.new(@host, **@client_opts)
+        expect(stub.an_rpc(req, k1: 'v1', k2: 'v2')).to be_a(EchoMsg)
+        wanted_md = [{ 'k1' => 'updated-v1', 'k2' => 'v2' }]
+        expect(service.received_md).to eq(wanted_md)
+        @srv.stop
+        t.join
+      end
+
       it 'should handle multiple parallel requests', server: true do
         @srv.handle(EchoService)
         Thread.new { @srv.run }

tools/gce_setup/interop_test_runner.sh

@@ -37,7 +37,7 @@ main() {
   source grpc_docker.sh
   test_cases=(large_unary empty_unary ping_pong client_streaming server_streaming cancel_after_begin cancel_after_first_response)
   clients=(cxx java go ruby node python csharp_mono)
-  servers=(cxx java go ruby node python)
+  servers=(cxx java go ruby node python csharp_mono)
   for test_case in "${test_cases[@]}"
   do
     for client in "${clients[@]}"