diff options
Diffstat (limited to 'src/ruby/spec/generic/client_stub_spec.rb')
| -rw-r--r-- | src/ruby/spec/generic/client_stub_spec.rb | 136 |
1 files changed, 74 insertions, 62 deletions
diff --git a/src/ruby/spec/generic/client_stub_spec.rb b/src/ruby/spec/generic/client_stub_spec.rb index 0c98fc40d9..193c5f2a03 100644 --- a/src/ruby/spec/generic/client_stub_spec.rb +++ b/src/ruby/spec/generic/client_stub_spec.rb @@ -30,15 +30,41 @@ require 'grpc' require 'xray/thread_dump_signal_handler' -NOOP = proc { |x| x } -FAKE_HOST = 'localhost:0' +# Notifier is useful high-level synchronization primitive. +class Notifier + attr_reader :payload, :notified + alias_method :notified?, :notified + + def initialize + @mutex = Mutex.new + @cvar = ConditionVariable.new + @notified = false + @payload = nil + end + + def wait + @mutex.synchronize do + @cvar.wait(@mutex) until notified? + end + end + + def notify(payload) + @mutex.synchronize do + return Error.new('already notified') if notified? + @payload = payload + @notified = true + @cvar.signal + return nil + end + end +end def wakey_thread(&blk) - awake_mutex, awake_cond = Mutex.new, ConditionVariable.new + n = Notifier.new t = Thread.new do - blk.call(awake_mutex, awake_cond) + blk.call(n) end - awake_mutex.synchronize { awake_cond.wait(awake_mutex) } + n.wait t end @@ -50,8 +76,11 @@ end include GRPC::Core::StatusCodes include GRPC::Core::TimeConsts +include GRPC::Core::CallOps describe 'ClientStub' do + let(:noop) { proc { |x| x } } + before(:each) do Thread.abort_on_exception = true @server = nil @@ -66,61 +95,56 @@ describe 'ClientStub' do end describe '#new' do + let(:fake_host) { 'localhost:0' } it 'can be created from a host and args' do - host = FAKE_HOST opts = { a_channel_arg: 'an_arg' } blk = proc do - GRPC::ClientStub.new(host, @cq, **opts) + GRPC::ClientStub.new(fake_host, @cq, **opts) end expect(&blk).not_to raise_error end it 'can be created with a default deadline' do - host = FAKE_HOST opts = { a_channel_arg: 'an_arg', deadline: 5 } blk = proc do - GRPC::ClientStub.new(host, @cq, **opts) + GRPC::ClientStub.new(fake_host, @cq, **opts) end expect(&blk).not_to raise_error end it 'can be created with an channel override' do - host = FAKE_HOST opts = { a_channel_arg: 'an_arg', channel_override: @ch } blk = proc do - GRPC::ClientStub.new(host, @cq, **opts) + GRPC::ClientStub.new(fake_host, @cq, **opts) end expect(&blk).not_to raise_error end it 'cannot be created with a bad channel override' do - host = FAKE_HOST blk = proc do opts = { a_channel_arg: 'an_arg', channel_override: Object.new } - GRPC::ClientStub.new(host, @cq, **opts) + GRPC::ClientStub.new(fake_host, @cq, **opts) end expect(&blk).to raise_error end it 'cannot be created with bad credentials' do - host = FAKE_HOST blk = proc do opts = { a_channel_arg: 'an_arg', creds: Object.new } - GRPC::ClientStub.new(host, @cq, **opts) + GRPC::ClientStub.new(fake_host, @cq, **opts) end expect(&blk).to raise_error end it 'can be created with test test credentials' do certs = load_test_certs - host = FAKE_HOST blk = proc do opts = { GRPC::Core::Channel::SSL_TARGET => 'foo.test.google.fr', a_channel_arg: 'an_arg', creds: GRPC::Core::Credentials.new(certs[0], nil, nil) } - GRPC::ClientStub.new(host, @cq, **opts) + GRPC::ClientStub.new(fake_host, @cq, **opts) end expect(&blk).to_not raise_error end @@ -187,7 +211,7 @@ describe 'ClientStub' do describe 'without a call operation' do def get_response(stub) - stub.request_response(@method, @sent_msg, NOOP, NOOP, + stub.request_response(@method, @sent_msg, noop, noop, k1: 'v1', k2: 'v2') end @@ -196,7 +220,7 @@ describe 'ClientStub' do describe 'via a call operation' do def get_response(stub) - op = stub.request_response(@method, @sent_msg, NOOP, NOOP, + op = stub.request_response(@method, @sent_msg, noop, noop, return_op: true, k1: 'v1', k2: 'v2') expect(op).to be_a(GRPC::ActiveCall::Operation) op.execute @@ -259,7 +283,7 @@ describe 'ClientStub' do describe 'without a call operation' do def get_response(stub) - stub.client_streamer(@method, @sent_msgs, NOOP, NOOP, + stub.client_streamer(@method, @sent_msgs, noop, noop, k1: 'v1', k2: 'v2') end @@ -268,7 +292,7 @@ describe 'ClientStub' do describe 'via a call operation' do def get_response(stub) - op = stub.client_streamer(@method, @sent_msgs, NOOP, NOOP, + op = stub.client_streamer(@method, @sent_msgs, noop, noop, return_op: true, k1: 'v1', k2: 'v2') expect(op).to be_a(GRPC::ActiveCall::Operation) op.execute @@ -333,7 +357,7 @@ describe 'ClientStub' do describe 'without a call operation' do def get_responses(stub) - e = stub.server_streamer(@method, @sent_msg, NOOP, NOOP, + e = stub.server_streamer(@method, @sent_msg, noop, noop, k1: 'v1', k2: 'v2') expect(e).to be_a(Enumerator) e @@ -344,7 +368,7 @@ describe 'ClientStub' do describe 'via a call operation' do def get_responses(stub) - op = stub.server_streamer(@method, @sent_msg, NOOP, NOOP, + op = stub.server_streamer(@method, @sent_msg, noop, noop, return_op: true, k1: 'v1', k2: 'v2') expect(op).to be_a(GRPC::ActiveCall::Operation) e = op.execute @@ -361,34 +385,30 @@ describe 'ClientStub' do before(:each) do @sent_msgs = Array.new(3) { |i| 'msg_' + (i + 1).to_s } @replys = Array.new(3) { |i| 'reply_' + (i + 1).to_s } + server_port = create_test_server + @host = "localhost:#{server_port}" end it 'supports sending all the requests first', bidi: true do - server_port = create_test_server - host = "localhost:#{server_port}" th = run_bidi_streamer_handle_inputs_first(@sent_msgs, @replys, @pass) - stub = GRPC::ClientStub.new(host, @cq) + stub = GRPC::ClientStub.new(@host, @cq) e = get_responses(stub) expect(e.collect { |r| r }).to eq(@replys) th.join end it 'supports client-initiated ping pong', bidi: true do - server_port = create_test_server - host = "localhost:#{server_port}" th = run_bidi_streamer_echo_ping_pong(@sent_msgs, @pass, true) - stub = GRPC::ClientStub.new(host, @cq) + stub = GRPC::ClientStub.new(@host, @cq) e = get_responses(stub) expect(e.collect { |r| r }).to eq(@sent_msgs) th.join end it 'supports a server-initiated ping pong', bidi: true do - server_port = create_test_server - host = "localhost:#{server_port}" th = run_bidi_streamer_echo_ping_pong(@sent_msgs, @pass, false) - stub = GRPC::ClientStub.new(host, @cq) + stub = GRPC::ClientStub.new(@host, @cq) e = get_responses(stub) expect(e.collect { |r| r }).to eq(@sent_msgs) th.join @@ -397,7 +417,7 @@ describe 'ClientStub' do describe 'without a call operation' do def get_responses(stub) - e = stub.bidi_streamer(@method, @sent_msgs, NOOP, NOOP) + e = stub.bidi_streamer(@method, @sent_msgs, noop, noop) expect(e).to be_a(Enumerator) e end @@ -407,7 +427,7 @@ describe 'ClientStub' do describe 'via a call operation' do def get_responses(stub) - op = stub.bidi_streamer(@method, @sent_msgs, NOOP, NOOP, + op = stub.bidi_streamer(@method, @sent_msgs, noop, noop, return_op: true) expect(op).to be_a(GRPC::ActiveCall::Operation) e = op.execute @@ -421,8 +441,8 @@ describe 'ClientStub' do def run_server_streamer(expected_input, replys, status, **kw) wanted_metadata = kw.clone - wakey_thread do |mtx, cnd| - c = expect_server_to_be_invoked(mtx, cnd) + wakey_thread do |notifier| + c = expect_server_to_be_invoked(notifier) wanted_metadata.each do |k, v| expect(c.metadata[k.to_s]).to eq(v) end @@ -434,8 +454,8 @@ describe 'ClientStub' do def run_bidi_streamer_handle_inputs_first(expected_inputs, replys, status) - wakey_thread do |mtx, cnd| - c = expect_server_to_be_invoked(mtx, cnd) + wakey_thread do |notifier| + c = expect_server_to_be_invoked(notifier) 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', true) @@ -443,8 +463,8 @@ describe 'ClientStub' do end def run_bidi_streamer_echo_ping_pong(expected_inputs, status, client_starts) - wakey_thread do |mtx, cnd| - c = expect_server_to_be_invoked(mtx, cnd) + wakey_thread do |notifier| + c = expect_server_to_be_invoked(notifier) expected_inputs.each do |i| if client_starts expect(c.remote_read).to eq(i) @@ -460,8 +480,8 @@ describe 'ClientStub' do def run_client_streamer(expected_inputs, resp, status, **kw) wanted_metadata = kw.clone - wakey_thread do |mtx, cnd| - c = expect_server_to_be_invoked(mtx, cnd) + wakey_thread do |notifier| + c = expect_server_to_be_invoked(notifier) expected_inputs.each { |i| expect(c.remote_read).to eq(i) } wanted_metadata.each do |k, v| expect(c.metadata[k.to_s]).to eq(v) @@ -473,8 +493,8 @@ describe 'ClientStub' do def run_request_response(expected_input, resp, status, **kw) wanted_metadata = kw.clone - wakey_thread do |mtx, cnd| - c = expect_server_to_be_invoked(mtx, cnd) + wakey_thread do |notifier| + c = expect_server_to_be_invoked(notifier) expect(c.remote_read).to eq(expected_input) wanted_metadata.each do |k, v| expect(c.metadata[k.to_s]).to eq(v) @@ -490,24 +510,16 @@ describe 'ClientStub' do @server.add_http2_port('0.0.0.0:0') end - def start_test_server(awake_mutex, awake_cond) + def expect_server_to_be_invoked(notifier) @server.start - @server_tag = Object.new - @server.request_call(@server_tag) - awake_mutex.synchronize { awake_cond.signal } - end - - def expect_server_to_be_invoked(awake_mutex, awake_cond) - start_test_server(awake_mutex, awake_cond) - ev = @server_queue.pluck(@server_tag, INFINITE_FUTURE) - fail OutOfTime if ev.nil? - server_call = ev.call - server_call.metadata = ev.result.metadata - finished_tag = Object.new - server_call.server_accept(@server_queue, finished_tag) - server_call.server_end_initial_metadata - GRPC::ActiveCall.new(server_call, @server_queue, NOOP, NOOP, - INFINITE_FUTURE, - finished_tag: finished_tag) + notifier.notify(nil) + server_tag = Object.new + recvd_rpc = @server.request_call(@server_queue, server_tag, + INFINITE_FUTURE) + recvd_call = recvd_rpc.call + recvd_call.metadata = recvd_rpc.metadata + recvd_call.run_batch(@server_queue, server_tag, Time.now + 2, + SEND_INITIAL_METADATA => nil) + GRPC::ActiveCall.new(recvd_call, @server_queue, noop, noop, INFINITE_FUTURE) end end |