# frozen_string_literal: true require 'grpc_kit/calls' module GrpcKit module Calls::Client class BidiStreamer < GrpcKit::Calls::Call alias outgoing_metadata metadata def initialize(*) super @mutex = Mutex.new @send = false end # @param data [Object] request message # @param last [Boolean] # @return [void] def send_msg(data, last: false) raise 'No method error' if @restrict if @reason raise "Upstream returns an error status: #{@reason}" end @mutex.synchronize do @stream.send_msg(data, last: last, metadata: outgoing_metadata) end @send = true end # This method not is expected to be call in the main thread where #send_msg is called # # @param last [Boolean] # @return [Object] response object def recv(last: false) raise 'No method error' if @restrict sleep 0.1 until @send loop do msg = @mutex.synchronize do @stream.recv_msg(last: last, blocking: false) end unless msg == :wait_readable return msg end end rescue GrpcKit::Errors::BadStatus => e @reason = e raise e end def close_and_send @mutex.synchronize do @stream.close_and_send end end end end end