Class: Up::Ruby::Server

Inherits:

Object

• Object
• Up::Ruby::Server

Defined in:

ext/up_ext/up_ext.c

Direct Known Subclasses

Cluster

Instance Method Summary

• #initialize(*args) ⇒ Object constructor
• #listen ⇒ Object
• #publish(channel, message) ⇒ Object
• #stop ⇒ Object

Constructor Details

#initialize(*args) ⇒ Object

# File 'ext/up_ext/up_ext.c', line 580

static VALUE up_server_init(int argc, VALUE *argv, VALUE self) {
  if (!rb_keyword_given_p())
    rb_raise(rb_eArgError, "no args given, must at least provide app:");
  ID kwargs[] = {id_app, id_host, id_port, id_logger};
  VALUE rargs[4] = {Qnil, Qnil, Qnil, Qnil};
  VALUE options = Qnil;

  rb_scan_args_kw(1, argc, argv, ":", &options);
  rb_get_kwargs(options, kwargs, 1, 2, rargs);

  VALUE rapp = rargs[0];
  VALUE rhost = rargs[1];
  VALUE rport = rargs[2];

  up_internal_check_arg_types(rapp, &rhost, &rport);

  server_s *s = DATA_PTR(self);
  s->rapp = rapp;
  s->host = rb_obj_freeze(rhost);
  s->port = rport;
  s->logger = rargs[3];

  return self;
}

Instance Method Details

#listen ⇒ Object

# File 'ext/up_ext/up_ext.c', line 752

static VALUE up_server_listen(VALUE self) {
  server_s *s = DATA_PTR(self);
  up_internal_check_arg_types(s->rapp, &s->host, &s->port);
  rb_ivar_set(mUp, at_instance, self);

  s->env_template = rb_hash_dup(rack_env_template);
  // When combined with SCRIPT_NAME and PATH_INFO, these variables can be used
  // to complete the URL.
  rb_hash_aset(s->env_template, SERVER_NAME, s->host);
  // An optional Integer which is the port the server is running on.
  rb_hash_aset(s->env_template, SERVER_PORT, s->port);
  if (s->logger && s->logger != Qundef && s->logger != Qnil) {
    rb_hash_aset(s->env_template, rack_logger, s->logger);
  }
  struct us_socket_context_options_t options = {.key_file_name = NULL,
                                                .cert_file_name = NULL,
                                                .ca_file_name = NULL,
                                                .passphrase = NULL,
                                                .dh_params_file_name = NULL,
                                                .ssl_ciphers = NULL};
  s->app = uws_create_app(USE_SSL, options);
  if (!s->app)
    rb_raise(rb_eRuntimeError, "could not init uws app");
  uws_app_listen_config_t config = {
      .port = FIX2INT(s->port), .host = RSTRING_PTR(s->host), .options = 0};
  VALUE rmember_id = rb_ivar_get(self, at_member_id);
  if (rmember_id != Qnil) {
    // got a cluster
    s->member_id = FIX2INT(rmember_id);
    // install signal handler
    cluster_app = s->app;
    cluster_socket = NULL;
    struct sigaction upclcl = {.sa_handler = up_internal_close_sockets,
                               .sa_flags = 0};
    sigemptyset(&upclcl.sa_mask);
    sigaction(SIGINT, &upclcl, NULL);
    // open publish ports
    VALUE rworkers = rb_ivar_get(self, at_workers);
    s->workers = FIX2INT(rworkers);
    VALUE rsecret = rb_ivar_get(self, at_secret);
    if (TYPE(rsecret) != T_STRING || RSTRING_LEN(rsecret) != 36)
      rb_raise(rb_eTypeError, "cluster secret of unknown type");
    memcpy(s->secret, RSTRING_PTR(rsecret), 36);
    s->secret[36] = '\0';
    uws_app_any(USE_SSL, s->app, INTERNAL_PUBLISH_PATH,
                up_internal_publish_handler, (void *)s);
    uws_app_listen_config_t config_internal = {
        .port = config.port + s->member_id, .host = "localhost", .options = 0};
    uws_app_listen_with_config(false, s->app, config_internal,
                               up_server_cluster_listen_handler, NULL);
  } else {
    cluster_app = s->app;
    cluster_socket = NULL;
    struct sigaction upclcl = {.sa_handler = up_internal_close_sockets,
                               .sa_flags = 0};
    sigemptyset(&upclcl.sa_mask);
    sigaction(SIGINT, &upclcl, NULL);
  }
  uws_app_any(USE_SSL, s->app, "/*", up_server_request_handler, (void *)s);
  uws_ws(USE_SSL, s->app, "/*",
         (uws_socket_behavior_t){.compression = DISABLED,
                                 .maxPayloadLength = 5 * 1024 * 1024,
                                 .idleTimeout = 120,
                                 .upgrade = up_ws_upgrade_handler,
                                 .open = up_ws_open_handler,
                                 .message = up_ws_message_handler,
                                 .close = up_ws_close_handler,
                                 .drain = up_ws_drain_handler,
                                 .ping = up_ws_ping_handler,
                                 .pong = up_ws_pong_handler},
         s);
  uws_app_listen_with_config(USE_SSL, s->app, config, up_server_listen_handler,
                             NULL);
  uws_app_run(USE_SSL, s->app);
  return self;
}

#publish(channel, message) ⇒ Object

# File 'ext/up_ext/up_ext.c', line 829

static VALUE up_server_publish(VALUE self, VALUE channel, VALUE message) {
  if (TYPE(channel) != T_STRING)
    channel = rb_obj_as_string(channel);
  if (TYPE(message) != T_STRING)
    message = rb_obj_as_string(message);
  server_s *s = DATA_PTR(self);
  VALUE members = rb_ivar_get(self, at_members);
  if (members != Qnil) {
    long i, mb_cnt = RARRAY_LEN(members);
    for (i = 0; i < mb_cnt; i++) {
      up_internal_publish_to_member(s, channel, message, i);
    }
  } else {
    uws_publish(USE_SSL, s->app, RSTRING_PTR(channel), RSTRING_LEN(channel),
                RSTRING_PTR(message), RSTRING_LEN(message), TEXT, false);
    if (s->member_id > 0) {
      // publish to cluster members
      int i;
      for (i = 1; i <= s->workers; i++) {
        if (i != s->member_id)
          up_internal_publish_to_member(s, channel, message, i);
      }
    }
  }
  return Qtrue;
}

#stop ⇒ Object

# File 'ext/up_ext/up_ext.c', line 856

static VALUE up_server_stop(VALUE self) {
  server_s *s = DATA_PTR(self);
  if (!s->app)
    rb_raise(rb_eRuntimeError, "no uws, did initialize call super?");
  uws_app_close(USE_SSL, s->app);
  uws_app_destroy(USE_SSL, s->app);
  s->app = NULL;
  return Qnil;
}