class GRPC::Core::Server

Private Class Methods

server = Server.new({'arg1': 'value1'}) click to toggle source

Initializes server instances.

static VALUE grpc_rb_server_init(VALUE self, VALUE channel_args) {
  grpc_completion_queue* cq = NULL;
  grpc_rb_server* wrapper = NULL;
  grpc_server* srv = NULL;
  grpc_channel_args args;
  MEMZERO(&args, grpc_channel_args, 1);

  cq = grpc_completion_queue_create_for_pluck(NULL);
  TypedData_Get_Struct(self, grpc_rb_server, &grpc_rb_server_data_type,
                       wrapper);
  grpc_rb_hash_convert_to_channel_args(channel_args, &args);
  srv = grpc_server_create(&args, NULL);

  if (args.args != NULL) {
    xfree(args.args); /* Allocated by grpc_rb_hash_convert_to_channel_args */
  }
  if (srv == NULL) {
    rb_raise(rb_eRuntimeError, "could not create a gRPC server, not sure why");
  }
  grpc_server_register_completion_queue(srv, cq, NULL);
  wrapper->wrapped = srv;
  wrapper->queue = cq;

  return self;
}

Private Instance Methods

// insecure port click to toggle source

insecure_server = Server.new(cq, {'arg1': 'value1'})

add_http2_port('mydomain:50051', :this_port_is_insecure)

// secure port server_creds = … secure_server = Server.new(cq, {'arg1': 'value1'}) secure_server.add_http_port('mydomain:50051', server_creds)

Adds a http2 port to server

static VALUE grpc_rb_server_add_http2_port(VALUE self, VALUE port,
                                           VALUE rb_creds) {
  grpc_rb_server* s = NULL;
  grpc_server_credentials* creds = NULL;
  int recvd_port = 0;

  TypedData_Get_Struct(self, grpc_rb_server, &grpc_rb_server_data_type, s);
  if (s->wrapped == NULL) {
    rb_raise(rb_eRuntimeError, "destroyed!");
    return Qnil;
  } else if (TYPE(rb_creds) == T_SYMBOL) {
    if (id_insecure_server != SYM2ID(rb_creds)) {
      rb_raise(rb_eTypeError, "bad creds symbol, want :this_port_is_insecure");
      return Qnil;
    }
    recvd_port =
        grpc_server_add_insecure_http2_port(s->wrapped, StringValueCStr(port));
    if (recvd_port == 0) {
      rb_raise(rb_eRuntimeError,
               "could not add port %s to server, not sure why",
               StringValueCStr(port));
    }
  } else {
    creds = grpc_rb_get_wrapped_server_credentials(rb_creds);
    recvd_port = grpc_server_add_secure_http2_port(
        s->wrapped, StringValueCStr(port), creds);
    if (recvd_port == 0) {
      rb_raise(rb_eRuntimeError,
               "could not add secure port %s to server, not sure why",
               StringValueCStr(port));
    }
  }
  return INT2NUM(recvd_port);
}

close()

Alias for: destroy

server = Server.new({'arg1': 'value1'}) click to toggle source

... // do stuff with server

...

... // initiate server shutdown

shutdown_and_notify(timeout)

... // to shutdown the server

destroy()

Destroys server instances.

static VALUE grpc_rb_server_destroy(VALUE self) {
  grpc_rb_server* s = NULL;
  TypedData_Get_Struct(self, grpc_rb_server, &grpc_rb_server_data_type, s);
  grpc_rb_server_maybe_destroy(s);
  return Qnil;
}

Also aliased as: close

request_call click to toggle source

Requests notification of a new call on a server.

static VALUE grpc_rb_server_request_call(VALUE self) {
  grpc_rb_server* s = NULL;
  grpc_call* call = NULL;
  grpc_event ev;
  grpc_call_error err;
  request_call_stack st;
  VALUE result;
  void* tag = (void*)&st;
  grpc_completion_queue* call_queue =
      grpc_completion_queue_create_for_pluck(NULL);
  gpr_timespec deadline;

  TypedData_Get_Struct(self, grpc_rb_server, &grpc_rb_server_data_type, s);
  if (s->wrapped == NULL) {
    rb_raise(rb_eRuntimeError, "destroyed!");
    return Qnil;
  }
  grpc_request_call_stack_init(&st);
  /* call grpc_server_request_call, then wait for it to complete using
   * pluck_event */
  err = grpc_server_request_call(s->wrapped, &call, &st.details, &st.md_ary,
                                 call_queue, s->queue, tag);
  if (err != GRPC_CALL_OK) {
    grpc_request_call_stack_cleanup(&st);
    rb_raise(grpc_rb_eCallError,
             "grpc_server_request_call failed: %s (code=%d)",
             grpc_call_error_detail_of(err), err);
    return Qnil;
  }

  ev = rb_completion_queue_pluck(s->queue, tag,
                                 gpr_inf_future(GPR_CLOCK_REALTIME), NULL);
  if (!ev.success) {
    grpc_request_call_stack_cleanup(&st);
    rb_raise(grpc_rb_eCallError, "request_call completion failed");
    return Qnil;
  }

  /* build the NewServerRpc struct result */
  deadline = gpr_convert_clock_type(st.details.deadline, GPR_CLOCK_REALTIME);
  result = rb_struct_new(
      grpc_rb_sNewServerRpc, grpc_rb_slice_to_ruby_string(st.details.method),
      grpc_rb_slice_to_ruby_string(st.details.host),
      rb_funcall(rb_cTime, id_at, 2, INT2NUM(deadline.tv_sec),
                 INT2NUM(deadline.tv_nsec / 1000)),
      grpc_rb_md_ary_to_h(&st.md_ary), grpc_rb_wrap_call(call, call_queue),
      NULL);
  grpc_request_call_stack_cleanup(&st);
  return result;
}

shutdown_and_notify(p1) click to toggle source

static VALUE grpc_rb_server_shutdown_and_notify(VALUE self, VALUE timeout) {
  gpr_timespec deadline;
  grpc_rb_server* s = NULL;

  TypedData_Get_Struct(self, grpc_rb_server, &grpc_rb_server_data_type, s);
  if (TYPE(timeout) == T_NIL) {
    deadline = gpr_inf_future(GPR_CLOCK_REALTIME);
  } else {
    deadline = grpc_rb_time_timeval(timeout, /* absolute time*/ 0);
  }

  grpc_rb_server_maybe_shutdown_and_notify(s, deadline);

  return Qnil;
}

start() click to toggle source

static VALUE grpc_rb_server_start(VALUE self) {
  grpc_rb_server* s = NULL;
  TypedData_Get_Struct(self, grpc_rb_server, &grpc_rb_server_data_type, s);

  grpc_ruby_fork_guard();
  if (s->wrapped == NULL) {
    rb_raise(rb_eRuntimeError, "destroyed!");
  } else {
    grpc_server_start(s->wrapped);
  }
  return Qnil;
}