class GRPC::ActiveCall

The ActiveCall class provides simple methods for sending marshallable data to a call

Constants

InterceptableView

InterceptableView further limits access to an ActiveCall’s methods for use in interceptors on the client, exposing only the deadline

MultiReqView

MultiReqView limits access to an ActiveCall’s methods for use in server client_streamer handlers.

Operation

Operation limits access to an ActiveCall’s methods for use as a Operation on the client. TODO(apolcyn): expose peer getter

SingleReqView

SingleReqView limits access to an ActiveCall’s methods for use in server handlers that receive just one request.

Attributes

deadline[R]
metadata_sent[R]
metadata_to_send[R]
peer[R]
peer_cert[R]

Public Class Methods

client_invoke(call, metadata = {}) click to toggle source

client_invoke begins a client invocation.

Flow Control note: this blocks until flow control accepts that client request can go ahead.

deadline is the absolute deadline for the call.

Keyword Arguments ==

any keyword arguments are treated as metadata to be sent to the server if a keyword value is a list, multiple metadata for it’s key are sent

@param call [Call] a call on which to start and invocation @param metadata [Hash] the metadata

# File src/ruby/lib/grpc/generic/active_call.rb, line 63
def self.client_invoke(call, metadata = {})
  fail(TypeError, '!Core::Call') unless call.is_a? Core::Call
  call.run_batch(SEND_INITIAL_METADATA => metadata)
end
new(call, marshal, unmarshal, deadline, started: true, metadata_received: false, metadata_to_send: nil) click to toggle source

Creates an ActiveCall.

ActiveCall should only be created after a call is accepted. That means different things on a client and a server. On the client, the call is accepted after calling call.invoke. On the server, this is after call.accept.

initialize cannot determine if the call is accepted or not; so if a call that’s not accepted is used here, the error won’t be visible until the ActiveCall methods are called.

deadline is the absolute deadline for the call.

@param call [Call] the call used by the ActiveCall @param marshal [Function] f(obj)->string that marshal requests @param unmarshal [Function] f(string)->obj that unmarshals responses @param deadline [Fixnum] the deadline for the call to complete @param started [true|false] indicates that metadata was sent @param metadata_received [true|false] indicates if metadata has already

been received. Should always be true for server calls
# File src/ruby/lib/grpc/generic/active_call.rb, line 88
def initialize(call, marshal, unmarshal, deadline, started: true,
               metadata_received: false, metadata_to_send: nil)
  fail(TypeError, '!Core::Call') unless call.is_a? Core::Call
  @call = call
  @deadline = deadline
  @marshal = marshal
  @unmarshal = unmarshal
  @metadata_received = metadata_received
  @metadata_sent = started
  @op_notifier = nil

  fail(ArgumentError, 'Already sent md') if started && metadata_to_send
  @metadata_to_send = metadata_to_send || {} unless started
  @send_initial_md_mutex = Mutex.new

  @output_stream_done = false
  @input_stream_done = false
  @call_finished = false
  @call_finished_mu = Mutex.new

  @client_call_executed = false
  @client_call_executed_mu = Mutex.new

  # set the peer now so that the accessor can still function
  # after the server closes the call
  @peer = call.peer
end

Private Class Methods

view_class(*visible_methods) click to toggle source
# File src/ruby/lib/grpc/generic/active_call.rb, line 637
def self.view_class(*visible_methods)
  Class.new do
    extend ::Forwardable
    def_delegators :@wrapped, *visible_methods

    # @param wrapped [ActiveCall] the call whose methods are shielded
    def initialize(wrapped)
      @wrapped = wrapped
    end
  end
end

Public Instance Methods

attach_peer_cert(peer_cert) click to toggle source
# File src/ruby/lib/grpc/generic/active_call.rb, line 585
def attach_peer_cert(peer_cert)
  @peer_cert = peer_cert
end
attach_status_results_and_complete_call(recv_status_batch_result) click to toggle source
# File src/ruby/lib/grpc/generic/active_call.rb, line 181
def attach_status_results_and_complete_call(recv_status_batch_result)
  unless recv_status_batch_result.status.nil?
    @call.trailing_metadata = recv_status_batch_result.status.metadata
  end
  @call.status = recv_status_batch_result.status

  # The RECV_STATUS in run_batch always succeeds
  # Check the status for a bad status or failed run batch
  recv_status_batch_result.check_status
end
bidi_streamer(requests, metadata: {}, &blk) click to toggle source

bidi_streamer sends a stream of requests to the GRPC server, and yields a stream of responses.

