Merge pull request #230 from tbetbetbe/grpc_ruby_remove_the_need_for_pick_unused_port

Grpc ruby remove the need for pick unused port
pull/233/head
Michael Lumish 10 years ago
commit d082b7fc53
  1. 2
      src/ruby/lib/grpc/generic/active_call.rb
  2. 26
      src/ruby/spec/call_spec.rb
  3. 21
      src/ruby/spec/channel_spec.rb
  4. 1
      src/ruby/spec/client_server_spec.rb
  5. 10
      src/ruby/spec/generic/active_call_spec.rb
  6. 144
      src/ruby/spec/generic/client_stub_spec.rb
  7. 7
      src/ruby/spec/generic/rpc_server_spec.rb
  8. 45
      src/ruby/spec/port_picker.rb
  9. 5
      src/ruby/spec/server_spec.rb

@ -67,7 +67,7 @@ module Google
fail(ArgumentError, 'not a CompletionQueue')
end
call.add_metadata(kw) if kw.length > 0
invoke_accepted, client_metadata_read = Object.new, Object.new
client_metadata_read = Object.new
finished_tag = Object.new
call.invoke(q, client_metadata_read, finished_tag)
[finished_tag, client_metadata_read]

@ -28,7 +28,6 @@
# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
require 'grpc'
require 'port_picker'
include GRPC::Core::StatusCodes
@ -71,16 +70,8 @@ describe GRPC::Core::Call do
before(:each) do
@tag = Object.new
@client_queue = GRPC::Core::CompletionQueue.new
@server_queue = GRPC::Core::CompletionQueue.new
port = find_unused_tcp_port
host = "localhost:#{port}"
@server = GRPC::Core::Server.new(@server_queue, nil)
@server.add_http2_port(host)
@ch = GRPC::Core::Channel.new(host, nil)
end
after(:each) do
@server.close
fake_host = 'localhost:10101'
@ch = GRPC::Core::Channel.new(fake_host, nil)
end
describe '#start_read' do
@ -122,19 +113,6 @@ describe GRPC::Core::Call do
end
end
describe '#start_write' do
it 'should cause the WRITE_ACCEPTED event' do
call = make_test_call
call.invoke(@client_queue, @tag, @tag)
expect(call.start_write(GRPC::Core::ByteBuffer.new('test_start_write'),
@tag)).to be_nil
ev = @client_queue.next(deadline)
expect(ev.call).to be_a(GRPC::Core::Call)
expect(ev.type).to be(GRPC::Core::CompletionType::WRITE_ACCEPTED)
expect(ev.tag).to be(@tag)
end
end
describe '#status' do
it 'can save the status and read it back' do
call = make_test_call

@ -28,7 +28,8 @@
# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
require 'grpc'
require 'port_picker'
FAKE_HOST='localhost:0'
def load_test_certs
test_root = File.join(File.dirname(__FILE__), 'testdata')
@ -114,8 +115,7 @@ describe GRPC::Core::Channel do
describe '#create_call' do
it 'creates a call OK' do
port = find_unused_tcp_port
host = "localhost:#{port}"
host = FAKE_HOST
ch = GRPC::Core::Channel.new(host, nil)
deadline = Time.now + 5
@ -127,8 +127,7 @@ describe GRPC::Core::Channel do
end
it 'raises an error if called on a closed channel' do
port = find_unused_tcp_port
host = "localhost:#{port}"
host = FAKE_HOST
ch = GRPC::Core::Channel.new(host, nil)
ch.close
@ -142,16 +141,14 @@ describe GRPC::Core::Channel do
describe '#destroy' do
it 'destroys a channel ok' do
port = find_unused_tcp_port
host = "localhost:#{port}"
host = FAKE_HOST
ch = GRPC::Core::Channel.new(host, nil)
blk = proc { ch.destroy }
expect(&blk).to_not raise_error
end
it 'can be called more than once without error' do
port = find_unused_tcp_port
host = "localhost:#{port}"
host = FAKE_HOST
ch = GRPC::Core::Channel.new(host, nil)
blk = proc { ch.destroy }
blk.call
@ -167,16 +164,14 @@ describe GRPC::Core::Channel do
describe '#close' do
it 'closes a channel ok' do
port = find_unused_tcp_port
host = "localhost:#{port}"
host = FAKE_HOST
ch = GRPC::Core::Channel.new(host, nil)
blk = proc { ch.close }
expect(&blk).to_not raise_error
end
it 'can be called more than once without error' do
port = find_unused_tcp_port
host = "localhost:#{port}"
host = FAKE_HOST
ch = GRPC::Core::Channel.new(host, nil)
blk = proc { ch.close }
blk.call

