Contents

# 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

Version data entries

1 entries across 1 versions & 1 rubygems

Version	Path
grpc_kit-0.1.9	lib/grpc_kit/calls/client_bidi_streamer.rb