This method takes an Enumerable of requests, and returns and enumerable of responses.

requests ==

requests provides an ‘iterable’ of Requests. I.e. it follows Ruby’s each enumeration protocol. In the simplest case, requests will be an array of marshallable objects; in typical case it will be an Enumerable that allows dynamic construction of the marshallable objects.

responses ==

This is an enumerator of responses. I.e, its next method blocks waiting for the next response. Also, if at any point the block needs to consume all the remaining responses, this can be done using each or collect. Calling each or collect should only be done if the_call#writes_done has been called, otherwise the block will loop forever.

@param requests [Object] an Enumerable of requests to send @param metadata [Hash] metadata to be sent to the server. If a value is a list, multiple metadata for its key are sent @return [Enumerator, nil] a response Enumerator

# File src/ruby/lib/grpc/generic/active_call.rb, line 500
def bidi_streamer(requests, metadata: {}, &blk)
  raise_error_if_already_executed
  # Metadata might have already been sent if this is an operation view
  begin
    send_initial_metadata(metadata)
  rescue GRPC::Core::CallError => e
    batch_result = @call.run_batch(RECV_STATUS_ON_CLIENT => nil)
    set_input_stream_done
    set_output_stream_done
    attach_status_results_and_complete_call(batch_result)
    raise e
  rescue => e
    set_input_stream_done
    set_output_stream_done
    raise e
  end

  bd = BidiCall.new(@call,
                    @marshal,
                    @unmarshal,
                    metadata_received: @metadata_received)

  bd.run_on_client(requests,
                   proc { set_input_stream_done },
                   proc { set_output_stream_done },
                   &blk)
end
cancelled?() click to toggle source

cancelled indicates if the call was cancelled

# File src/ruby/lib/grpc/generic/active_call.rb, line 134
def cancelled?
  !@call.status.nil? && @call.status.code == Core::StatusCodes::CANCELLED
end
client_streamer(requests, metadata: {}) click to toggle source

client_streamer sends a stream of requests to a GRPC server, and returns a single response.

requests provides an ‘iterable’ of Requests. I.e. it follows Ruby’s each enumeration protocol. In the simplest case, requests will be an array of marshallable objects; in typical case it will be an Enumerable that allows dynamic construction of the marshallable objects.

@param requests [Object] an Enumerable of requests to send @param metadata [Hash] metadata to be sent to the server. If a value is a list, multiple metadata for its key are sent @return [Object] the response received from the server

# File src/ruby/lib/grpc/generic/active_call.rb, line 399
def client_streamer(requests, metadata: {})
  raise_error_if_already_executed
  begin
    send_initial_metadata(metadata)
    requests.each { |r| @call.run_batch(SEND_MESSAGE => @marshal.call(r)) }
  rescue GRPC::Core::CallError => e
    receive_and_check_status # check for Cancelled
    raise e
  rescue => e
    set_input_stream_done
    raise e
  ensure
    set_output_stream_done
  end

  batch_result = @call.run_batch(
    SEND_CLOSE_FROM_CLIENT => nil,
    RECV_INITIAL_METADATA => nil,
    RECV_MESSAGE => nil,
    RECV_STATUS_ON_CLIENT => nil
  )

  set_input_stream_done

  @call.metadata = batch_result.metadata
  attach_status_results_and_complete_call(batch_result)
  get_message_from_batch_result(batch_result)
end
each_remote_read() { |resp| ... } click to toggle source

each_remote_read passes each response to the given block or returns an enumerator the responses if no block is given. Used to generate the request enumerable for server-side client-streaming RPC’s.

Enumerator ==

  • next blocks until the remote endpoint sends a READ or FINISHED

  • for each read, enumerator#next yields the response

  • on status

    * if it's is OK, enumerator#next raises StopException
    * if is not OK, enumerator#next raises RuntimeException

Block ==

  • if provided it is executed for each response

  • the call blocks until no more responses are provided

@return [Enumerator] if no block was given

# File src/ruby/lib/grpc/generic/active_call.rb, line 303
def each_remote_read
  return enum_for(:each_remote_read) unless block_given?
  begin
    loop do
      resp = remote_read
      break if resp.nil?  # the last response was received
      yield resp
    end
  ensure
    set_input_stream_done
  end
end
each_remote_read_then_finish() { |resp| ... } click to toggle source

each_remote_read_then_finish passes each response to the given block or returns an enumerator of the responses if no block is given.

It is like each_remote_read, but it blocks on finishing on detecting the final message.

