Support initial bidirectional stream messages.

Author: samuel-williams-shopify

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)

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

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)