Class: GRPC::ActiveCall

Inherits:
Object
  • Object
show all
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

Class Method Summary collapse

Instance Method Summary collapse

Methods included from Core::TimeConsts

from_relative_time

Constructor Details

#initialize(call, q, marshal, unmarshal, deadline, started: true, metadata_tag: 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.

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

  • metadata_tag (Object)

    the object use obtain metadata for clients

  • started (true|false)

    indicates if the call has begun


109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
# File 'src/ruby/lib/grpc/generic/active_call.rb', line 109

def initialize(call, q, marshal, unmarshal, deadline, started: true,
               metadata_tag: nil)
  fail(TypeError, '!Core::Call') unless call.is_a? Core::Call
  unless q.is_a? Core::CompletionQueue
    fail(TypeError, '!Core::CompletionQueue')
  end
  @call = call
  @cq = q
  @deadline = deadline
  @marshal = marshal
  @started = started
  @unmarshal = unmarshal
  @metadata_tag = 
  @op_notifier = nil
end

Instance Attribute Details

#deadlineObject (readonly)

Returns the value of attribute deadline


61
62
63
# File 'src/ruby/lib/grpc/generic/active_call.rb', line 61

def deadline
  @deadline
end

Class Method Details

.client_invoke(call, q, **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


77
78
79
80
81
82
83
84
85
86
# File 'src/ruby/lib/grpc/generic/active_call.rb', line 77

def self.client_invoke(call, q, **kw)
  fail(TypeError, '!Core::Call') unless call.is_a? Core::Call
  unless q.is_a? Core::CompletionQueue
    fail(TypeError, '!Core::CompletionQueue')
  end
   = Object.new
  call.run_batch(q, , INFINITE_FUTURE,
                 SEND_INITIAL_METADATA => kw)
  
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


414
415
416
417
418
419
420
# File 'src/ruby/lib/grpc/generic/active_call.rb', line 414

def bidi_streamer(requests, **kw, &blk)
  start_call(**kw) unless @started
  bd = BidiCall.new(@call, @cq, @marshal, @unmarshal,
                    metadata_tag: @metadata_tag)
  @metadata_tag = nil  # run_on_client ensures metadata is read
  bd.run_on_client(requests, @op_notifier, &blk)
end

#cancelledObject

cancelled indicates if the call was cancelled


132
133
134
# File 'src/ruby/lib/grpc/generic/active_call.rb', line 132

def cancelled
  !@call.status.nil? && @call.status.code == Core::StatusCodes::CANCELLED
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


344
345
346
347
348
349
350
351
352
353
354
# File 'src/ruby/lib/grpc/generic/active_call.rb', line 344

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
rescue GRPC::Core::CallError => e
  finished  # checks for Cancelled
  raise e
end

#each_remote_readEnumerator

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


267
268
269
270
271
272
273
274
# File 'src/ruby/lib/grpc/generic/active_call.rb', line 267

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_finishEnumerator

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


296
297
298
299
300
301
302
303
304
305
306
307
# File 'src/ruby/lib/grpc/generic/active_call.rb', line 296

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

#finishedObject

finished waits until a client call is completed.

It blocks until the remote endpoint acknowledges by sending a status.


178
179
180
181
182
183
184
185
186
187
188
189
190
191
# File 'src/ruby/lib/grpc/generic/active_call.rb', line 178

def finished
  batch_result = @call.run_batch(@cq, self, INFINITE_FUTURE,
                                 RECV_STATUS_ON_CLIENT => nil)
  unless batch_result.status.nil?
    if @call..nil?
      @call. = batch_result.status.
    else
      @call..merge!(batch_result.status.)
    end
  end
  @call.status = batch_result.status
  op_is_done
  batch_result.check_status
end

#multi_req_viewObject

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


138
139
140
# File 'src/ruby/lib/grpc/generic/active_call.rb', line 138

def multi_req_view
  MultiReqView.new(self)
end

#op_is_doneObject

Signals that an operation is done


446
447
448
449
# File 'src/ruby/lib/grpc/generic/active_call.rb', line 446

def op_is_done
  return if @op_notifier.nil?
  @op_notifier.notify(self)
end

#operationObject

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


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

def operation
  @op_notifier = Notifier.new
  Operation.new(self)
end

#output_metadataObject

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


127
128
129
# File 'src/ruby/lib/grpc/generic/active_call.rb', line 127

def 
  @output_metadata ||= {}
end

#remote_readObject

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


231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
# File 'src/ruby/lib/grpc/generic/active_call.rb', line 231

def remote_read
  ops = { RECV_MESSAGE => nil }
  ops[RECV_INITIAL_METADATA] = nil unless @metadata_tag.nil?
  batch_result = @call.run_batch(@cq, self, INFINITE_FUTURE, ops)
  unless @metadata_tag.nil?
    @call. = batch_result.
    @metadata_tag = nil
  end
  GRPC.logger.debug("received req: #{batch_result}")
  unless batch_result.nil? || batch_result.message.nil?
    GRPC.logger.debug("received req.to_s: #{batch_result.message}")
    res = @unmarshal.call(batch_result.message)
    GRPC.logger.debug("received_req (unmarshalled): #{res.inspect}")
    return res
  end
  GRPC.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 accepts the message.

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


200
201
202
203
204
# File 'src/ruby/lib/grpc/generic/active_call.rb', line 200

def remote_send(req, marshalled = false)
  GRPC.logger.debug("sending #{req}, marshalled? #{marshalled}")
  payload = marshalled ? req : @marshal.call(req)
  @call.run_batch(@cq, self, INFINITE_FUTURE, SEND_MESSAGE => payload)
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


318
319
320
321
322
323
324
325
326
327
328
# File 'src/ruby/lib/grpc/generic/active_call.rb', line 318

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
rescue GRPC::Core::CallError => e
  finished  # checks for Cancelled
  raise e
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


433
434
435
436
# File 'src/ruby/lib/grpc/generic/active_call.rb', line 433

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

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

send_status sends a status to the remote endpoint.

FINISHED.

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:

  • 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


216
217
218
219
220
221
222
223
# File 'src/ruby/lib/grpc/generic/active_call.rb', line 216

def send_status(code = OK, details = '', assert_finished = false, **kw)
  ops = {
    SEND_STATUS_FROM_SERVER => Struct::Status.new(code, details, kw)
  }
  ops[RECV_CLOSE_ON_SERVER] = nil if assert_finished
  @call.run_batch(@cq, self, INFINITE_FUTURE, ops)
  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


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

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 }
rescue GRPC::Core::CallError => e
  finished  # checks for Cancelled
  raise e
end

#single_req_viewObject

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


144
145
146
# File 'src/ruby/lib/grpc/generic/active_call.rb', line 144

def single_req_view
  SingleReqView.new(self)
end

#waitObject

Waits till an operation completes


439
440
441
442
443
# File 'src/ruby/lib/grpc/generic/active_call.rb', line 439

def wait
  return if @op_notifier.nil?
  GRPC.logger.debug("active_call.wait: on #{@op_notifier}")
  @op_notifier.wait
end

#writes_done(assert_finished = true) ⇒ Object

writes_done indicates that all writes are completed.

It blocks until the remote endpoint acknowledges with at status 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


163
164
165
166
167
168
169
170
171
172
173
# File 'src/ruby/lib/grpc/generic/active_call.rb', line 163

def writes_done(assert_finished = true)
  ops = {
    SEND_CLOSE_FROM_CLIENT => nil
  }
  ops[RECV_STATUS_ON_CLIENT] = nil if assert_finished
  batch_result = @call.run_batch(@cq, self, INFINITE_FUTURE, ops)
  return unless assert_finished
  @call.status = batch_result.status
  op_is_done
  batch_result.check_status
end