Enumerator ==

  • next blocks until the remote endpoint sends a READ or FINISHED

  • for each read, enumerator#next yields the response

  • on status

    * if it's is OK, enumerator#next raises StopException
    * if is not OK, enumerator#next raises RuntimeException

Block ==

  • if provided it is executed for each response

  • the call blocks until no more responses are provided

@return [Enumerator] if no block was given

# File src/ruby/lib/grpc/generic/active_call.rb, line 336
def each_remote_read_then_finish
  return enum_for(:each_remote_read_then_finish) unless block_given?
  loop do
    resp = remote_read
    break if resp.nil?  # the last response was received
    yield resp
  end

  receive_and_check_status
ensure
  set_input_stream_done
end
get_message_from_batch_result(recv_message_batch_result) click to toggle source
# File src/ruby/lib/grpc/generic/active_call.rb, line 275
def get_message_from_batch_result(recv_message_batch_result)
  unless recv_message_batch_result.nil? ||
         recv_message_batch_result.message.nil?
    return @unmarshal.call(recv_message_batch_result.message)
  end
  GRPC.logger.debug('found nil; the final response has been sent')
  nil
end
interceptable() click to toggle source

Returns a restricted view of this ActiveCall for use in interceptors

@return [InterceptableView]

# File src/ruby/lib/grpc/generic/active_call.rb, line 162
def interceptable
  InterceptableView.new(self)
end
merge_metadata_to_send(new_metadata = {}) click to toggle source

Add to the metadata that will be sent from the server. Fails if metadata has already been sent. Unused by client calls.

# File src/ruby/lib/grpc/generic/active_call.rb, line 578
def merge_metadata_to_send(new_metadata = {})
  @send_initial_md_mutex.synchronize do
    fail('cant change metadata after already sent') if @metadata_sent
    @metadata_to_send.merge!(new_metadata)
  end
end
multi_req_view() click to toggle source

multi_req_view provides a restricted view of this ActiveCall for use in a server client-streaming handler.

# File src/ruby/lib/grpc/generic/active_call.rb, line 140
def multi_req_view
  MultiReqView.new(self)
end
op_is_done() click to toggle source

Signals that an operation is done. Only relevant on the client-side (this is a no-op on the server-side)

# File src/ruby/lib/grpc/generic/active_call.rb, line 570
def op_is_done
  return if @op_notifier.nil?
  @op_notifier.notify(self)
end
operation() click to toggle source

operation provides a restricted view of this ActiveCall for use as a Operation.

# File src/ruby/lib/grpc/generic/active_call.rb, line 152
def operation
  @op_notifier = Notifier.new
  Operation.new(self)
end
output_metadata() click to toggle source

output_metadata are provides access to hash that can be used to save metadata to be sent as trailer

# File src/ruby/lib/grpc/generic/active_call.rb, line 129
def output_metadata
  @output_metadata ||= {}
end
read_unary_request() click to toggle source

Intended for use on server-side calls when a single request from the client is expected (i.e., unary and server-streaming RPC types).

# File src/ruby/lib/grpc/generic/active_call.rb, line 229
def read_unary_request
  req = remote_read
  set_input_stream_done
  req
end
receive_and_check_status() click to toggle source
# File src/ruby/lib/grpc/generic/active_call.rb, line 166
def receive_and_check_status
  ops = { RECV_STATUS_ON_CLIENT => nil }
  ops[RECV_INITIAL_METADATA] = nil unless @metadata_received
  batch_result = @call.run_batch(ops)
  unless @metadata_received
    @call.metadata = batch_result.metadata
  end
  set_input_stream_done
  attach_status_results_and_complete_call(batch_result)
ensure
  # Ensure we don't attempt to request the initial metadata again
  # in case an exception occurs.
  @metadata_received = true
end
remote_read() click to toggle source

remote_read reads a response from the remote endpoint.

It blocks until the remote endpoint replies with a message or status. On receiving a message, it returns the response after unmarshalling it. On receiving a status, it returns nil if the status is OK, otherwise raising BadStatus

# File src/ruby/lib/grpc/generic/active_call.rb, line 258
def remote_read
  ops = { RECV_MESSAGE => nil }
  ops[RECV_INITIAL_METADATA] = nil unless @metadata_received
  batch_result = @call.run_batch(ops)
  unless @metadata_received
    @call.metadata = batch_result.metadata
  end
  get_message_from_batch_result(batch_result)
rescue GRPC::Core::CallError => e
  GRPC.logger.info("remote_read: #{e}")
  nil
