/* * * Copyright 2015, Google Inc. * All rights reserved. * * Redistribution and use in source and binary forms, with or without * modification, are permitted provided that the following conditions are * met: * * * Redistributions of source code must retain the above copyright * notice, this list of conditions and the following disclaimer. * * Redistributions in binary form must reproduce the above * copyright notice, this list of conditions and the following disclaimer * in the documentation and/or other materials provided with the * distribution. * * Neither the name of Google Inc. nor the names of its * contributors may be used to endorse or promote products derived from * this software without specific prior written permission. * * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. * */ #include #include #include "rb_byte_buffer.h" #include "rb_channel.h" #include "rb_grpc_imports.generated.h" #include #include #include #include #include #include "rb_call.h" #include "rb_channel_args.h" #include "rb_channel_credentials.h" #include "rb_completion_queue.h" #include "rb_grpc.h" #include "rb_server.h" /* id_channel is the name of the hidden ivar that preserves a reference to the * channel on a call, so that calls are not GCed before their channel. */ static ID id_channel; /* id_target is the name of the hidden ivar that preserves a reference to the * target string used to create the call, preserved so that it does not get * GCed before the channel */ static ID id_target; /* id_insecure_channel is used to indicate that a channel is insecure */ static VALUE id_insecure_channel; /* grpc_rb_cChannel is the ruby class that proxies grpc_channel. */ static VALUE grpc_rb_cChannel = Qnil; /* Used during the conversion of a hash to channel args during channel setup */ static VALUE grpc_rb_cChannelArgs; /* grpc_rb_channel wraps a grpc_channel. */ typedef struct grpc_rb_channel { VALUE credentials; /* The actual channel */ grpc_channel *wrapped; int request_safe_destroy; int safe_to_destroy; grpc_connectivity_state current_connectivity_state; int mu_init_done; int abort_watch_connectivity_state; gpr_mu channel_mu; gpr_cv channel_cv; } grpc_rb_channel; /* Forward declarations of functions involved in temporary fix to * https://github.com/grpc/grpc/issues/9941 */ static void grpc_rb_channel_try_register_connection_polling( grpc_rb_channel *wrapper); static void grpc_rb_channel_safe_destroy(grpc_rb_channel *wrapper); static void *wait_until_channel_polling_thread_started_no_gil(void *); static void wait_until_channel_polling_thread_started_unblocking_func(void *); static grpc_completion_queue *channel_polling_cq; static gpr_mu global_connection_polling_mu; static gpr_cv global_connection_polling_cv; static int abort_channel_polling = 0; static int channel_polling_thread_started = 0; /* Destroys Channel instances. */ static void grpc_rb_channel_free(void *p) { grpc_rb_channel *ch = NULL; if (p == NULL) { return; }; ch = (grpc_rb_channel *)p; if (ch->wrapped != NULL) { grpc_rb_channel_safe_destroy(ch); ch->wrapped = NULL; } if (ch->mu_init_done) { gpr_mu_destroy(&ch->channel_mu); gpr_cv_destroy(&ch->channel_cv); } xfree(p); } /* Protects the mark object from GC */ static void grpc_rb_channel_mark(void *p) { grpc_rb_channel *channel = NULL; if (p == NULL) { return; } channel = (grpc_rb_channel *)p; if (channel->credentials != Qnil) { rb_gc_mark(channel->credentials); } } static rb_data_type_t grpc_channel_data_type = {"grpc_channel", {grpc_rb_channel_mark, grpc_rb_channel_free, GRPC_RB_MEMSIZE_UNAVAILABLE, {NULL, NULL}}, NULL, NULL, #ifdef RUBY_TYPED_FREE_IMMEDIATELY RUBY_TYPED_FREE_IMMEDIATELY #endif }; /* Allocates grpc_rb_channel instances. */ static VALUE grpc_rb_channel_alloc(VALUE cls) { grpc_rb_channel *wrapper = ALLOC(grpc_rb_channel); wrapper->wrapped = NULL; wrapper->credentials = Qnil; return TypedData_Wrap_Struct(cls, &grpc_channel_data_type, wrapper); } /* call-seq: insecure_channel = Channel:new("myhost:8080", {'arg1': 'value1'}, :this_channel_is_insecure) creds = ... secure_channel = Channel:new("myhost:443", {'arg1': 'value1'}, creds) Creates channel instances. */ static VALUE grpc_rb_channel_init(int argc, VALUE *argv, VALUE self) { VALUE channel_args = Qnil; VALUE credentials = Qnil; VALUE target = Qnil; grpc_rb_channel *wrapper = NULL; grpc_channel *ch = NULL; grpc_channel_credentials *creds = NULL; char *target_chars = NULL; grpc_channel_args args; MEMZERO(&args, grpc_channel_args, 1); grpc_ruby_once_init(); rb_thread_call_without_gvl( wait_until_channel_polling_thread_started_no_gil, NULL, wait_until_channel_polling_thread_started_unblocking_func, NULL); /* "3" == 3 mandatory args */ rb_scan_args(argc, argv, "3", &target, &channel_args, &credentials); TypedData_Get_Struct(self, grpc_rb_channel, &grpc_channel_data_type, wrapper); wrapper->mu_init_done = 0; target_chars = StringValueCStr(target); grpc_rb_hash_convert_to_channel_args(channel_args, &args); if (TYPE(credentials) == T_SYMBOL) { if (id_insecure_channel != SYM2ID(credentials)) { rb_raise(rb_eTypeError, "bad creds symbol, want :this_channel_is_insecure"); return Qnil; } ch = grpc_insecure_channel_create(target_chars, &args, NULL); } else { wrapper->credentials = credentials; creds = grpc_rb_get_wrapped_channel_credentials(credentials); ch = grpc_secure_channel_create(creds, target_chars, &args, NULL); } GPR_ASSERT(ch); wrapper->wrapped = ch; gpr_mu_init(&wrapper->channel_mu); gpr_cv_init(&wrapper->channel_cv); wrapper->mu_init_done = 1; gpr_mu_lock(&wrapper->channel_mu); wrapper->abort_watch_connectivity_state = 0; wrapper->current_connectivity_state = grpc_channel_check_connectivity_state(wrapper->wrapped, 0); wrapper->safe_to_destroy = 0; wrapper->request_safe_destroy = 0; gpr_cv_broadcast(&wrapper->channel_cv); gpr_mu_unlock(&wrapper->channel_mu); grpc_rb_channel_try_register_connection_polling(wrapper); if (args.args != NULL) { xfree(args.args); /* Allocated by grpc_rb_hash_convert_to_channel_args */ } if (ch == NULL) { rb_raise(rb_eRuntimeError, "could not create an rpc channel to target:%s", target_chars); return Qnil; } rb_ivar_set(self, id_target, target); wrapper->wrapped = ch; return self; } /* call-seq: ch.connectivity_state -> state ch.connectivity_state(true) -> state Indicates the current state of the channel, whose value is one of the constants defined in GRPC::Core::ConnectivityStates. It also tries to connect if the chennel is idle in the second form. */ static VALUE grpc_rb_channel_get_connectivity_state(int argc, VALUE *argv, VALUE self) { VALUE try_to_connect_param = Qfalse; int grpc_try_to_connect = 0; grpc_rb_channel *wrapper = NULL; grpc_channel *ch = NULL; /* "01" == 0 mandatory args, 1 (try_to_connect) is optional */ rb_scan_args(argc, argv, "01", &try_to_connect_param); grpc_try_to_connect = RTEST(try_to_connect_param) ? 1 : 0; TypedData_Get_Struct(self, grpc_rb_channel, &grpc_channel_data_type, wrapper); ch = wrapper->wrapped; if (ch == NULL) { rb_raise(rb_eRuntimeError, "closed!"); return Qnil; } return LONG2NUM(grpc_channel_check_connectivity_state(wrapper->wrapped, grpc_try_to_connect)); } typedef struct watch_state_stack { grpc_rb_channel *wrapper; gpr_timespec deadline; int last_state; } watch_state_stack; static void *watch_channel_state_without_gvl(void *arg) { watch_state_stack *stack = (watch_state_stack *)arg; gpr_timespec deadline = stack->deadline; grpc_rb_channel *wrapper = stack->wrapper; int last_state = stack->last_state; void *return_value = (void *)0; gpr_mu_lock(&wrapper->channel_mu); while (wrapper->current_connectivity_state == last_state && !wrapper->request_safe_destroy && !wrapper->safe_to_destroy && !wrapper->abort_watch_connectivity_state && gpr_time_cmp(deadline, gpr_now(GPR_CLOCK_REALTIME)) > 0) { gpr_cv_wait(&wrapper->channel_cv, &wrapper->channel_mu, deadline); } if (wrapper->current_connectivity_state != last_state) { return_value = (void *)1; } gpr_mu_unlock(&wrapper->channel_mu); return return_value; } static void watch_channel_state_unblocking_func(void *arg) { grpc_rb_channel *wrapper = (grpc_rb_channel *)arg; gpr_log(GPR_DEBUG, "GRPC_RUBY: watch channel state unblocking func called"); gpr_mu_lock(&wrapper->channel_mu); wrapper->abort_watch_connectivity_state = 1; gpr_cv_broadcast(&wrapper->channel_cv); gpr_mu_unlock(&wrapper->channel_mu); } /* Wait until the channel's connectivity state becomes different from * "last_state", or "deadline" expires. * Returns true if the the channel's connectivity state becomes * different from "last_state" within "deadline". * Returns false if "deadline" expires before the channel's connectivity * state changes from "last_state". * */ static VALUE grpc_rb_channel_watch_connectivity_state(VALUE self, VALUE last_state, VALUE deadline) { grpc_rb_channel *wrapper = NULL; watch_state_stack stack; void *out; TypedData_Get_Struct(self, grpc_rb_channel, &grpc_channel_data_type, wrapper); if (wrapper->wrapped == NULL) { rb_raise(rb_eRuntimeError, "closed!"); return Qnil; } if (!FIXNUM_P(last_state)) { rb_raise( rb_eTypeError, "bad type for last_state. want a GRPC::Core::ChannelState constant"); return Qnil; } stack.wrapper = wrapper; stack.deadline = grpc_rb_time_timeval(deadline, 0); stack.last_state = NUM2LONG(last_state); out = rb_thread_call_without_gvl(watch_channel_state_without_gvl, &stack, watch_channel_state_unblocking_func, wrapper); if (out) { return Qtrue; } return Qfalse; } /* Create a call given a grpc_channel, in order to call method. The request is not sent until grpc_call_invoke is called. */ static VALUE grpc_rb_channel_create_call(VALUE self, VALUE parent, VALUE mask, VALUE method, VALUE host, VALUE deadline) { VALUE res = Qnil; grpc_rb_channel *wrapper = NULL; grpc_call *call = NULL; grpc_call *parent_call = NULL; grpc_channel *ch = NULL; grpc_completion_queue *cq = NULL; int flags = GRPC_PROPAGATE_DEFAULTS; grpc_slice method_slice; grpc_slice host_slice; grpc_slice *host_slice_ptr = NULL; char *tmp_str = NULL; if (host != Qnil) { host_slice = grpc_slice_from_copied_buffer(RSTRING_PTR(host), RSTRING_LEN(host)); host_slice_ptr = &host_slice; } if (mask != Qnil) { flags = NUM2UINT(mask); } if (parent != Qnil) { parent_call = grpc_rb_get_wrapped_call(parent); } cq = grpc_completion_queue_create_for_pluck(NULL); TypedData_Get_Struct(self, grpc_rb_channel, &grpc_channel_data_type, wrapper); ch = wrapper->wrapped; if (ch == NULL) { rb_raise(rb_eRuntimeError, "closed!"); return Qnil; } method_slice = grpc_slice_from_copied_buffer(RSTRING_PTR(method), RSTRING_LEN(method)); call = grpc_channel_create_call(ch, parent_call, flags, cq, method_slice, host_slice_ptr, grpc_rb_time_timeval(deadline, /* absolute time */ 0), NULL); if (call == NULL) { tmp_str = grpc_slice_to_c_string(method_slice); rb_raise(rb_eRuntimeError, "cannot create call with method %s", tmp_str); return Qnil; } grpc_slice_unref(method_slice); if (host_slice_ptr != NULL) { grpc_slice_unref(host_slice); } res = grpc_rb_wrap_call(call, cq); /* Make this channel an instance attribute of the call so that it is not GCed * before the call. */ rb_ivar_set(res, id_channel, self); return res; } /* Closes the channel, calling it's destroy method */ static VALUE grpc_rb_channel_destroy(VALUE self) { grpc_rb_channel *wrapper = NULL; grpc_channel *ch = NULL; TypedData_Get_Struct(self, grpc_rb_channel, &grpc_channel_data_type, wrapper); ch = wrapper->wrapped; if (ch != NULL) { grpc_rb_channel_safe_destroy(wrapper); wrapper->wrapped = NULL; } return Qnil; } /* Called to obtain the target that this channel accesses. */ static VALUE grpc_rb_channel_get_target(VALUE self) { grpc_rb_channel *wrapper = NULL; VALUE res = Qnil; char *target = NULL; TypedData_Get_Struct(self, grpc_rb_channel, &grpc_channel_data_type, wrapper); target = grpc_channel_get_target(wrapper->wrapped); res = rb_str_new2(target); gpr_free(target); return res; } // Either start polling channel connection state or signal that it's free to // destroy. // Not safe to call while a channel's connection state is polled. static void grpc_rb_channel_try_register_connection_polling( grpc_rb_channel *wrapper) { grpc_connectivity_state conn_state; gpr_timespec sleep_time = gpr_time_add( gpr_now(GPR_CLOCK_REALTIME), gpr_time_from_millis(20, GPR_TIMESPAN)); GPR_ASSERT(wrapper); GPR_ASSERT(wrapper->wrapped); gpr_mu_lock(&wrapper->channel_mu); if (wrapper->request_safe_destroy) { wrapper->safe_to_destroy = 1; gpr_cv_broadcast(&wrapper->channel_cv); gpr_mu_unlock(&wrapper->channel_mu); return; } gpr_mu_lock(&global_connection_polling_mu); GPR_ASSERT(channel_polling_thread_started || abort_channel_polling); conn_state = grpc_channel_check_connectivity_state(wrapper->wrapped, 0); if (conn_state != wrapper->current_connectivity_state) { wrapper->current_connectivity_state = conn_state; gpr_cv_broadcast(&wrapper->channel_cv); } // avoid posting work to the channel polling cq if it's been shutdown if (!abort_channel_polling && conn_state != GRPC_CHANNEL_SHUTDOWN) { grpc_channel_watch_connectivity_state( wrapper->wrapped, conn_state, sleep_time, channel_polling_cq, wrapper); } else { wrapper->safe_to_destroy = 1; gpr_cv_broadcast(&wrapper->channel_cv); } gpr_mu_unlock(&global_connection_polling_mu); gpr_mu_unlock(&wrapper->channel_mu); } // Note requires wrapper->wrapped, wrapper->channel_mu/cv initialized static void grpc_rb_channel_safe_destroy(grpc_rb_channel *wrapper) { gpr_mu_lock(&wrapper->channel_mu); wrapper->request_safe_destroy = 1; while (!wrapper->safe_to_destroy) { gpr_cv_wait(&wrapper->channel_cv, &wrapper->channel_mu, gpr_inf_future(GPR_CLOCK_REALTIME)); } GPR_ASSERT(wrapper->safe_to_destroy); gpr_mu_unlock(&wrapper->channel_mu); grpc_channel_destroy(wrapper->wrapped); } // Note this loop breaks out with a single call of // "run_poll_channels_loop_no_gil". // This assumes that a ruby call the unblocking func // indicates process shutdown. // In the worst case, this stops polling channel connectivity // early and falls back to current behavior. static void *run_poll_channels_loop_no_gil(void *arg) { grpc_event event; (void)arg; gpr_log(GPR_DEBUG, "GRPC_RUBY: run_poll_channels_loop_no_gil - begin"); gpr_mu_lock(&global_connection_polling_mu); GPR_ASSERT(!channel_polling_thread_started); channel_polling_thread_started = 1; gpr_cv_broadcast(&global_connection_polling_cv); gpr_mu_unlock(&global_connection_polling_mu); for (;;) { event = grpc_completion_queue_next( channel_polling_cq, gpr_inf_future(GPR_CLOCK_REALTIME), NULL); if (event.type == GRPC_QUEUE_SHUTDOWN) { break; } if (event.type == GRPC_OP_COMPLETE) { grpc_rb_channel_try_register_connection_polling( (grpc_rb_channel *)event.tag); } } grpc_completion_queue_destroy(channel_polling_cq); gpr_log(GPR_DEBUG, "GRPC_RUBY: run_poll_channels_loop_no_gil - exit connection polling " "loop"); return NULL; } // Notify the channel polling loop to cleanup and shutdown. static void run_poll_channels_loop_unblocking_func(void *arg) { (void)arg; gpr_mu_lock(&global_connection_polling_mu); gpr_log(GPR_DEBUG, "GRPC_RUBY: grpc_rb_event_unblocking_func - begin aborting " "connection polling"); abort_channel_polling = 1; grpc_completion_queue_shutdown(channel_polling_cq); gpr_mu_unlock(&global_connection_polling_mu); } // Poll channel connectivity states in background thread without the GIL. static VALUE run_poll_channels_loop(VALUE arg) { (void)arg; gpr_log( GPR_DEBUG, "GRPC_RUBY: run_poll_channels_loop - create connection polling thread"); rb_thread_call_without_gvl(run_poll_channels_loop_no_gil, NULL, run_poll_channels_loop_unblocking_func, NULL); return Qnil; } static void *wait_until_channel_polling_thread_started_no_gil(void *arg) { (void)arg; gpr_log(GPR_DEBUG, "GRPC_RUBY: wait for channel polling thread to start"); gpr_mu_lock(&global_connection_polling_mu); while (!channel_polling_thread_started && !abort_channel_polling) { gpr_cv_wait(&global_connection_polling_cv, &global_connection_polling_mu, gpr_inf_future(GPR_CLOCK_REALTIME)); } gpr_mu_unlock(&global_connection_polling_mu); return NULL; } static void wait_until_channel_polling_thread_started_unblocking_func( void *arg) { (void)arg; gpr_mu_lock(&global_connection_polling_mu); gpr_log(GPR_DEBUG, "GRPC_RUBY: " "wait_until_channel_polling_thread_started_unblocking_func - begin " "aborting connection polling"); abort_channel_polling = 1; gpr_cv_broadcast(&global_connection_polling_cv); gpr_mu_unlock(&global_connection_polling_mu); } /* Temporary fix for * https://github.com/GoogleCloudPlatform/google-cloud-ruby/issues/899. * Transports in idle channels can get destroyed. Normally c-core re-connects, * but in grpc-ruby core never gets a thread until an RPC is made, because ruby * only calls c-core's "completion_queu_pluck" API. * This uses a global background thread that calls * "completion_queue_next" on registered "watch_channel_connectivity_state" * calls - so that c-core can reconnect if needed, when there aren't any RPC's. * TODO(apolcyn) remove this when core handles new RPCs on dead connections. */ void grpc_rb_channel_polling_thread_start() { VALUE background_thread = Qnil; GPR_ASSERT(!abort_channel_polling); GPR_ASSERT(!channel_polling_thread_started); GPR_ASSERT(channel_polling_cq == NULL); gpr_mu_init(&global_connection_polling_mu); gpr_cv_init(&global_connection_polling_cv); channel_polling_cq = grpc_completion_queue_create_for_next(NULL); background_thread = rb_thread_create(run_poll_channels_loop, NULL); if (!RTEST(background_thread)) { gpr_log(GPR_DEBUG, "GRPC_RUBY: failed to spawn channel polling thread"); gpr_mu_lock(&global_connection_polling_mu); abort_channel_polling = 1; gpr_cv_broadcast(&global_connection_polling_cv); gpr_mu_unlock(&global_connection_polling_mu); } } static void Init_grpc_propagate_masks() { /* Constants representing call propagation masks in grpc.h */ VALUE grpc_rb_mPropagateMasks = rb_define_module_under(grpc_rb_mGrpcCore, "PropagateMasks"); rb_define_const(grpc_rb_mPropagateMasks, "DEADLINE", UINT2NUM(GRPC_PROPAGATE_DEADLINE)); rb_define_const(grpc_rb_mPropagateMasks, "CENSUS_STATS_CONTEXT", UINT2NUM(GRPC_PROPAGATE_CENSUS_STATS_CONTEXT)); rb_define_const(grpc_rb_mPropagateMasks, "CENSUS_TRACING_CONTEXT", UINT2NUM(GRPC_PROPAGATE_CENSUS_TRACING_CONTEXT)); rb_define_const(grpc_rb_mPropagateMasks, "CANCELLATION", UINT2NUM(GRPC_PROPAGATE_CANCELLATION)); rb_define_const(grpc_rb_mPropagateMasks, "DEFAULTS", UINT2NUM(GRPC_PROPAGATE_DEFAULTS)); } static void Init_grpc_connectivity_states() { /* Constants representing call propagation masks in grpc.h */ VALUE grpc_rb_mConnectivityStates = rb_define_module_under(grpc_rb_mGrpcCore, "ConnectivityStates"); rb_define_const(grpc_rb_mConnectivityStates, "IDLE", LONG2NUM(GRPC_CHANNEL_IDLE)); rb_define_const(grpc_rb_mConnectivityStates, "CONNECTING", LONG2NUM(GRPC_CHANNEL_CONNECTING)); rb_define_const(grpc_rb_mConnectivityStates, "READY", LONG2NUM(GRPC_CHANNEL_READY)); rb_define_const(grpc_rb_mConnectivityStates, "TRANSIENT_FAILURE", LONG2NUM(GRPC_CHANNEL_TRANSIENT_FAILURE)); rb_define_const(grpc_rb_mConnectivityStates, "FATAL_FAILURE", LONG2NUM(GRPC_CHANNEL_SHUTDOWN)); } void Init_grpc_channel() { grpc_rb_cChannelArgs = rb_define_class("TmpChannelArgs", rb_cObject); grpc_rb_cChannel = rb_define_class_under(grpc_rb_mGrpcCore, "Channel", rb_cObject); /* Allocates an object managed by the ruby runtime */ rb_define_alloc_func(grpc_rb_cChannel, grpc_rb_channel_alloc); /* Provides a ruby constructor and support for dup/clone. */ rb_define_method(grpc_rb_cChannel, "initialize", grpc_rb_channel_init, -1); rb_define_method(grpc_rb_cChannel, "initialize_copy", grpc_rb_cannot_init_copy, 1); /* Add ruby analogues of the Channel methods. */ rb_define_method(grpc_rb_cChannel, "connectivity_state", grpc_rb_channel_get_connectivity_state, -1); rb_define_method(grpc_rb_cChannel, "watch_connectivity_state", grpc_rb_channel_watch_connectivity_state, 2); rb_define_method(grpc_rb_cChannel, "create_call", grpc_rb_channel_create_call, 5); rb_define_method(grpc_rb_cChannel, "target", grpc_rb_channel_get_target, 0); rb_define_method(grpc_rb_cChannel, "destroy", grpc_rb_channel_destroy, 0); rb_define_alias(grpc_rb_cChannel, "close", "destroy"); id_channel = rb_intern("__channel"); id_target = rb_intern("__target"); rb_define_const(grpc_rb_cChannel, "SSL_TARGET", ID2SYM(rb_intern(GRPC_SSL_TARGET_NAME_OVERRIDE_ARG))); rb_define_const(grpc_rb_cChannel, "ENABLE_CENSUS", ID2SYM(rb_intern(GRPC_ARG_ENABLE_CENSUS))); rb_define_const(grpc_rb_cChannel, "MAX_CONCURRENT_STREAMS", ID2SYM(rb_intern(GRPC_ARG_MAX_CONCURRENT_STREAMS))); rb_define_const(grpc_rb_cChannel, "MAX_MESSAGE_LENGTH", ID2SYM(rb_intern(GRPC_ARG_MAX_RECEIVE_MESSAGE_LENGTH))); id_insecure_channel = rb_intern("this_channel_is_insecure"); Init_grpc_propagate_masks(); Init_grpc_connectivity_states(); } /* Gets the wrapped channel from the ruby wrapper */ grpc_channel *grpc_rb_get_wrapped_channel(VALUE v) { grpc_rb_channel *wrapper = NULL; TypedData_Get_Struct(v, grpc_rb_channel, &grpc_channel_data_type, wrapper); return wrapper->wrapped; }