Class: GRPC::ActiveCall

Inherits:
Object
  • Object
show all
Includes:
Core::CompletionType, Core::StatusCodes, Core::TimeConsts
Defined in:
lib/grpc/generic/active_call.rb

Overview

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

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Methods included from Core::TimeConsts

from_relative_time

Constructor Details

#initialize(call, q, marshal, unmarshal, deadline, finished_tag: nil, read_metadata_tag: nil, started: true) ⇒ 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.

Parameters:

  • call (Call) —

    the call used by the ActiveCall

  • q (CompletionQueue) —

    the completion queue used to accept the call

  • marshal (Function) —

    f(obj)->string that marshal requests

  • unmarshal (Function) —

    f(string)->obj that unmarshals responses

  • deadline (Fixnum) —

    the deadline for the call to complete

  • finished_tag (Object) (defaults to: nil) —

    the object used as the call’s finish tag, if the call has begun

  • read_metadata_tag (Object) (defaults to: nil) —

    the object used as the call’s finish tag, if the call has begun

  • started (true|false) (defaults to: true) —

    indicates if the call has begun



99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
 
# File 'lib/grpc/generic/active_call.rb', line 99

def initialize(call, q, marshal, unmarshal, deadline, finished_tag: nil,
               read_metadata_tag: nil, started: true)
  fail(ArgumentError, 'not a call') unless call.is_a? Core::Call
  unless q.is_a? Core::CompletionQueue
    fail(ArgumentError, 'not a CompletionQueue')
  end
  @call = call
  @cq = q
  @deadline = deadline
  @finished_tag = finished_tag
  @read_metadata_tag = read_metadata_tag
  @marshal = marshal
  @started = started
  @unmarshal = unmarshal
end

Instance Attribute Details

#deadline ⇒ Object (readonly)

Returns the value of attribute deadline.



47
48
49
 
# File 'lib/grpc/generic/active_call.rb', line 47

def deadline
  @deadline
end

Class Method Details

.client_invoke(call, q, _deadline, **kw) ⇒ 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

Parameters:

  • call (Call) —

    a call on which to start and invocation

  • q (CompletionQueue) —

    the completion queue

  • deadline (Fixnum, TimeSpec) —

    the deadline



63
64
65
66
67
68
69
70
71
72
73
 
# File 'lib/grpc/generic/active_call.rb', line 63

def self.client_invoke(call, q, _deadline, **kw)
  fail(ArgumentError, 'not a call') unless call.is_a? Core::Call
  unless q.is_a? Core::CompletionQueue
    fail(ArgumentError, 'not a CompletionQueue')
  end
  call.add_metadata(kw) if kw.length > 0
  client_metadata_read = Object.new
  finished_tag = Object.new
  call.invoke(q, client_metadata_read, finished_tag)
  [finished_tag, client_metadata_read]
end

Instance Method Details

#bidi_streamer(requests, **kw, &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.

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

Parameters:

  • requests (Object) —

    an Enumerable of requests to send

Returns:

  • (Enumerator, nil) —

    a response Enumerator



462
463
464
465
466
467
 
# File 'lib/grpc/generic/active_call.rb', line 462

def bidi_streamer(requests, **kw, &blk)
  start_call(**kw) unless @started
  bd = BidiCall.new(@call, @cq, @marshal, @unmarshal, @deadline,
                    @finished_tag)
  bd.run_on_client(requests, &blk)
end

#cancel ⇒ Object

Cancels the call.

Cancels the call. The call does not return any result, but once this it has been called, the call should eventually terminate. Due to potential races between the execution of the cancel and the in-flight request, the result of the call after calling #cancel is indeterminate:

  • the call may terminate with a BadStatus exception, with code=CANCELLED

  • the call may terminate with OK Status, and return a response

  • the call may terminate with a different BadStatus exception if that was happening



145
146
147
 
# File 'lib/grpc/generic/active_call.rb', line 145

def cancel
  @call.cancel
end

#cancelled ⇒ Object

indicates if the call is cancelled.



155
156
157
 
# File 'lib/grpc/generic/active_call.rb', line 155

def cancelled
  @cancelled ||= false
end

#client_streamer(requests, **kw) ⇒ 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.

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

Parameters:

  • requests (Object) —

    an Enumerable of requests to send