ensure
  # Ensure we don't attempt to request the initial metadata again
  # in case an exception occurs.
  @metadata_received = true
end
remote_send(req, marshalled = false) click to toggle source

remote_send sends a request to the remote endpoint.

It blocks until the remote endpoint accepts the message.

@param req [Object, String] the object to send or it’s marshal form. @param marshalled [false, true] indicates if the object is already marshalled.

# File src/ruby/lib/grpc/generic/active_call.rb, line 199
def remote_send(req, marshalled = false)
  send_initial_metadata
  GRPC.logger.debug("sending #{req}, marshalled? #{marshalled}")
  payload = marshalled ? req : @marshal.call(req)
  @call.run_batch(SEND_MESSAGE => payload)
end
request_response(req, metadata: {}) click to toggle source

request_response sends a request to a GRPC server, and returns the response.

@param req [Object] the request sent to the server @param metadata [Hash] metadata to be sent to the server. If a value is a list, multiple metadata for its key are sent @return [Object] the response received from the server

# File src/ruby/lib/grpc/generic/active_call.rb, line 356
def request_response(req, metadata: {})
  raise_error_if_already_executed
  ops = {
    SEND_MESSAGE => @marshal.call(req),
    SEND_CLOSE_FROM_CLIENT => nil,
    RECV_INITIAL_METADATA => nil,
    RECV_MESSAGE => nil,
    RECV_STATUS_ON_CLIENT => nil
  }
  @send_initial_md_mutex.synchronize do
    # Metadata might have already been sent if this is an operation view
    unless @metadata_sent
      ops[SEND_INITIAL_METADATA] = @metadata_to_send.merge!(metadata)
    end
    @metadata_sent = true
  end

  begin
    batch_result = @call.run_batch(ops)
    # no need to check for cancellation after a CallError because this
    # batch contains a RECV_STATUS op
  ensure
    set_input_stream_done
    set_output_stream_done
  end

  @call.metadata = batch_result.metadata
  attach_status_results_and_complete_call(batch_result)
  get_message_from_batch_result(batch_result)
end
run_server_bidi(mth, interception_ctx) click to toggle source

run_server_bidi orchestrates a BiDi stream processing on a server.

N.B. gen_each_reply is a func(Enumerable<Requests>)

It takes an enumerable of requests as an arg, in case there is a relationship between the stream of requests and the stream of replies.

This does not mean that must necessarily be one. E.g, the replies produced by gen_each_reply could ignore the received_msgs

@param mth [Proc] generates the BiDi stream replies @param interception_ctx [InterceptionContext]

# File src/ruby/lib/grpc/generic/active_call.rb, line 541
def run_server_bidi(mth, interception_ctx)
  view = multi_req_view
  bidi_call = BidiCall.new(
    @call,
    @marshal,
    @unmarshal,
    metadata_received: @metadata_received,
    req_view: view
  )
  requests = bidi_call.read_next_loop(proc { set_input_stream_done }, false)
  interception_ctx.intercept!(
    :bidi_streamer,
    call: view,
    method: mth,
    requests: requests
  ) do
    bidi_call.run_on_server(mth, requests)
  end
end
send_initial_metadata(new_metadata = {}) click to toggle source

Sends the initial metadata that has yet to be sent. Does nothing if metadata has already been sent for this call.

# File src/ruby/lib/grpc/generic/active_call.rb, line 118
def send_initial_metadata(new_metadata = {})
  @send_initial_md_mutex.synchronize do
    return if @metadata_sent
    @metadata_to_send.merge!(new_metadata)
    ActiveCall.client_invoke(@call, @metadata_to_send)
    @metadata_sent = true
  end
end
send_status(code = OK, details = '', assert_finished = false, metadata: {}) click to toggle source

send_status sends a status to the remote endpoint.

@param code [int] the status code to send @param details [String] details @param assert_finished [true, false] when true(default), waits for FINISHED. @param metadata [Hash] metadata to send to the server. If a value is a list, mulitple metadata for its key are sent

# File src/ruby/lib/grpc/generic/active_call.rb, line 214
def send_status(code = OK, details = '', assert_finished = false,
                metadata: {})
  send_initial_metadata
  ops = {
    SEND_STATUS_FROM_SERVER => Struct::Status.new(code, details, metadata)
  }
  ops[RECV_CLOSE_ON_SERVER] = nil if assert_finished
  @call.run_batch(ops)
  set_output_stream_done

  nil
end
server_streamer(req, metadata: {}) { |r| ... } click to toggle source

server_streamer sends one request to the GRPC server, which yields a stream of responses.