@ -28,7 +28,6 @@
# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
require 'grpc'
require 'port_picker'
require 'spec_helper'
include GRPC::Core::CompletionType

@ -28,7 +28,6 @@
# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
require 'grpc'
require_relative '../port_picker'
include GRPC::Core::StatusCodes
@ -45,12 +44,11 @@ describe GRPC::ActiveCall do
@client_queue = GRPC::Core::CompletionQueue.new
@server_queue = GRPC::Core::CompletionQueue.new
port = find_unused_tcp_port
host = "localhost:#{port}"
host = '0.0.0.0:0'
@server = GRPC::Core::Server.new(@server_queue, nil)
@server.add_http2_port(host)
server_port = @server.add_http2_port(host)
@server.start
@ch = GRPC::Core::Channel.new(host, nil)
@ch = GRPC::Core::Channel.new("localhost:#{server_port}", nil)
end
after(:each) do
@ -206,7 +204,7 @@ describe GRPC::ActiveCall do
it 'get a nil msg before a status when an OK status is sent' do
call = make_test_call
done_tag, meta_tag = ActiveCall.client_invoke(call, @client_queue,
deadline)
deadline)
client_call = ActiveCall.new(call, @client_queue, @pass_through,
@pass_through, deadline,
finished_tag: done_tag,

@ -29,9 +29,9 @@
require 'grpc'
require 'xray/thread_dump_signal_handler'
require_relative '../port_picker'
NOOP = proc { |x| x }
FAKE_HOST = 'localhost:0'
def wakey_thread(&blk)
awake_mutex, awake_cond = Mutex.new, ConditionVariable.new
@ -67,7 +67,7 @@ describe 'ClientStub' do
describe '#new' do
it 'can be created from a host and args' do
host = new_test_host
host = FAKE_HOST
opts = { a_channel_arg: 'an_arg' }
blk = proc do
GRPC::ClientStub.new(host, @cq, **opts)
@ -76,7 +76,7 @@ describe 'ClientStub' do
end
it 'can be created with a default deadline' do
host = new_test_host
host = FAKE_HOST
opts = { a_channel_arg: 'an_arg', deadline: 5 }
blk = proc do
GRPC::ClientStub.new(host, @cq, **opts)
@ -85,7 +85,7 @@ describe 'ClientStub' do
end
it 'can be created with an channel override' do
host = new_test_host
host = FAKE_HOST
opts = { a_channel_arg: 'an_arg', channel_override: @ch }
blk = proc do
GRPC::ClientStub.new(host, @cq, **opts)
@ -94,7 +94,7 @@ describe 'ClientStub' do
end
it 'cannot be created with a bad channel override' do
host = new_test_host
host = FAKE_HOST
blk = proc do
opts = { a_channel_arg: 'an_arg', channel_override: Object.new }
GRPC::ClientStub.new(host, @cq, **opts)
@ -103,7 +103,7 @@ describe 'ClientStub' do
end
it 'cannot be created with bad credentials' do
host = new_test_host
host = FAKE_HOST
blk = proc do
opts = { a_channel_arg: 'an_arg', creds: Object.new }
GRPC::ClientStub.new(host, @cq, **opts)
@ -113,7 +113,7 @@ describe 'ClientStub' do
it 'can be created with test test credentials' do
certs = load_test_certs
host = new_test_host
host = FAKE_HOST
blk = proc do
opts = {
GRPC::Core::Channel::SSL_TARGET => 'foo.test.google.com',
@ -133,16 +133,17 @@ describe 'ClientStub' do
shared_examples 'request response' do
it 'should send a request to/receive a reply from a server' do
host = new_test_host
th = run_request_response(host, @sent_msg, @resp, @pass)
stub = GRPC::ClientStub.new(host, @cq)
server_port = create_test_server
th = run_request_response(@sent_msg, @resp, @pass)
stub = GRPC::ClientStub.new("localhost:#{server_port}", @cq)
expect(get_response(stub)).to eq(@resp)
th.join
end
it 'should send metadata to the server ok' do
host = new_test_host
th = run_request_response(host, @sent_msg, @resp, @pass,
server_port = create_test_server
host = "localhost:#{server_port}"
th = run_request_response(@sent_msg, @resp, @pass,
k1: 'v1', k2: 'v2')
stub = GRPC::ClientStub.new(host, @cq)
expect(get_response(stub)).to eq(@resp)
@ -150,8 +151,9 @@ describe 'ClientStub' do
end
it 'should update the sent metadata with a provided metadata updater' do
host = new_test_host
th = run_request_response(host, @sent_msg, @resp, @pass,
server_port = create_test_server
host = "localhost:#{server_port}"
th = run_request_response(@sent_msg, @resp, @pass,
k1: 'updated-v1', k2: 'v2')
update_md = proc do |md|
md[:k1] = 'updated-v1'
@ -163,8 +165,9 @@ describe 'ClientStub' do
end
it 'should send a request when configured using an override channel' do
alt_host = new_test_host
th = run_request_response(alt_host, @sent_msg, @resp, @pass)
server_port = create_test_server
alt_host = "localhost:#{server_port}"
th = run_request_response(@sent_msg, @resp, @pass)
ch = GRPC::Core::Channel.new(alt_host, nil)
stub = GRPC::ClientStub.new('ignored-host', @cq, channel_override: ch)
expect(get_response(stub)).to eq(@resp)
@ -172,8 +175,9 @@ describe 'ClientStub' do
end
it 'should raise an error if the status is not OK' do
host = new_test_host
th = run_request_response(host, @sent_msg, @resp, @fail)
server_port = create_test_server
host = "localhost:#{server_port}"
th = run_request_response(@sent_msg, @resp, @fail)
stub = GRPC::ClientStub.new(host, @cq)
blk = proc { get_response(stub) }
expect(&blk).to raise_error(GRPC::BadStatus)
@ -210,16 +214,18 @@ describe 'ClientStub' do
end
it 'should send requests to/receive a reply from a server' do
host = new_test_host
th = run_client_streamer(host, @sent_msgs, @resp, @pass)
server_port = create_test_server
host = "localhost:#{server_port}"
th = run_client_streamer(@sent_msgs, @resp, @pass)
stub = GRPC::ClientStub.new(host, @cq)
expect(get_response(stub)).to eq(@resp)
th.join
end
it 'should send metadata to the server ok' do
host = new_test_host
th = run_client_streamer(host, @sent_msgs, @resp, @pass,
server_port = create_test_server
host = "localhost:#{server_port}"
th = run_client_streamer(@sent_msgs, @resp, @pass,
k1: 'v1', k2: 'v2')
stub = GRPC::ClientStub.new(host, @cq)
expect(get_response(stub)).to eq(@resp)
@ -227,8 +233,9 @@ describe 'ClientStub' do
end
it 'should update the sent metadata with a provided metadata updater' do
host = new_test_host
th = run_client_streamer(host, @sent_msgs, @resp, @pass,
server_port = create_test_server
host = "localhost:#{server_port}"
th = run_client_streamer(@sent_msgs, @resp, @pass,
k1: 'updated-v1', k2: 'v2')
update_md = proc do |md|
md[:k1] = 'updated-v1'
@ -240,8 +247,9 @@ describe 'ClientStub' do
end
it 'should raise an error if the status is not ok' do
host = new_test_host
th = run_client_streamer(host, @sent_msgs, @resp, @fail)
server_port = create_test_server
host = "localhost:#{server_port}"
th = run_client_streamer(@sent_msgs, @resp, @fail)
stub = GRPC::ClientStub.new(host, @cq)
blk = proc { get_response(stub) }
expect(&blk).to raise_error(GRPC::BadStatus)
@ -278,16 +286,18 @@ describe 'ClientStub' do
end
it 'should send a request to/receive replies from a server' do
host = new_test_host
th = run_server_streamer(host, @sent_msg, @replys, @pass)
server_port = create_test_server
host = "localhost:#{server_port}"
th = run_server_streamer(@sent_msg, @replys, @pass)
stub = GRPC::ClientStub.new(host, @cq)
expect(get_responses(stub).collect { |r| r }).to eq(@replys)
th.join
end
it 'should raise an error if the status is not ok' do
host = new_test_host
th = run_server_streamer(host, @sent_msg, @replys, @fail)
server_port = create_test_server
host = "localhost:#{server_port}"
th = run_server_streamer(@sent_msg, @replys, @fail)
stub = GRPC::ClientStub.new(host, @cq)
e = get_responses(stub)
expect { e.collect { |r| r } }.to raise_error(GRPC::BadStatus)
@ -295,8 +305,9 @@ describe 'ClientStub' do
end
it 'should send metadata to the server ok' do
host = new_test_host
th = run_server_streamer(host, @sent_msg, @replys, @fail,
server_port = create_test_server
host = "localhost:#{server_port}"
th = run_server_streamer(@sent_msg, @replys, @fail,
k1: 'v1', k2: 'v2')
stub = GRPC::ClientStub.new(host, @cq)
e = get_responses(stub)
@ -305,8 +316,9 @@ describe 'ClientStub' do
end
it 'should update the sent metadata with a provided metadata updater' do
host = new_test_host
th = run_server_streamer(host, @sent_msg, @replys, @pass,
server_port = create_test_server
host = "localhost:#{server_port}"
th = run_server_streamer(@sent_msg, @replys, @pass,
k1: 'updated-v1', k2: 'v2')
update_md = proc do |md|
md[:k1] = 'updated-v1'
@ -352,8 +364,9 @@ describe 'ClientStub' do
end
it 'supports sending all the requests first', bidi: true do
host = new_test_host
th = run_bidi_streamer_handle_inputs_first(host, @sent_msgs, @replys,
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)
e = get_responses(stub)
@ -362,8 +375,9 @@ describe 'ClientStub' do
end
it 'supports client-initiated ping pong', bidi: true do
host = new_test_host
th = run_bidi_streamer_echo_ping_pong(host, @sent_msgs, @pass, true)
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)
e = get_responses(stub)
expect(e.collect { |r| r }).to eq(@sent_msgs)
@ -377,8 +391,9 @@ describe 'ClientStub' do
# they receive a message from the client. Without receiving all the
# metadata, the server does not accept the call, so this test hangs.
xit 'supports a server-initiated ping pong', bidi: true do
host = new_test_host
th = run_bidi_streamer_echo_ping_pong(host, @sent_msgs, @pass, false)
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)
e = get_responses(stub)
expect(e.collect { |r| r }).to eq(@sent_msgs)
@ -410,10 +425,10 @@ describe 'ClientStub' do
end
end
def run_server_streamer(hostname, expected_input, replys, status, **kw)
def run_server_streamer(expected_input, replys, status, **kw)
wanted_metadata = kw.clone
wakey_thread do |mtx, cnd|
c = expect_server_to_be_invoked(hostname, mtx, cnd)
c = expect_server_to_be_invoked(mtx, cnd)
wanted_metadata.each do |k, v|
expect(c.metadata[k.to_s]).to eq(v)
end
@ -423,20 +438,19 @@ describe 'ClientStub' do
end
end
def run_bidi_streamer_handle_inputs_first(hostname, expected_inputs, replys,
def run_bidi_streamer_handle_inputs_first(expected_inputs, replys,
status)
wakey_thread do |mtx, cnd|
c = expect_server_to_be_invoked(hostname, mtx, cnd)
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', true)
end
end
def run_bidi_streamer_echo_ping_pong(hostname, expected_inputs, status,
client_starts)
def run_bidi_streamer_echo_ping_pong(expected_inputs, status, client_starts)
wakey_thread do |mtx, cnd|
c = expect_server_to_be_invoked(hostname, mtx, cnd)
c = expect_server_to_be_invoked(mtx, cnd)
expected_inputs.each do |i|
if client_starts
expect(c.remote_read).to eq(i)
@ -450,10 +464,10 @@ describe 'ClientStub' do
end
end
def run_client_streamer(hostname, expected_inputs, resp, status, **kw)
def run_client_streamer(expected_inputs, resp, status, **kw)
wanted_metadata = kw.clone
wakey_thread do |mtx, cnd|
c = expect_server_to_be_invoked(hostname, mtx, cnd)
c = expect_server_to_be_invoked(mtx, cnd)
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)
@ -463,10 +477,10 @@ describe 'ClientStub' do
end
end
def run_request_response(hostname, expected_input, resp, status, **kw)
def run_request_response(expected_input, resp, status, **kw)
wanted_metadata = kw.clone
wakey_thread do |mtx, cnd|
c = expect_server_to_be_invoked(hostname, mtx, cnd)
c = expect_server_to_be_invoked(mtx, cnd)
expect(c.remote_read).to eq(expected_input)
wanted_metadata.each do |k, v|
expect(c.metadata[k.to_s]).to eq(v)
@ -476,32 +490,30 @@ describe 'ClientStub' do
end
end
def start_test_server(hostname, awake_mutex, awake_cond)
server_queue = GRPC::Core::CompletionQueue.new
@server = GRPC::Core::Server.new(server_queue, nil)
@server.add_http2_port(hostname)
def create_test_server
@server_queue = GRPC::Core::CompletionQueue.new
@server = GRPC::Core::Server.new(@server_queue, nil)
@server.add_http2_port('0.0.0.0:0')
end
def start_test_server(awake_mutex, awake_cond)
@server.start
@server_tag = Object.new
@server.request_call(@server_tag)
awake_mutex.synchronize { awake_cond.signal }
server_queue
end
def expect_server_to_be_invoked(hostname, awake_mutex, awake_cond)
server_queue = start_test_server(hostname, awake_mutex, awake_cond)
ev = server_queue.pluck(@server_tag, INFINITE_FUTURE)
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_accept(@server_queue, finished_tag)
server_call.server_end_initial_metadata
GRPC::ActiveCall.new(server_call, server_queue, NOOP, NOOP, INFINITE_FUTURE,
GRPC::ActiveCall.new(server_call, @server_queue, NOOP, NOOP,
INFINITE_FUTURE,
finished_tag: finished_tag)
end
def new_test_host
port = find_unused_tcp_port
"localhost:#{port}"
end
end

@ -29,7 +29,6 @@
require 'grpc'
require 'xray/thread_dump_signal_handler'
require_relative '../port_picker'
def load_test_certs
test_root = File.join(File.dirname(File.dirname(__FILE__)), 'testdata')
@ -104,10 +103,10 @@ describe GRPC::RpcServer do
@noop = proc { |x| x }
@server_queue = GRPC::Core::CompletionQueue.new
port = find_unused_tcp_port
@host = "localhost:#{port}"
server_host = '0.0.0.0:0'
@server = GRPC::Core::Server.new(@server_queue, nil)
@server.add_http2_port(@host)
server_port = @server.add_http2_port(server_host)
@host = "localhost:#{server_port}"
@ch = GRPC::Core::Channel.new(@host, nil)
end

@ -1,45 +0,0 @@
# Copyright 2014, 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.
require 'socket'
# @param [Fixnum] the minimum port number to accept
# @param [Fixnum] the maximum port number to accept
# @return [Fixnum ]a free tcp port
def find_unused_tcp_port(min = 32_768, max = 60_000)
# Allow the system to assign a port, by specifying 0.
# Loop until a port is assigned in the required range
loop do
socket = Socket.new(:INET, :STREAM, 0)
socket.bind(Addrinfo.tcp('127.0.0.1', 0))
p = socket.local_address.ip_port
socket.close
return p if p > min && p < max
end
end

@ -28,7 +28,6 @@
# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
require 'grpc'
require 'port_picker'
def load_test_certs
test_root = File.join(File.dirname(__FILE__), 'testdata')
@ -205,10 +204,8 @@ describe Server do
end
def start_a_server
port = find_unused_tcp_port
host = "localhost:#{port}"
s = Server.new(@cq, nil)
s.add_http2_port(host)
s.add_http2_port('0.0.0.0:0')
s.start
s
end

Loading…
Cancel
Save