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 serverclient_streamer
handlers.- Operation
Operation
limits access to an ActiveCall's methods for use as aOperation
on the client.- SingleReqView
SingleReqView
limits access to an ActiveCall's methods for use in server handlers that receive just one request.
Attributes
Private Class Methods
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
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
# File src/ruby/lib/grpc/generic/active_call.rb, line 629 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
Private Instance Methods
# File src/ruby/lib/grpc/generic/active_call.rb, line 579 def attach_peer_cert(peer_cert) @peer_cert = peer_cert end
# File src/ruby/lib/grpc/generic/active_call.rb, line 178 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
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 494 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 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
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 393 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
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 297 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
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 330 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
# File src/ruby/lib/grpc/generic/active_call.rb, line 269 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
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
# File src/ruby/lib/grpc/generic/active_call.rb, line 605 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
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 572 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
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
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 564 def op_is_done return if @op_notifier.nil? @op_notifier.notify(self) end
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
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
# File src/ruby/lib/grpc/generic/active_call.rb, line 620 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
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 226 def read_unary_request req = remote_read set_input_stream_done req end
# 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 @metadata_received = true end set_input_stream_done attach_status_results_and_complete_call(batch_result) end
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 255 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 @metadata_received = true end get_message_from_batch_result(batch_result) rescue GRPC::Core::CallError => e GRPC.logger.info("remote_read: #{e}") nil end
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 196 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
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 350 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
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 535 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
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
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 211 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
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 436 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
# File src/ruby/lib/grpc/generic/active_call.rb, line 232 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
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 588 def set_input_stream_done @call_finished_mu.synchronize do @input_stream_done = true maybe_finish_and_close_call_locked end end
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 598 def set_output_stream_done @call_finished_mu.synchronize do @output_stream_done = true maybe_finish_and_close_call_locked end end
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
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 616 def start_call(metadata = {}) merge_metadata_to_send(metadata) && send_initial_metadata end
Waits till an operation completes
# File src/ruby/lib/grpc/generic/active_call.rb, line 556 def wait return if @op_notifier.nil? GRPC.logger.debug("active_call.wait: on #{@op_notifier}") @op_notifier.wait end