responses provides an enumerator over the streamed responses, i.e. it follows Ruby’s each iteration protocol. The enumerator blocks while waiting for each response, stops when the server signals that no further responses will be supplied. If the implicit block is provided, it is executed with each response as the argument and no result is returned.

@param req [Object] the request sent to the server @param metadata [Hash] metadata to be sent to the server. If a value is a list, multiple metadata for its key are sent @return [Enumerator|nil] a response Enumerator

# File src/ruby/lib/grpc/generic/active_call.rb, line 442
def server_streamer(req, metadata: {})
  raise_error_if_already_executed
  ops = {
    SEND_MESSAGE => @marshal.call(req),
    SEND_CLOSE_FROM_CLIENT => nil
  }
  @send_initial_md_mutex.synchronize do
    # Metadata might have already been sent if this is an operation view
    unless @metadata_sent
      ops[SEND_INITIAL_METADATA] = @metadata_to_send.merge!(metadata)
    end
    @metadata_sent = true
  end

  begin
    @call.run_batch(ops)
  rescue GRPC::Core::CallError => e
    receive_and_check_status # checks for Cancelled
    raise e
  rescue => e
    set_input_stream_done
    raise e
  ensure
    set_output_stream_done
  end

  replies = enum_for(:each_remote_read_then_finish)
  return replies unless block_given?
  replies.each { |r| yield r }
end
server_unary_response(req, trailing_metadata: {}, code: Core::StatusCodes::OK, details: 'OK') click to toggle source
# File src/ruby/lib/grpc/generic/active_call.rb, line 235
def server_unary_response(req, trailing_metadata: {},
                          code: Core::StatusCodes::OK, details: 'OK')
  ops = {}
  ops[SEND_MESSAGE] = @marshal.call(req)
  ops[SEND_STATUS_FROM_SERVER] = Struct::Status.new(
    code, details, trailing_metadata)
  ops[RECV_CLOSE_ON_SERVER] = nil

  @send_initial_md_mutex.synchronize do
    ops[SEND_INITIAL_METADATA] = @metadata_to_send unless @metadata_sent
    @metadata_sent = true
  end

  @call.run_batch(ops)
  set_output_stream_done
end
single_req_view() click to toggle source

single_req_view provides a restricted view of this ActiveCall for use in a server request-response handler.

# File src/ruby/lib/grpc/generic/active_call.rb, line 146
def single_req_view
  SingleReqView.new(self)
end
wait() click to toggle source

Waits till an operation completes

# File src/ruby/lib/grpc/generic/active_call.rb, line 562
def wait
  return if @op_notifier.nil?
  GRPC.logger.debug("active_call.wait: on #{@op_notifier}")
  @op_notifier.wait
end

Private Instance Methods

maybe_finish_and_close_call_locked() click to toggle source
# File src/ruby/lib/grpc/generic/active_call.rb, line 611
def maybe_finish_and_close_call_locked
  return unless @output_stream_done && @input_stream_done
  return if @call_finished
  @call_finished = true
  op_is_done
  @call.close
end
raise_error_if_already_executed() click to toggle source
# File src/ruby/lib/grpc/generic/active_call.rb, line 628
def raise_error_if_already_executed
  @client_call_executed_mu.synchronize do
    if @client_call_executed
      fail GRPC::Core::CallError, 'attempting to re-run a call'
    end
    @client_call_executed = true
  end
end
set_input_stream_done() click to toggle source

To be called once the “input stream” has been completelly read through (i.e, done reading from client or received status) note this is idempotent

# File src/ruby/lib/grpc/generic/active_call.rb, line 594
def set_input_stream_done
  @call_finished_mu.synchronize do
    @input_stream_done = true
    maybe_finish_and_close_call_locked
  end
end
set_output_stream_done() click to toggle source

To be called once the “output stream” has been completelly sent through (i.e, done sending from client or sent status) note this is idempotent

# File src/ruby/lib/grpc/generic/active_call.rb, line 604
def set_output_stream_done
  @call_finished_mu.synchronize do
    @output_stream_done = true
    maybe_finish_and_close_call_locked
  end
end
start_call(metadata = {}) click to toggle source

Starts the call if not already started @param metadata [Hash] metadata to be sent to the server. If a value is a list, multiple metadata for its key are sent

# File src/ruby/lib/grpc/generic/active_call.rb, line 622
def start_call(metadata = {})
  # TODO(apolcyn): we should cancel and clean up the call in case this
  # send initial MD op fails.
  merge_metadata_to_send(metadata) && send_initial_metadata
end