From 0e9a919fb5a102d065b22dddde4db811862cec8e Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Mon, 25 May 2026 10:12:31 +0900 Subject: [PATCH] Support initial bidirectional stream messages. --- lib/async/grpc/client.rb | 9 ++++++--- lib/async/grpc/stub.rb | 7 ++++--- test/async/grpc/client.rb | 19 ++++++++++++++++++- 3 files changed, 28 insertions(+), 7 deletions(-) diff --git a/lib/async/grpc/client.rb b/lib/async/grpc/client.rb index db2ea50..2129772 100644 --- a/lib/async/grpc/client.rb +++ b/lib/async/grpc/client.rb @@ -113,11 +113,12 @@ def call(request) # @parameter metadata [Hash] Custom metadata headers # @parameter timeout [Numeric | Nil] Optional timeout in seconds # @parameter encoding [String | Nil] Optional compression encoding + # @parameter initial [Object | Array | Nil] Optional initial message(s) for bidirectional streaming # @yields {|input, output| ...} Block for streaming calls # @returns [Object | Protocol::GRPC::Body::ReadableBody] Response message or readable body for streaming # @raises [ArgumentError] If method is unknown or streaming type is invalid # @raises [Protocol::GRPC::Error] If the gRPC call fails - def invoke(service, method, request = nil, metadata: {}, timeout: nil, encoding: nil, &block) + def invoke(service, method, request = nil, metadata: {}, timeout: nil, encoding: nil, initial: nil, &block) rpc = service.class.lookup_rpc(method) raise ArgumentError, "Unknown method: #{method}" unless rpc @@ -141,7 +142,7 @@ def invoke(service, method, request = nil, metadata: {}, timeout: nil, encoding: when :client_streaming client_streaming_call(path, headers, request_class, response_class, encoding, &block) when :bidirectional - bidirectional_call(path, headers, request_class, response_class, encoding, &block) + bidirectional_call(path, headers, request_class, response_class, encoding, initial: initial, &block) else raise ArgumentError, "Unknown streaming type: #{streaming}" end @@ -273,14 +274,16 @@ def client_streaming_call(path, headers, request_class, response_class, encoding # @parameter request_class [Class] Request message class # @parameter response_class [Class] Response message class # @parameter encoding [String | Nil] Compression 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 # @raises [Protocol::GRPC::Error] If the gRPC call fails - def bidirectional_call(path, headers, request_class, response_class, encoding, &block) + def bidirectional_call(path, headers, request_class, response_class, encoding, initial: nil, &block) body = Protocol::GRPC::Body::WritableBody.new( encoding: encoding, message_class: request_class ) + Array(initial).each{|message| body.write(message)} http_request = Protocol::HTTP::Request["POST", path, headers, body] response = call(http_request) diff --git a/lib/async/grpc/stub.rb b/lib/async/grpc/stub.rb index 7d0dca8..5501f50 100644 --- a/lib/async/grpc/stub.rb +++ b/lib/async/grpc/stub.rb @@ -36,7 +36,7 @@ def initialize(client, interface) # Uses snake_case method names (Ruby convention). # @parameter method_name [Symbol] The method name to call (snake_case) # @parameter args [Array] Positional arguments (first is the request message) - # @parameter options [Hash] Keyword arguments (metadata, timeout, encoding) + # @parameter options [Hash] Keyword arguments (metadata, timeout, encoding, initial) # @yields {|input, output| ...} Block for streaming calls # @returns [Object | Protocol::GRPC::Body::ReadableBody] Response message or readable body # @raises [NoMethodError] If the method is not found @@ -47,13 +47,14 @@ def method_missing(method_name, *args, **options, &block) # Extract request from args (first positional argument): request = args.first - # Extract metadata, timeout, encoding from options: + # Extract metadata, timeout, encoding and initial messages from options: metadata = options.delete(:metadata) || {} timeout = options.delete(:timeout) encoding = options.delete(:encoding) + initial = options.delete(:initial) # Delegate to client.invoke with PascalCase method name (for interface lookup): - @client.invoke(@interface, interface_method_name, request, metadata: metadata, timeout: timeout, encoding: encoding, &block) + @client.invoke(@interface, interface_method_name, request, metadata: metadata, timeout: timeout, encoding: encoding, initial: initial, &block) else super end diff --git a/test/async/grpc/client.rb b/test/async/grpc/client.rb index 59b4375..d11e184 100644 --- a/test/async/grpc/client.rb +++ b/test/async/grpc/client.rb @@ -109,7 +109,24 @@ expect(received_messages[1].value).to be == "Echo: message2" expect(received_messages[2].value).to be == "Echo: message3" end - + it "can send initial bidirectional streaming messages before receiving a response" 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") + ] + received_messages = [] + stub.bidirectional_call(initial: initial_messages) do |output, input| + output.close_write + input.each do |response| + received_messages << response + end + end + expect(received_messages.length).to be == 2 + expect(received_messages[0].value).to be == "Echo: initial1" + expect(received_messages[1].value).to be == "Echo: initial2" + end it "handles metadata" do grpc_client = Async::GRPC::Client.new(client) stub = grpc_client.stub(Async::GRPC::Fixtures::TestInterface, service_name)