Returns:

  • (Object) —

    the response received from the server



398
399
400
401
402
403
404
405
 
# File 'lib/grpc/generic/active_call.rb', line 398

def client_streamer(requests, **kw)
  start_call(**kw) unless @started
  requests.each { |r| remote_send(r) }
  writes_done(false)
  response = remote_read
  finished unless response.is_a? Struct::Status
  response
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

Returns:

  • (Enumerator) —

    if no block was given



323
324
325
326
327
328
329
330
331
 
# File 'lib/grpc/generic/active_call.rb', line 323

def each_remote_read
  return enum_for(:each_remote_read) unless block_given?
  loop do
    resp = remote_read
    break if resp.is_a? Struct::Status  # is an OK status
    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

Returns:

  • (Enumerator) —

    if no block was given



353
354
355
356
357
358
359
360
361
362
363
364
 
# File 'lib/grpc/generic/active_call.rb', line 353

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.is_a? Struct::Status  # is an OK status
    if resp.nil?  # the last response was received, but not finished yet
      finished
      break
    end
    yield resp
  end
end

#finished ⇒ Object

finished waits until the call is completed.

It blocks until the remote endpoint acknowledges by sending a FINISHED event.



206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
 
# File 'lib/grpc/generic/active_call.rb', line 206

def finished
  ev = @cq.pluck(@finished_tag, INFINITE_FUTURE)
  begin
    fail "unexpected event: #{ev.inspect}" unless ev.type == FINISHED
    if @call.metadata.nil?
      @call.metadata = ev.result.metadata
    else
      @call.metadata.merge!(ev.result.metadata)
    end

    if ev.result.code != Core::StatusCodes::OK
      fail BadStatus.new(ev.result.code, ev.result.details)
    end
    res = ev.result
  ensure
    ev.close
  end
  res
end

#metadata ⇒ Object

Obtains the metadata of the call.

At the start of the call this will be nil. During the call this gets some values as soon as the other end of the connection acknowledges the request.

Returns:

  • this calls’s metadata



130
131
132
 
# File 'lib/grpc/generic/active_call.rb', line 130

def metadata
  @call.metadata
end

#multi_req_view ⇒ Object

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



161
162
163
 
# File 'lib/grpc/generic/active_call.rb', line 161

def multi_req_view
  MultiReqView.new(self)
end

#operation ⇒ Object

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



173
174
175
 
# File 'lib/grpc/generic/active_call.rb', line 173

def operation
  Operation.new(self)
end

#remote_read ⇒ Object

remote_read reads a response from the remote endpoint.

It blocks until the remote endpoint sends a READ or FINISHED event. On a READ, it returns the response after unmarshalling it. On FINISHED, it returns nil if the status is OK, otherwise raising BadStatus



280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
 
# File 'lib/grpc/generic/active_call.rb', line 280

def remote_read
  if @call.metadata.nil? && !@read_metadata_tag.nil?
    ev = @cq.pluck(@read_metadata_tag, INFINITE_FUTURE)
    assert_event_type(ev, CLIENT_METADATA_READ)
    @call.metadata = ev.result
    @read_metadata_tag = nil
  end

  @call.start_read(self)
  ev = @cq.pluck(self, INFINITE_FUTURE)
  begin
    assert_event_type(ev, READ)
    logger.debug("received req: #{ev.result.inspect}")
    unless ev.result.nil?
      logger.debug("received req.to_s: #{ev.result}")
      res = @unmarshal.call(ev.result.to_s)
      logger.debug("received_req (unmarshalled): #{res.inspect}")
      return res
    end
  ensure
    ev.close
  end
  logger.debug('found nil; the final response has been sent')
  nil
end

#remote_send(req, marshalled = false) ⇒ Object

remote_send sends a request to the remote endpoint.

It blocks until the remote endpoint acknowledges by sending a WRITE_ACCEPTED. req can be marshalled already.

marshalled.

Parameters:

  • req (Object, String) —

    the object to send or it’s marshal form.

  • marshalled (false, true) (defaults to: false) —

    indicates if the object is already



234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
 
# File 'lib/grpc/generic/active_call.rb', line 234

