Class: GRPC::ClientStub

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

Overview

ClientStub represents a client connection to a gRPC server, and can be used to send requests.

Constant Summary collapse

DEFAULT_TIMEOUT =

Default timeout is infinity.

INFINITE_FUTURE

Constants included from GRPC::Core::StatusCodes

GRPC::Core::StatusCodes::ABORTED, GRPC::Core::StatusCodes::ALREADY_EXISTS, GRPC::Core::StatusCodes::CANCELLED, GRPC::Core::StatusCodes::DATA_LOSS, GRPC::Core::StatusCodes::DEADLINE_EXCEEDED, GRPC::Core::StatusCodes::FAILED_PRECONDITION, GRPC::Core::StatusCodes::INTERNAL, GRPC::Core::StatusCodes::INVALID_ARGUMENT, GRPC::Core::StatusCodes::NOT_FOUND, GRPC::Core::StatusCodes::OK, GRPC::Core::StatusCodes::OUT_OF_RANGE, GRPC::Core::StatusCodes::PERMISSION_DENIED, GRPC::Core::StatusCodes::RESOURCE_EXHAUSTED, GRPC::Core::StatusCodes::UNAUTHENTICATED, GRPC::Core::StatusCodes::UNAVAILABLE, GRPC::Core::StatusCodes::UNIMPLEMENTED, GRPC::Core::StatusCodes::UNKNOWN

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Methods included from GRPC::Core::TimeConsts

from_relative_time

Constructor Details

#initialize(host, creds, channel_override: nil, timeout: nil, propagate_mask: nil, channel_args: {}, interceptors: []) ⇒ ClientStub

Creates a new ClientStub.

Minimally, a stub is created with the just the host of the gRPC service it wishes to access, e.g.,

my_stub = ClientStub.new(example.host.com:50505,
                         :this_channel_is_insecure)

If a channel_override argument is passed, it will be used as the underlying channel. Otherwise, the channel_args argument will be used to construct a new underlying channel.

There are some specific keyword args that are not used to configure the channel:

  • :channel_override

when present, this must be a pre-created GRPC::Core::Channel. If it’s present the host and arbitrary keyword arg areignored, and the RPC connection uses this channel.

  • :timeout

when present, this is the default timeout used for calls

Parameters:

  • host (String) —

    the host the stub connects to

  • creds (Core::ChannelCredentials|Symbol) —

    the channel credentials, or :this_channel_is_insecure, which explicitly indicates that the client should be created with an insecure connection. Note: this argument is ignored if the channel_override argument is provided.

  • channel_override (Core::Channel) (defaults to: nil) —

    a pre-created channel

  • timeout (Number) (defaults to: nil) —

    the default timeout to use in requests

  • propagate_mask (Number) (defaults to: nil) —

    A bitwise combination of flags in GRPC::Core::PropagateMasks. Indicates how data should be propagated from parent server calls to child client calls if this client is being used within a gRPC server.

  • channel_args (Hash) (defaults to: {}) —

    the channel arguments. Note: this argument is ignored if the channel_override argument is provided.

  • interceptors (Array<GRPC::ClientInterceptor>) (defaults to: []) —

    An array of GRPC::ClientInterceptor objects that will be used for intercepting calls before they are executed Interceptors are an EXPERIMENTAL API.



98
99
100
101
102
103
104
105
106
107
108
109
110
111
 
# File 'src/ruby/lib/grpc/generic/client_stub.rb', line 98

def initialize(host, creds,
               channel_override: nil,
               timeout: nil,
               propagate_mask: nil,
               channel_args: {},
               interceptors: [])
  @ch = ClientStub.setup_channel(channel_override, host, creds,
                                 channel_args.dup)
  alt_host = channel_args[Core::Channel::SSL_TARGET]
  @host = alt_host.nil? ? host : alt_host
  @propagate_mask = propagate_mask
  @timeout = timeout.nil? ? DEFAULT_TIMEOUT : timeout
  @interceptors = InterceptorRegistry.new(interceptors)
end

Instance Attribute Details

