class GRPC::Core::Server
Private Class Methods
cq = CompletionQueue.new
click to toggle source
server = Server.new(cq, {'arg1': 'value1'})
Initializes server instances.
static VALUE grpc_rb_server_init(VALUE self, VALUE cqueue, 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_rb_get_wrapped_completion_queue(cqueue);
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;
/* Add the cq as the server's mark object. This ensures the ruby cq can't be
GCed before the server */
wrapper->mark = cqueue;
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 = ::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);
}
cq = CompletionQueue.new
click to toggle source
server = Server.new(cq, {'arg1': 'value1'})
... // do stuff with server
...
... // to shutdown the server
destroy(cq)
... // to shutdown the server with a timeout server.destroy(cq, timeout)
Destroys server instances.
static VALUE grpc_rb_server_destroy(int argc, VALUE *argv, VALUE self) {
VALUE cqueue = Qnil;
VALUE timeout = Qnil;
grpc_completion_queue *cq = NULL;
grpc_event ev;
grpc_rb_server *s = NULL;
/* "11" == 1 mandatory args, 1 (timeout) is optional */
rb_scan_args(argc, argv, "11", &cqueue, &timeout);
cq = grpc_rb_get_wrapped_completion_queue(cqueue);
TypedData_Get_Struct(self, grpc_rb_server, &grpc_rb_server_data_type, s);
if (s->wrapped != NULL) {
grpc_server_shutdown_and_notify(s->wrapped, cq, NULL);
ev = grpc_rb_completion_queue_pluck_event(cqueue, Qnil, timeout);
if (!ev.success) {
rb_warn("server shutdown failed, cancelling the calls, objects may leak");
grpc_server_cancel_all_calls(s->wrapped);
return Qfalse;
}
grpc_server_destroy(s->wrapped);
s->wrapped = NULL;
}
return Qtrue;
}
Also aliased as: close
initialize_copy(p1)
click to toggle source
Clones Server instances.
Gives Server a consistent implementation of Ruby's object copy/dup protocol.
static VALUE grpc_rb_server_init_copy(VALUE copy, VALUE orig) {
grpc_rb_server *orig_srv = NULL;
grpc_rb_server *copy_srv = NULL;
if (copy == orig) {
return copy;
}
/* Raise an error if orig is not a server object or a subclass. */
if (TYPE(orig) != T_DATA ||
RDATA(orig)->dfree != (RUBY_DATA_FUNC)grpc_rb_server_free) {
rb_raise(rb_eTypeError, "not a %s", rb_obj_classname(grpc_rb_cServer));
}
TypedData_Get_Struct(orig, grpc_rb_server, &grpc_rb_server_data_type,
orig_srv);
TypedData_Get_Struct(copy, grpc_rb_server, &grpc_rb_server_data_type,
copy_srv);
/* use ruby's MEMCPY to make a byte-for-byte copy of the server wrapper
object. */
MEMCPY(copy_srv, orig_srv, grpc_rb_server, 1);
return copy;
}
cq = CompletionQueue.new
click to toggle source
tag = Object.new
timeout = 10
request_call(cqueue, tag, timeout)
Requests notification of a new call on a server.
static VALUE grpc_rb_server_request_call(VALUE self, VALUE cqueue,
VALUE tag_new, VALUE timeout) {
grpc_rb_server *s = NULL;
grpc_call *call = NULL;
grpc_event ev;
grpc_call_error err;
request_call_stack st;
VALUE result;
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;
} else {
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,
grpc_rb_get_wrapped_completion_queue(cqueue),
grpc_rb_get_wrapped_completion_queue(cqueue),
ROBJECT(tag_new));
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 = grpc_rb_completion_queue_pluck_event(cqueue, tag_new, timeout);
if (ev.type == GRPC_QUEUE_TIMEOUT) {
grpc_request_call_stack_cleanup(&st);
return Qnil;
}
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, rb_str_new2(st.details.method),
rb_str_new2(st.details.host),
rb_funcall(rb_cTime, id_at, 2, INT2NUM(deadline.tv_sec),
INT2NUM(deadline.tv_nsec)),
grpc_rb_md_ary_to_h(&st.md_ary), grpc_rb_wrap_call(call), NULL);
grpc_request_call_stack_cleanup(&st);
return result;
}
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);
if (s->wrapped == NULL) {
rb_raise(rb_eRuntimeError, "destroyed!");
} else {
grpc_server_start(s->wrapped);
}
return Qnil;
}