def remote_send(req, marshalled = false)
  assert_queue_is_ready
  logger.debug("sending #{req.inspect}, marshalled? #{marshalled}")
  if marshalled
    payload = req
  else
    payload = @marshal.call(req)
  end
  @call.start_write(Core::ByteBuffer.new(payload), self)

  # call queue#pluck, and wait for WRITE_ACCEPTED, so as not to return
  # until the flow control allows another send on this call.
  ev = @cq.pluck(self, INFINITE_FUTURE)
  begin
    assert_event_type(ev, WRITE_ACCEPTED)
  ensure
    ev.close
  end
end

#request_response(req, **kw) ⇒ Object

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

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

Parameters:

  • req (Object) —

    the request sent to the server

Returns:

  • (Object) —

    the response received from the server



375
376
377
378
379
380
381
382
 
# File 'lib/grpc/generic/active_call.rb', line 375

def request_response(req, **kw)
  start_call(**kw) unless @started
  remote_send(req)
  writes_done(false)
  response = remote_read
  finished unless response.is_a? Struct::Status
  response
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

Parameters:

  • gen_each_reply (Proc) —

    generates the BiDi stream replies



480
481
482
483
484
 
# File 'lib/grpc/generic/active_call.rb', line 480

def run_server_bidi(gen_each_reply)
  bd = BidiCall.new(@call, @cq, @marshal, @unmarshal, @deadline,
                    @finished_tag)
  bd.run_on_server(gen_each_reply)
end

#send_status(code = OK, details = '', assert_finished = false) ⇒ Object

send_status sends a status to the remote endpoint

FINISHED.

Parameters:

  • code (int) (defaults to: OK) —

    the status code to send

  • details (String) (defaults to: '') —

    details

  • assert_finished (true, false) (defaults to: false) —

    when true(default), waits for



260
261
262
263
264
265
266
267
268
269
270
271
272
 
# File 'lib/grpc/generic/active_call.rb', line 260

def send_status(code = OK, details = '', assert_finished = false)
  assert_queue_is_ready
  @call.start_write_status(code, details, self)
  ev = @cq.pluck(self, INFINITE_FUTURE)
  begin
    assert_event_type(ev, FINISH_ACCEPTED)
  ensure
    ev.close
  end
  logger.debug("Status sent: #{code}:'#{details}'")
  return finished if assert_finished
  nil
end

#server_streamer(req, **kw) ⇒ 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.

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 any keyword arguments are treated as metadata to be sent to the server.

Parameters:

  • req (Object) —

    the request sent to the server

Returns:

  • (Enumerator|nil) —

    a response Enumerator



424
425
426
427
428
429
430
431
 
# File 'lib/grpc/generic/active_call.rb', line 424

def server_streamer(req, **kw)
  start_call(**kw) unless @started
  remote_send(req)
  writes_done(false)
  replies = enum_for(:each_remote_read_then_finish)
  return replies unless block_given?
  replies.each { |r| yield r }
end

#shutdown ⇒ Object

indicates if the call is shutdown



150
151
152
 
# File 'lib/grpc/generic/active_call.rb', line 150

def shutdown
  @shutdown ||= false
end

#single_req_view ⇒ Object

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



167
168
169
 
# File 'lib/grpc/generic/active_call.rb', line 167

def single_req_view
  SingleReqView.new(self)
end

#status ⇒ Object

Obtains the status of the call.

this value is nil until the call completes

Returns:

  • this call’s status



119
120
121
 
# File 'lib/grpc/generic/active_call.rb', line 119

def status
  @call.status
end

#writes_done(assert_finished = true) ⇒ Object

writes_done indicates that all writes are completed.

It blocks until the remote endpoint acknowledges by sending a FINISHED event, unless assert_finished is set to false. Any calls to #remote_send after this call will fail.

FINISHED.

Parameters:

  • assert_finished (true, false) (defaults to: true) —

    when true(default), waits for



185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
 
# File 'lib/grpc/generic/active_call.rb', line 185

def writes_done(assert_finished = true)
  @call.writes_done(self)
  ev = @cq.pluck(self, INFINITE_FUTURE)
  begin
    assert_event_type(ev, FINISH_ACCEPTED)
    logger.debug("Writes done: waiting for finish? #{assert_finished}")
  ensure
    ev.close
  end

  return unless assert_finished
  ev = @cq.pluck(@finished_tag, INFINITE_FUTURE)
  fail 'unexpected nil event' if ev.nil?
  ev.close
  @call.status
end