#propagate_mask=(value) ⇒ Object (writeonly)

Allows users of the stub to modify the propagate mask.

This is an advanced feature for use when making calls to another gRPC server whilst running in the handler of an existing one.



56
57
58
 
# File 'src/ruby/lib/grpc/generic/client_stub.rb', line 56

def propagate_mask=(value)
  @propagate_mask = value
end

Class Method Details

.setup_channel(alt_chan, host, creds, channel_args = {}) ⇒ Object

setup_channel is used by #initialize to constuct a channel from its arguments.



33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
 
# File 'src/ruby/lib/grpc/generic/client_stub.rb', line 33

def self.setup_channel(alt_chan, host, creds, channel_args = {})
  unless alt_chan.nil?
    fail(TypeError, '!Channel') unless alt_chan.is_a?(Core::Channel)
    return alt_chan
  end
  if channel_args['grpc.primary_user_agent'].nil?
    channel_args['grpc.primary_user_agent'] = ''
  else
    channel_args['grpc.primary_user_agent'] += ' '
  end
  channel_args['grpc.primary_user_agent'] += "grpc-ruby/#{VERSION}"
  unless creds.is_a?(Core::ChannelCredentials) ||
         creds.is_a?(Core::XdsChannelCredentials) ||
         creds.is_a?(Symbol)
    fail(TypeError, 'creds is not a ChannelCredentials, XdsChannelCredentials, or Symbol')
  end
  Core::Channel.new(host, channel_args, creds)
end

Instance Method Details

#bidi_streamer(method, requests, marshal, unmarshal, deadline: nil, return_op: false, parent: nil, credentials: nil, metadata: {}, &blk) ⇒ Enumerator|nil|Operation

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.

Flow Control ==

This is a blocking call.

  • the call completes when the next call to provided block returns false

  • the execution block parameters are two objects for sending and receiving responses, each of which blocks waiting for flow control. E.g, calles to bidi_call#remote_send will wait until flow control allows another write before returning; and obviously calls to responses#next block until the next response is available.

Termination ==

As well as sending and receiving messages, the block passed to the function is also responsible for:

  • calling bidi_call#writes_done to indicate no further reqs will be sent.

  • returning false if once the bidi stream is functionally completed.

Note that response#next will indicate that there are no further responses by throwing StopIteration, but can only happen either if bidi_call#writes_done is called.

To properly terminate the RPC, the responses should be completely iterated through; one way to do this is to loop on responses#next until no further responses are available.

Errors ==

An RuntimeError is raised if

  • the server responds with a non-OK status when any response is

  • retrieved

  • the deadline is exceeded

Return Value ==

if the return_op is false, the return value is an Enumerator of the results, unless a block is provided, in which case the block is executed with each response.

if return_op is true, the function returns an Operation whose #execute method runs the Bidi call. Again, Operation#execute either calls a given block with each response or returns an Enumerator of the responses.

Parameters:

  • method (String) —

    the RPC method to call on the GRPC server

  • requests (Object) —

    an Enumerable of requests to send

  • marshal (Function) —

    f(obj)->string that marshals requests

  • unmarshal (Function) —

    f(string)->obj that unmarshals responses

  • deadline (Time) (defaults to: nil) —

    (optional) the time the request should complete

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

    return an Operation if true

  • parent (Core::Call) (defaults to: nil) —

    a prior call whose reserved metadata will be propagated by this one.

  • credentials (Core::CallCredentials) (defaults to: nil) —

    credentials to use when making the call

  • metadata (Hash) (defaults to: {}) —

    metadata to be sent to the server

  • blk (Block) —

    when provided, is executed for each response

Returns:

  • (Enumerator|nil|Operation) —

    as discussed above



440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
 
# File 'src/ruby/lib/grpc/generic/client_stub.rb', line 440

