Class: GRPC::BidiCall

Inherits:
Object
  • Object
show all
Includes:
Core::CallOps, Core::StatusCodes, Core::TimeConsts
Defined in:
src/ruby/lib/grpc/generic/bidi_call.rb

Overview

The BiDiCall class orchestrates exection of a BiDi stream on a client or server.

Instance Method Summary collapse

Methods included from Core::TimeConsts

from_relative_time

Constructor Details

#initialize(call, q, marshal, unmarshal, metadata_tag: nil) ⇒ BidiCall

Creates a BidiCall.

BidiCall 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 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 BidiCall#run is 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

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

    tag object used to collect metadata



60
61
62
63
64
65
66
67
68
69
70
71
72
 
# File 'src/ruby/lib/grpc/generic/bidi_call.rb', line 60

def initialize(call, q, marshal, unmarshal, metadata_tag: nil)
  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
  @marshal = marshal
  @op_notifier = nil  # signals completion on clients
  @readq = Queue.new
  @unmarshal = unmarshal
  @metadata_tag = metadata_tag
end

Instance Method Details

#run_on_client(requests, op_notifier, &blk) ⇒ Object

Begins orchestration of the Bidi stream for a client sending requests.

The method either returns an Enumerator of the responses, or accepts a block that can be invoked with each response.

Parameters:

  • requests —

    the Enumerable of requests to send

Returns:

  • an Enumerator of requests to yield



82
83
84
85
86
87
 
# File 'src/ruby/lib/grpc/generic/bidi_call.rb', line 82

def run_on_client(requests, op_notifier, &blk)
  @op_notifier = op_notifier
  @enq_th = Thread.new { write_loop(requests) }
  @loop_th = start_read_loop
  each_queued_msg(&blk)
end

#run_on_server(gen_each_reply) ⇒ Object

Begins orchestration of the Bidi stream for a server generating replies.

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.



100
101
102
103
104
 
# File 'src/ruby/lib/grpc/generic/bidi_call.rb', line 100

def run_on_server(gen_each_reply)
  replys = gen_each_reply.call(each_queued_msg)
  @loop_th = start_read_loop(is_client: false)
  write_loop(replys, is_client: false)
end