diff --git a/lib/async/grpc/client.rb b/lib/async/grpc/client.rb index 0e90d8d..f7acc55 100644 --- a/lib/async/grpc/client.rb +++ b/lib/async/grpc/client.rb @@ -276,6 +276,7 @@ def client_streaming_call(path, headers, request_class, response_class, encoding # @parameter initial [Object | Array | Nil] Optional initial message(s) to write before waiting for the response # @yields {|input, output| ...} Block to handle bidirectional streaming # @returns [Protocol::GRPC::Body::ReadableBody] Readable body for streaming messages + # Without a block, `initial` is sent as the complete request stream and the caller owns the returned readable body. # @raises [Protocol::GRPC::Error] If the gRPC call fails def bidirectional_call(path, headers, request_class, response_class, encoding, initial: nil, &block) body = Protocol::GRPC::Body::WritableBody.new( @@ -283,6 +284,7 @@ def bidirectional_call(path, headers, request_class, response_class, encoding, i message_class: request_class ) Array(initial).each{|message| body.write(message)} + body.close_write unless block_given? http_request = Protocol::HTTP::Request["POST", path, headers, body] response = call(http_request) @@ -297,6 +299,8 @@ def bidirectional_call(path, headers, request_class, response_class, encoding, i ) unless block_given? + # The caller owns the readable body and its underlying response: + response = nil return readable_body end diff --git a/test/async/grpc/client.rb b/test/async/grpc/client.rb index d6aab5f..a7337fa 100644 --- a/test/async/grpc/client.rb +++ b/test/async/grpc/client.rb @@ -127,6 +127,28 @@ expect(received_messages[0].value).to be == "Echo: initial1" expect(received_messages[1].value).to be == "Echo: initial2" end + + it "can return a bidirectional response stream for initial messages" do + grpc_client = Async::GRPC::Client.new(client) + stub = grpc_client.stub(Async::GRPC::Fixtures::TestInterface, service_name) + initial_messages = [ + Protocol::GRPC::Fixtures::TestMessage.new(value: "initial1"), + Protocol::GRPC::Fixtures::TestMessage.new(value: "initial2") + ] + + input = stub.bidirectional_call(initial: initial_messages) + begin + received_messages = [] + while message = input.read + received_messages << message.value + end + ensure + input.close + end + + expect(received_messages).to be == ["Echo: initial1", "Echo: initial2"] + end + it "handles metadata" do grpc_client = Async::GRPC::Client.new(client) stub = grpc_client.stub(Async::GRPC::Fixtures::TestInterface, service_name) @@ -277,6 +299,82 @@ end describe Async::GRPC::Client do + let(:endpoint) {Async::HTTP::Endpoint.parse("http://localhost:0")} + + with ".open" do + it "returns a client without a block" do + client = subject.open(endpoint) + + expect(client).to be_a(subject) + expect(client.delegate).to be_a(Async::HTTP::Client) + ensure + client&.close + end + + it "parses string endpoints and closes the client after yielding" do + closed = false + connected_endpoint = nil + delegate = Object.new + delegate.define_singleton_method(:close){closed = true} + + client_class = Class.new(subject) + client_class.define_singleton_method(:connect) do |endpoint| + connected_endpoint = endpoint + delegate + end + + result = client_class.open("http://localhost:0") do |client| + expect(client.delegate).to be == delegate + :result + end + + expect(result).to be == :result + expect(connected_endpoint).to be_a(Async::HTTP::Endpoint) + expect(closed).to be == true + end + end + + with ".with" do + it "inherits the delegate and merges headers" do + headers = Protocol::HTTP::Headers.new + headers["authorization"] = "Bearer token" + parent = subject.new(Protocol::HTTP::Middleware::Okay, headers: headers) + + client = subject.with(parent, headers: {"request-id" => "1234"}) + + expect(client.delegate).to be == parent.delegate + expect(client.headers["authorization"]).to be == "Bearer token" + expect(client.headers["request-id"]).to be == ["1234"] + end + end + + with "string representations" do + it "describes the client and its headers" do + headers = Protocol::HTTP::Headers.new + client = subject.new(Protocol::HTTP::Middleware::Okay, headers: headers) + + expect(client.to_s).to be == "#" + expect(client.inspect).to be == "#" + end + end + + with "an invalid streaming mode" do + it "rejects the RPC" do + interface_class = Class.new(Protocol::GRPC::Interface) do + rpc :InvalidCall, + request_class: Protocol::GRPC::Fixtures::TestMessage, + response_class: Protocol::GRPC::Fixtures::TestMessage, + streaming: :invalid + end + client = subject.new(Protocol::HTTP::Middleware::Okay) + interface = interface_class.new("test.InvalidService") + + expect do + client.invoke(interface, :InvalidCall) + end.to raise_exception(ArgumentError, message: be == "Unknown streaming type: invalid") + end + end + include Sus::Fixtures::Async::HTTP::ServerContext with "http/1" do diff --git a/test/async/grpc/dispatcher.rb b/test/async/grpc/dispatcher.rb index 980f062..5647be8 100644 --- a/test/async/grpc/dispatcher.rb +++ b/test/async/grpc/dispatcher.rb @@ -95,6 +95,36 @@ expect(status).to be == Protocol::GRPC::Status::UNIMPLEMENTED end + it "returns UNIMPLEMENTED when the registered service name does not match" do + registered_name = "alias.Service" + dispatcher = subject.new(services: {registered_name => service}) + path = Protocol::GRPC::Route.build(registered_name, "UnaryCall") + request = Protocol::HTTP::Request.new("http", "localhost", "POST", path, nil, headers, request_body) + + response = dispatcher.call(request) + + expect(Protocol::GRPC::Metadata.extract_status(response.headers)).to be == Protocol::GRPC::Status::UNIMPLEMENTED + expect(Protocol::GRPC::Metadata.extract_message(response.headers)).to be == "Service name mismatch: expected test.Service, got alias.Service" + end + + it "returns UNIMPLEMENTED when the service handler is missing" do + interface_class = Class.new(Protocol::GRPC::Interface) do + rpc :MissingCall, + request_class: Protocol::GRPC::Fixtures::TestMessage, + response_class: Protocol::GRPC::Fixtures::TestMessage + end + service_name = "test.IncompleteService" + service = Async::GRPC::Service.new(interface_class, service_name) + dispatcher = subject.new(services: {service_name => service}) + path = Protocol::GRPC::Route.build(service_name, "MissingCall") + request = Protocol::HTTP::Request.new("http", "localhost", "POST", path, nil, headers, request_body) + + response = dispatcher.call(request) + + expect(Protocol::GRPC::Metadata.extract_status(response.headers)).to be == Protocol::GRPC::Status::UNIMPLEMENTED + expect(Protocol::GRPC::Metadata.extract_message(response.headers)).to be == "Handler method not implemented: missing_call" + end + it "passes non-gRPC requests to next middleware" do next_middleware = proc{Protocol::HTTP::Response[404, {}, ["Not Found"]]} dispatcher = subject.new(next_middleware, services: { service_name => service })