def bidi_streamer(method, requests, marshal, unmarshal,
                  deadline: nil,
                  return_op: false,
                  parent: nil,
                  credentials: nil,
                  metadata: {},
                  &blk)
  c = new_active_call(method, marshal, unmarshal,
                      deadline: deadline,
                      parent: parent,
                      credentials: credentials)
  interception_context = @interceptors.build_context
  intercept_args = {
    method: method,
    requests: requests,
    call: c.interceptable,
    metadata: metadata
  }
  if return_op
    # return the operation view of the active_call; define #execute
    # as a new method for this instance that invokes #bidi_streamer
    c.merge_metadata_to_send(metadata)
    op = c.operation
    op.define_singleton_method(:execute) do
      interception_context.intercept!(:bidi_streamer, intercept_args) do
        c.bidi_streamer(requests, &blk)
      end
    end
    op
  else
    interception_context.intercept!(:bidi_streamer, intercept_args) do
      c.bidi_streamer(requests, metadata: metadata, &blk)
    end
  end
end

#client_streamer(method, requests, marshal, unmarshal, deadline: nil, return_op: false, parent: nil, credentials: nil, metadata: {}) ⇒ Object|Operation

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.

Flow Control ==

This is a blocking call.

  • it does not return until a response is received.

  • each requests is sent only when GRPC core’s flow control allows it to be sent.

Errors ==

An RuntimeError is raised if

  • the server responds with a non-OK status

  • the deadline is exceeded

Return Value ==

If return_op is false, the call consumes the requests and returns the response.

If return_op is true, the call returns the response.

Parameters:

  • method (String) —

    the RPC method to call on the GRPC server

  • requests (Object) —

    an Enumerable of requests to send

  • marshal (Function) —

    f(obj)->string that marshals requests

  • unmarshal (Function) —

    f(string)->obj that unmarshals responses

  • deadline (Time) (defaults to: nil) —

    (optional) the time the request should complete

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

    return an Operation if true

  • parent (Core::Call) (defaults to: nil) —

    a prior call whose reserved metadata will be propagated by this one.

  • credentials (Core::CallCredentials) (defaults to: nil) —

    credentials to use when making the call

  • metadata (Hash) (defaults to: {}) —

    metadata to be sent to the server

Returns:

  • (Object|Operation) —

    the response received from the server



227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
 
# File 'src/ruby/lib/grpc/generic/client_stub.rb', line 227

def client_streamer(method, requests, marshal, unmarshal,
                    deadline: nil,
                    return_op: false,
                    parent: nil,
                    credentials: nil,
                    metadata: {})
  c = new_active_call(method, marshal, unmarshal,
                      deadline: deadline,
                      parent: parent,
                      credentials: credentials)
  interception_context = @interceptors.build_context
  intercept_args = {
    method: method,
    requests: requests,
    call: c.interceptable,
    metadata: metadata
  }
  if return_op
    # return the operation view of the active_call; define #execute as a
    # new method for this instance that invokes #client_streamer.
    c.merge_metadata_to_send(metadata)
    op = c.operation
    op.define_singleton_method(:execute) do
      interception_context.intercept!(:client_streamer, intercept_args) do
        c.client_streamer(requests)
      end
    end
    op
  else
    interception_context.intercept!(:client_streamer, intercept_args) do
      c.client_streamer(requests, metadata: metadata)
    end
  end
end

#request_response(method, req, marshal, unmarshal, deadline: nil, return_op: false, parent: nil, credentials: nil, metadata: {}) ⇒ Object

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

Flow Control ==

This is a blocking call.

  • it does not return until a response is received.

  • the requests is sent only when GRPC core’s flow control allows it to be sent.

Errors ==

An RuntimeError is raised if

  • the server responds with a non-OK status

  • the deadline is exceeded

Return Value ==

If return_op is false, the call returns the response

If return_op is true, the call returns an Operation, calling execute on the Operation returns the response.

Parameters:

  • method (String) —

    the RPC method to call on the GRPC server

  • req (Object) —

    the request sent to the server

  • marshal (Function) —

    f(obj)->string that marshals requests

  • unmarshal (Function) —

    f(string)->obj that unmarshals responses

  • deadline (Time) (defaults to: nil) —

    (optional) the time the request should complete

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

    return an Operation if true

  • parent (Core::Call) (defaults to: nil) —

    a prior call whose reserved metadata will be propagated by this one.

  • credentials (Core::CallCredentials) (defaults to: nil) —

    credentials to use when making the call

  • metadata (Hash) (defaults to: {}) —

    metadata to be sent to the server

