Class: GRPC::ActiveCall
- Inherits:
-
Object
- Object
- GRPC::ActiveCall
- Extended by:
- Forwardable
- Includes:
- Core::CallOps, Core::TimeConsts
- Defined in:
- src/ruby/lib/grpc/generic/active_call.rb
Overview
The ActiveCall class provides simple methods for sending marshallable data to a call
Instance Attribute Summary collapse
-
#deadline ⇒ Object
readonly
Returns the value of attribute deadline.
-
#metadata_sent ⇒ Object
readonly
Returns the value of attribute metadata_sent.
-
#metadata_to_send ⇒ Object
readonly
Returns the value of attribute metadata_to_send.
Class Method Summary collapse
-
.client_invoke(call, metadata = {}) ⇒ Object
client_invoke begins a client invocation.
Instance Method Summary collapse
- #attach_status_results_and_complete_call(recv_status_batch_result) ⇒ Object
-
#bidi_streamer(requests, metadata: {}, &blk) ⇒ Enumerator?
bidi_streamer sends a stream of requests to the GRPC server, and yields a stream of responses.
-
#cancelled? ⇒ Boolean
cancelled indicates if the call was cancelled.
-
#client_streamer(requests, metadata: {}) ⇒ Object
client_streamer sends a stream of requests to a GRPC server, and returns a single response.
-
#each_remote_read ⇒ Enumerator
each_remote_read passes each response to the given block or returns an enumerator the responses if no block is given.
-
#each_remote_read_then_finish ⇒ Enumerator
each_remote_read_then_finish passes each response to the given block or returns an enumerator of the responses if no block is given.
-
#finished ⇒ Object
finished waits until a client call is completed.
- #get_message_from_batch_result(recv_message_batch_result) ⇒ Object
-
#initialize(call, marshal, unmarshal, deadline, started: true, metadata_received: false, metadata_to_send: nil) ⇒ ActiveCall
constructor
Creates an ActiveCall.
- #merge_metadata_and_send_if_not_already_sent(new_metadata = {}) ⇒ Object
-
#merge_metadata_to_send(new_metadata = {}) ⇒ Object
Add to the metadata that will be sent from the server.
-
#multi_req_view ⇒ Object
multi_req_view provides a restricted view of this ActiveCall for use in a server client-streaming handler.
-
#op_is_done ⇒ Object
Signals that an operation is done.
-
#operation ⇒ Object
operation provides a restricted view of this ActiveCall for use as a Operation.
-
#output_metadata ⇒ Object
output_metadata are provides access to hash that can be used to save metadata to be sent as trailer.
-
#remote_read ⇒ Object
remote_read reads a response from the remote endpoint.
-
#remote_send(req, marshalled = false) ⇒ Object
remote_send sends a request to the remote endpoint.
-
#request_response(req, metadata: {}) ⇒ Object
request_response sends a request to a GRPC server, and returns the response.
-
#run_server_bidi(gen_each_reply) ⇒ Object
run_server_bidi orchestrates a BiDi stream processing on a server.
-
#send_initial_metadata ⇒ Object
Sends the initial metadata that has yet to be sent.
-
#send_status(code = OK, details = '', assert_finished = false, metadata: {}) ⇒ Object
send_status sends a status to the remote endpoint.
-
#server_streamer(req, metadata: {}) ⇒ Enumerator|nil
server_streamer sends one request to the GRPC server, which yields a stream of responses.
- #server_unary_response(req, trailing_metadata: {}, code: Core::StatusCodes::OK, details: 'OK') ⇒ Object
-
#single_req_view ⇒ Object
single_req_view provides a restricted view of this ActiveCall for use in a server request-response handler.
-
#wait ⇒ Object
Waits till an operation completes.
Methods included from Core::TimeConsts
Constructor Details
#initialize(call, marshal, unmarshal, deadline, started: true, metadata_received: false, metadata_to_send: nil) ⇒ ActiveCall
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.
104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 |
# File 'src/ruby/lib/grpc/generic/active_call.rb', line 104 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_sent = started @op_notifier = nil fail(ArgumentError, 'Already sent md') if started && @metadata_to_send = || {} unless started @send_initial_md_mutex = Mutex.new end |
Instance Attribute Details
#deadline ⇒ Object (readonly)
Returns the value of attribute deadline.
62 63 64 |
# File 'src/ruby/lib/grpc/generic/active_call.rb', line 62 def deadline @deadline end |
#metadata_sent ⇒ Object (readonly)
Returns the value of attribute metadata_sent.
62 63 64 |
# File 'src/ruby/lib/grpc/generic/active_call.rb', line 62 def @metadata_sent end |
#metadata_to_send ⇒ Object (readonly)
Returns the value of attribute metadata_to_send.
62 63 64 |
# File 'src/ruby/lib/grpc/generic/active_call.rb', line 62 def @metadata_to_send end |
Class Method Details
.client_invoke(call, metadata = {}) ⇒ Object
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
79 80 81 82 |
# File 'src/ruby/lib/grpc/generic/active_call.rb', line 79 def self.client_invoke(call, = {}) fail(TypeError, '!Core::Call') unless call.is_a? Core::Call call.run_batch(SEND_INITIAL_METADATA => ) end |
Instance Method Details
#attach_status_results_and_complete_call(recv_status_batch_result) ⇒ Object
168 169 170 171 172 173 174 175 176 177 178 179 |
# File 'src/ruby/lib/grpc/generic/active_call.rb', line 168 def attach_status_results_and_complete_call(recv_status_batch_result) unless recv_status_batch_result.status.nil? @call. = recv_status_batch_result.status. end @call.status = recv_status_batch_result.status @call.close op_is_done # 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) ⇒ Enumerator?
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.
a list, multiple metadata for its key are sent
438 439 440 441 442 443 444 445 446 447 |
# File 'src/ruby/lib/grpc/generic/active_call.rb', line 438 def bidi_streamer(requests, metadata: {}, &blk) # Metadata might have already been sent if this is an operation view () bd = BidiCall.new(@call, @marshal, @unmarshal, metadata_received: @metadata_received) bd.run_on_client(requests, @op_notifier, &blk) end |
#cancelled? ⇒ Boolean
cancelled indicates if the call was cancelled
137 138 139 |
# File 'src/ruby/lib/grpc/generic/active_call.rb', line 137 def cancelled? !@call.status.nil? && @call.status.code == Core::StatusCodes::CANCELLED end |
#client_streamer(requests, metadata: {}) ⇒ Object
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.
a list, multiple metadata for its key are sent
356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 |
# File 'src/ruby/lib/grpc/generic/active_call.rb', line 356 def client_streamer(requests, metadata: {}) # Metadata might have already been sent if this is an operation view () requests.each { |r| @call.run_batch(SEND_MESSAGE => @marshal.call(r)) } batch_result = @call.run_batch( SEND_CLOSE_FROM_CLIENT => nil, RECV_INITIAL_METADATA => nil, RECV_MESSAGE => nil, RECV_STATUS_ON_CLIENT => nil ) @call. = batch_result. attach_status_results_and_complete_call(batch_result) (batch_result) rescue GRPC::Core::CallError => e finished # checks for Cancelled raise e end |
#each_remote_read ⇒ Enumerator
each_remote_read passes each response to the given block or returns an enumerator the responses if no block is given.
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
274 275 276 277 278 279 280 281 |
# File 'src/ruby/lib/grpc/generic/active_call.rb', line 274 def each_remote_read return enum_for(:each_remote_read) unless block_given? loop do resp = remote_read break if resp.nil? # the last response was received yield resp end end |
#each_remote_read_then_finish ⇒ Enumerator
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
303 304 305 306 307 308 309 310 311 312 313 |
# File 'src/ruby/lib/grpc/generic/active_call.rb', line 303 def each_remote_read_then_finish return enum_for(:each_remote_read_then_finish) unless block_given? loop do resp = remote_read if resp.nil? # the last response was received, but not finished yet finished break end yield resp end end |
#finished ⇒ Object
finished waits until a client call is completed.
It blocks until the remote endpoint acknowledges by sending a status.
163 164 165 166 |
# File 'src/ruby/lib/grpc/generic/active_call.rb', line 163 def finished batch_result = @call.run_batch(RECV_STATUS_ON_CLIENT => nil) attach_status_results_and_complete_call(batch_result) end |
#get_message_from_batch_result(recv_message_batch_result) ⇒ Object
248 249 250 251 252 253 254 255 |
# File 'src/ruby/lib/grpc/generic/active_call.rb', line 248 def () unless .nil? || ..nil? return @unmarshal.call(.) end GRPC.logger.debug('found nil; the final response has been sent') nil end |
#merge_metadata_and_send_if_not_already_sent(new_metadata = {}) ⇒ Object
493 494 495 496 497 498 499 500 |
# File 'src/ruby/lib/grpc/generic/active_call.rb', line 493 def ( = {}) @send_initial_md_mutex.synchronize do return if @metadata_sent @metadata_to_send.merge!() @call.run_batch(SEND_INITIAL_METADATA => @metadata_to_send) @metadata_sent = true end end |
#merge_metadata_to_send(new_metadata = {}) ⇒ Object
Add to the metadata that will be sent from the server. Fails if metadata has already been sent. Unused by client calls.
486 487 488 489 490 491 |
# File 'src/ruby/lib/grpc/generic/active_call.rb', line 486 def ( = {}) @send_initial_md_mutex.synchronize do fail('cant change metadata after already sent') if @metadata_sent @metadata_to_send.merge!() end end |
#multi_req_view ⇒ Object
multi_req_view provides a restricted view of this ActiveCall for use in a server client-streaming handler.
143 144 145 |
# File 'src/ruby/lib/grpc/generic/active_call.rb', line 143 def multi_req_view MultiReqView.new(self) end |
#op_is_done ⇒ Object
Signals that an operation is done
478 479 480 481 |
# File 'src/ruby/lib/grpc/generic/active_call.rb', line 478 def op_is_done return if @op_notifier.nil? @op_notifier.notify(self) end |
#operation ⇒ Object
operation provides a restricted view of this ActiveCall for use as a Operation.
155 156 157 158 |
# File 'src/ruby/lib/grpc/generic/active_call.rb', line 155 def operation @op_notifier = Notifier.new Operation.new(self) end |
#output_metadata ⇒ Object
output_metadata are provides access to hash that can be used to save metadata to be sent as trailer
132 133 134 |
# File 'src/ruby/lib/grpc/generic/active_call.rb', line 132 def @output_metadata ||= {} end |
#remote_read ⇒ Object
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
237 238 239 240 241 242 243 244 245 246 |
# File 'src/ruby/lib/grpc/generic/active_call.rb', line 237 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. = batch_result. @metadata_received = true end (batch_result) end |
#remote_send(req, marshalled = false) ⇒ Object
remote_send sends a request to the remote endpoint.
It blocks until the remote endpoint accepts the message.
marshalled.
188 189 190 191 192 193 |
# File 'src/ruby/lib/grpc/generic/active_call.rb', line 188 def remote_send(req, marshalled = false) GRPC.logger.debug("sending #{req}, marshalled? #{marshalled}") payload = marshalled ? req : @marshal.call(req) @call.run_batch(SEND_MESSAGE => payload) end |
#request_response(req, metadata: {}) ⇒ Object
request_response sends a request to a GRPC server, and returns the response.
a list, multiple metadata for its key are sent
322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 |
# File 'src/ruby/lib/grpc/generic/active_call.rb', line 322 def request_response(req, metadata: {}) 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!() end @metadata_sent = true end batch_result = @call.run_batch(ops) @call. = batch_result. attach_status_results_and_complete_call(batch_result) (batch_result) end |
#run_server_bidi(gen_each_reply) ⇒ Object
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
460 461 462 463 464 465 466 467 468 |
# File 'src/ruby/lib/grpc/generic/active_call.rb', line 460 def run_server_bidi(gen_each_reply) bd = BidiCall.new(@call, @marshal, @unmarshal, metadata_received: @metadata_received, req_view: MultiReqView.new(self)) bd.run_on_server(gen_each_reply) end |
#send_initial_metadata ⇒ Object
Sends the initial metadata that has yet to be sent. Does nothing if metadata has already been sent for this call.
122 123 124 125 126 127 128 |
# File 'src/ruby/lib/grpc/generic/active_call.rb', line 122 def @send_initial_md_mutex.synchronize do return if @metadata_sent @metadata_tag = ActiveCall.client_invoke(@call, @metadata_to_send) @metadata_sent = true end end |
#send_status(code = OK, details = '', assert_finished = false, metadata: {}) ⇒ Object
send_status sends a status to the remote endpoint.
FINISHED. list, mulitple metadata for its key are sent
203 204 205 206 207 208 209 210 211 212 |
# File 'src/ruby/lib/grpc/generic/active_call.rb', line 203 def send_status(code = OK, details = '', assert_finished = false, metadata: {}) ops = { SEND_STATUS_FROM_SERVER => Struct::Status.new(code, details, ) } ops[RECV_CLOSE_ON_SERVER] = nil if assert_finished @call.run_batch(ops) nil end |
#server_streamer(req, metadata: {}) ⇒ Enumerator|nil
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.
a list, multiple metadata for its key are sent
390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 |
# File 'src/ruby/lib/grpc/generic/active_call.rb', line 390 def server_streamer(req, metadata: {}) 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!() end @metadata_sent = true end @call.run_batch(ops) replies = enum_for(:each_remote_read_then_finish) return replies unless block_given? replies.each { |r| yield r } rescue GRPC::Core::CallError => e finished # checks for Cancelled raise e end |
#server_unary_response(req, trailing_metadata: {}, code: Core::StatusCodes::OK, details: 'OK') ⇒ Object
214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 |
# File 'src/ruby/lib/grpc/generic/active_call.rb', line 214 def server_unary_response(req, trailing_metadata: {}, code: Core::StatusCodes::OK, details: 'OK') ops = {} @send_initial_md_mutex.synchronize do ops[SEND_INITIAL_METADATA] = @metadata_to_send unless @metadata_sent @metadata_sent = true end payload = @marshal.call(req) ops[SEND_MESSAGE] = payload ops[SEND_STATUS_FROM_SERVER] = Struct::Status.new( code, details, ) ops[RECV_CLOSE_ON_SERVER] = nil @call.run_batch(ops) end |
#single_req_view ⇒ Object
single_req_view provides a restricted view of this ActiveCall for use in a server request-response handler.
149 150 151 |
# File 'src/ruby/lib/grpc/generic/active_call.rb', line 149 def single_req_view SingleReqView.new(self) end |