Returns:

  • (Object) —

    the response received from the server



150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
 
# File 'src/ruby/lib/grpc/generic/client_stub.rb', line 150

def request_response(method, req, marshal, unmarshal,
                     deadline: nil,
                     return_op: false,
                     parent: nil,
                     credentials: nil,
                     metadata: {})
  c = new_active_call(method, marshal, unmarshal,
                      deadline: deadline,
                      parent: parent,
                      credentials: credentials)
  interception_context = @interceptors.build_context
  intercept_args = {
    method: method,
    request: req,
    call: c.interceptable,
    metadata: metadata
  }
  if return_op
    # return the operation view of the active_call; define #execute as a
    # new method for this instance that invokes #request_response.
    c.merge_metadata_to_send(metadata)
    op = c.operation
    op.define_singleton_method(:execute) do
      interception_context.intercept!(:request_response, intercept_args) do
        c.request_response(req, metadata: metadata)
      end
    end
    op
  else
    interception_context.intercept!(:request_response, intercept_args) do
      c.request_response(req, metadata: metadata)
    end
  end
end

#server_streamer(method, req, marshal, unmarshal, deadline: nil, return_op: false, parent: nil, credentials: nil, metadata: {}, &blk) ⇒ Enumerator|Operation|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.

Flow Control ==

This is a blocking call.

  • the request is sent only when GRPC core’s flow control allows it to be sent.

  • the request will not complete until the server sends the final response followed by a status message.

Errors ==

An RuntimeError is raised if

  • the server responds with a non-OK status when any response is

  • retrieved

  • the deadline is exceeded

Return Value ==

if the return_op is false, the return value is an Enumerator of the results, unless a block is provided, in which case the block is executed with each response.

if return_op is true, the function returns an Operation whose #execute method runs server streamer call. Again, Operation#execute either calls the given block with each response or returns an Enumerator of the responses.

Keyword Args ==

Unspecified keyword arguments are treated as metadata to be sent to the server.

Parameters:

  • method (String) —

    the RPC method to call on the GRPC server

  • req (Object) —

    the request sent to the server

  • marshal (Function) —

    f(obj)->string that marshals requests

  • unmarshal (Function) —

    f(string)->obj that unmarshals responses

  • deadline (Time) (defaults to: nil) —

    (optional) the time the request should complete

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

    return an Operation if true

  • parent (Core::Call) (defaults to: nil) —

    a prior call whose reserved metadata will be propagated by this one.

  • credentials (Core::CallCredentials) (defaults to: nil) —

    credentials to use when making the call

  • metadata (Hash) (defaults to: {}) —

    metadata to be sent to the server

  • blk (Block) —

    when provided, is executed for each response

Returns:

  • (Enumerator|Operation|nil) —

    as discussed above



318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
 
# File 'src/ruby/lib/grpc/generic/client_stub.rb', line 318

def server_streamer(method, req, marshal, unmarshal,
                    deadline: nil,
                    return_op: false,
                    parent: nil,
                    credentials: nil,
                    metadata: {},
                    &blk)
  c = new_active_call(method, marshal, unmarshal,
                      deadline: deadline,
                      parent: parent,
                      credentials: credentials)
  interception_context = @interceptors.build_context
  intercept_args = {
    method: method,
    request: req,
    call: c.interceptable,
    metadata: metadata
  }
  if return_op
    # return the operation view of the active_call; define #execute
    # as a new method for this instance that invokes #server_streamer
    c.merge_metadata_to_send(metadata)
    op = c.operation
    op.define_singleton_method(:execute) do
      interception_context.intercept!(:server_streamer, intercept_args) do
        c.server_streamer(req, &blk)
      end
    end
    op
  else
    interception_context.intercept!(:server_streamer, intercept_args) do
      c.server_streamer(req, metadata: metadata, &blk)
    end
  end
end