summaryrefslogtreecommitdiffhomepage
path: root/src/ruby
diff options
context:
space:
mode:
Diffstat (limited to 'src/ruby')
-rw-r--r--src/ruby/nxt_ruby.c911
-rw-r--r--src/ruby/nxt_ruby.h34
-rw-r--r--src/ruby/nxt_ruby_stream_io.c290
3 files changed, 1235 insertions, 0 deletions
diff --git a/src/ruby/nxt_ruby.c b/src/ruby/nxt_ruby.c
new file mode 100644
index 00000000..abb9640e
--- /dev/null
+++ b/src/ruby/nxt_ruby.c
@@ -0,0 +1,911 @@
+/*
+ * Copyright (C) Alexander Borisov
+ * Copyright (C) NGINX, Inc.
+ */
+
+#include <ruby/nxt_ruby.h>
+
+
+#define NXT_RUBY_RACK_API_VERSION_MAJOR 1
+#define NXT_RUBY_RACK_API_VERSION_MINOR 3
+
+#define NXT_RUBY_STRINGIZE_HELPER(x) #x
+#define NXT_RUBY_STRINGIZE(x) NXT_RUBY_STRINGIZE_HELPER(x)
+
+#define NXT_RUBY_LIB_VERSION \
+ NXT_RUBY_STRINGIZE(RUBY_API_VERSION_MAJOR) \
+ "." NXT_RUBY_STRINGIZE(RUBY_API_VERSION_MINOR) \
+ "." NXT_RUBY_STRINGIZE(RUBY_API_VERSION_TEENY)
+
+
+typedef struct {
+ nxt_task_t *task;
+ nxt_str_t *script;
+ VALUE builder;
+} nxt_ruby_rack_init_t;
+
+
+static nxt_int_t nxt_ruby_init(nxt_task_t *task, nxt_common_app_conf_t *conf);
+static VALUE nxt_ruby_init_basic(VALUE arg);
+static nxt_int_t nxt_ruby_init_io(nxt_task_t *task);
+static VALUE nxt_ruby_rack_init(nxt_ruby_rack_init_t *rack_init);
+
+static VALUE nxt_ruby_require_rubygems(VALUE arg);
+static VALUE nxt_ruby_require_rack(VALUE arg);
+static VALUE nxt_ruby_rack_parse_script(VALUE ctx);
+static VALUE nxt_ruby_rack_env_create(VALUE arg);
+static nxt_int_t nxt_ruby_run(nxt_task_t *task, nxt_app_rmsg_t *rmsg,
+ nxt_app_wmsg_t *wmsg);
+
+static VALUE nxt_ruby_rack_app_run(VALUE arg);
+static nxt_int_t nxt_ruby_read_request(nxt_ruby_run_ctx_t *run_ctx,
+ VALUE hash_env);
+nxt_inline nxt_int_t nxt_ruby_read_add_env(nxt_task_t *task,
+ nxt_app_rmsg_t *rmsg, VALUE hash_env, const char *name, nxt_str_t *str);
+static nxt_int_t nxt_ruby_rack_result_status(VALUE result);
+nxt_inline nxt_int_t nxt_ruby_write(nxt_task_t *task, nxt_app_wmsg_t *wmsg,
+ const u_char *data, size_t len, nxt_bool_t flush, nxt_bool_t last);
+static nxt_int_t nxt_ruby_rack_result_headers(VALUE result);
+static int nxt_ruby_hash_foreach(VALUE r_key, VALUE r_value, VALUE arg);
+static nxt_int_t nxt_ruby_head_send_part(const char *key, size_t key_size,
+ const char *value, size_t value_size);
+static nxt_int_t nxt_ruby_rack_result_body(VALUE result);
+static nxt_int_t nxt_ruby_rack_result_body_file_write(VALUE filepath);
+static VALUE nxt_ruby_rack_result_body_each(VALUE body);
+
+static void nxt_ruby_exception_log(nxt_task_t *task, uint32_t level,
+ const char *desc);
+
+static void nxt_ruby_atexit(nxt_task_t *task);
+
+
+static uint32_t compat[] = {
+ NXT_VERNUM, NXT_DEBUG,
+};
+
+static VALUE nxt_ruby_rackup;
+static VALUE nxt_ruby_call;
+static VALUE nxt_ruby_env;
+static VALUE nxt_ruby_io_input;
+static VALUE nxt_ruby_io_error;
+static nxt_ruby_run_ctx_t nxt_ruby_run_ctx;
+
+NXT_EXPORT nxt_application_module_t nxt_app_module = {
+ sizeof(compat),
+ compat,
+ nxt_string("ruby"),
+ nxt_string(NXT_RUBY_LIB_VERSION),
+ nxt_ruby_init,
+ nxt_ruby_run,
+ nxt_ruby_atexit,
+};
+
+
+static nxt_int_t
+nxt_ruby_init(nxt_task_t *task, nxt_common_app_conf_t *conf)
+{
+ int state;
+ VALUE dummy, res;
+ nxt_ruby_rack_init_t rack_init;
+
+ ruby_init();
+ Init_stack(&dummy);
+ ruby_init_loadpath();
+ ruby_script("NGINX_Unit");
+
+ rack_init.task = task;
+ rack_init.script = &conf->u.ruby.script;
+
+ res = rb_protect(nxt_ruby_init_basic,
+ (VALUE) (uintptr_t) &rack_init, &state);
+ if (nxt_slow_path(res == Qnil || state != 0)) {
+ nxt_ruby_exception_log(task, NXT_LOG_ALERT,
+ "Failed to init basic variables");
+ return NXT_ERROR;
+ }
+
+ nxt_ruby_rackup = nxt_ruby_rack_init(&rack_init);
+ if (nxt_slow_path(nxt_ruby_rackup == Qnil)) {
+ return NXT_ERROR;
+ }
+
+ nxt_ruby_call = rb_intern("call");
+ if (nxt_slow_path(nxt_ruby_call == Qnil)) {
+ nxt_alert(task, "Ruby: Unable to find rack entry point");
+
+ return NXT_ERROR;
+ }
+
+ nxt_ruby_env = rb_protect(nxt_ruby_rack_env_create, Qnil, &state);
+ if (nxt_slow_path(state != 0)) {
+ nxt_ruby_exception_log(task, NXT_LOG_ALERT,
+ "Failed to create 'environ' variable");
+ return NXT_ERROR;
+ }
+
+ rb_gc_register_address(&nxt_ruby_rackup);
+ rb_gc_register_address(&nxt_ruby_call);
+ rb_gc_register_address(&nxt_ruby_env);
+
+ return NXT_OK;
+}
+
+
+static VALUE
+nxt_ruby_init_basic(VALUE arg)
+{
+ int state;
+ nxt_int_t rc;
+ nxt_ruby_rack_init_t *rack_init;
+
+ rack_init = (nxt_ruby_rack_init_t *) (uintptr_t) arg;
+
+ state = rb_enc_find_index("encdb");
+ if (nxt_slow_path(state == 0)) {
+ nxt_alert(rack_init->task,
+ "Ruby: Failed to find encoding index 'encdb'");
+
+ return Qnil;
+ }
+
+ rc = nxt_ruby_init_io(rack_init->task);
+ if (nxt_slow_path(rc != NXT_OK)) {
+ return Qnil;
+ }
+
+ return arg;
+}
+
+
+static nxt_int_t
+nxt_ruby_init_io(nxt_task_t *task)
+{
+ VALUE rb, io_input, io_error;
+
+ io_input = nxt_ruby_stream_io_input_init();
+ rb = Data_Wrap_Struct(io_input, 0, 0, &nxt_ruby_run_ctx);
+
+ nxt_ruby_io_input = rb_funcall(io_input, rb_intern("new"), 1, rb);
+ if (nxt_slow_path(nxt_ruby_io_input == Qnil)) {
+ nxt_alert(task, "Ruby: Failed to create environment 'rack.input' var");
+
+ return NXT_ERROR;
+ }
+
+ io_error = nxt_ruby_stream_io_error_init();
+ rb = Data_Wrap_Struct(io_error, 0, 0, &nxt_ruby_run_ctx);
+
+ nxt_ruby_io_error = rb_funcall(io_error, rb_intern("new"), 1, rb);
+ if (nxt_slow_path(nxt_ruby_io_error == Qnil)) {
+ nxt_alert(task, "Ruby: Failed to create environment 'rack.error' var");
+
+ return NXT_ERROR;
+ }
+
+ rb_gc_register_address(&nxt_ruby_io_input);
+ rb_gc_register_address(&nxt_ruby_io_error);
+
+ return NXT_OK;
+}
+
+
+static VALUE
+nxt_ruby_rack_init(nxt_ruby_rack_init_t *rack_init)
+{
+ int state;
+ VALUE rack, rackup;
+
+ rb_protect(nxt_ruby_require_rubygems, Qnil, &state);
+ if (nxt_slow_path(state != 0)) {
+ nxt_ruby_exception_log(rack_init->task, NXT_LOG_ALERT,
+ "Failed to require 'rubygems' package");
+ return Qnil;
+ }
+
+ rb_protect(nxt_ruby_require_rack, Qnil, &state);
+ if (nxt_slow_path(state != 0)) {
+ nxt_ruby_exception_log(rack_init->task, NXT_LOG_ALERT,
+ "Failed to require 'rack' package");
+ return Qnil;
+ }
+
+ rack = rb_const_get(rb_cObject, rb_intern("Rack"));
+ rack_init->builder = rb_const_get(rack, rb_intern("Builder"));
+
+ rackup = rb_protect(nxt_ruby_rack_parse_script,
+ (VALUE) (uintptr_t) rack_init, &state);
+ if (nxt_slow_path(TYPE(rackup) != T_ARRAY || state != 0)) {
+ nxt_ruby_exception_log(rack_init->task, NXT_LOG_ALERT,
+ "Failed to parse rack script");
+ return Qnil;
+ }
+
+ if (nxt_slow_path(RARRAY_LEN(rackup) < 1)) {
+ nxt_alert(rack_init->task, "Ruby: Invalid rack config file");
+ return Qnil;
+ }
+
+ return RARRAY_PTR(rackup)[0];
+}
+
+
+static VALUE
+nxt_ruby_require_rubygems(VALUE arg)
+{
+ return rb_funcall(rb_cObject, rb_intern("require"), 1,
+ rb_str_new2("rubygems"));
+}
+
+
+static VALUE
+nxt_ruby_require_rack(VALUE arg)
+{
+ return rb_funcall(rb_cObject, rb_intern("require"), 1, rb_str_new2("rack"));
+}
+
+
+static VALUE
+nxt_ruby_rack_parse_script(VALUE ctx)
+{
+ VALUE script, res;
+ nxt_ruby_rack_init_t *rack_init;
+
+ rack_init = (nxt_ruby_rack_init_t *) (uintptr_t) ctx;
+
+ script = rb_str_new((const char *) rack_init->script->start,
+ (long) rack_init->script->length);
+
+ res = rb_funcall(rack_init->builder, rb_intern("parse_file"), 1, script);
+
+ rb_str_free(script);
+
+ return res;
+}
+
+
+static VALUE
+nxt_ruby_rack_env_create(VALUE arg)
+{
+ VALUE hash_env, version;
+
+ hash_env = rb_hash_new();
+ version = rb_ary_new();
+
+ rb_ary_push(version, UINT2NUM(NXT_RUBY_RACK_API_VERSION_MAJOR));
+ rb_ary_push(version, UINT2NUM(NXT_RUBY_RACK_API_VERSION_MINOR));
+
+ rb_hash_aset(hash_env, rb_str_new2("rack.version"), version);
+ rb_hash_aset(hash_env, rb_str_new2("rack.url_scheme"), rb_str_new2("http"));
+ rb_hash_aset(hash_env, rb_str_new2("rack.input"), nxt_ruby_io_input);
+ rb_hash_aset(hash_env, rb_str_new2("rack.errors"), nxt_ruby_io_error);
+ rb_hash_aset(hash_env, rb_str_new2("rack.multithread"), Qfalse);
+ rb_hash_aset(hash_env, rb_str_new2("rack.multiprocess"), Qtrue);
+ rb_hash_aset(hash_env, rb_str_new2("rack.run_once"), Qfalse);
+ rb_hash_aset(hash_env, rb_str_new2("rack.hijack?"), Qfalse);
+ rb_hash_aset(hash_env, rb_str_new2("rack.hijack"), Qnil);
+ rb_hash_aset(hash_env, rb_str_new2("rack.hijack_io"), Qnil);
+
+ return hash_env;
+}
+
+
+static nxt_int_t
+nxt_ruby_run(nxt_task_t *task, nxt_app_rmsg_t *rmsg, nxt_app_wmsg_t *wmsg)
+{
+ int state;
+ VALUE res;
+
+ nxt_ruby_run_ctx.task = task;
+ nxt_ruby_run_ctx.rmsg = rmsg;
+ nxt_ruby_run_ctx.wmsg = wmsg;
+
+ res = rb_protect(nxt_ruby_rack_app_run, Qnil, &state);
+ if (nxt_slow_path(state != 0)) {
+ nxt_ruby_exception_log(nxt_ruby_run_ctx.task, NXT_LOG_ERR,
+ "Failed to run ruby script");
+ return NXT_ERROR;
+ }
+
+ if (nxt_slow_path(res == Qnil)) {
+ return NXT_ERROR;
+ }
+
+ return NXT_OK;
+}
+
+
+static VALUE
+nxt_ruby_rack_app_run(VALUE arg)
+{
+ VALUE env, result;
+ nxt_int_t rc;
+
+ env = rb_hash_dup(nxt_ruby_env);
+
+ rc = nxt_ruby_read_request(&nxt_ruby_run_ctx, env);
+ if (nxt_slow_path(rc != NXT_OK)) {
+ nxt_alert(nxt_ruby_run_ctx.task,
+ "Ruby: Failed to process incoming request");
+
+ goto fail;
+ }
+
+ result = rb_funcall(nxt_ruby_rackup, nxt_ruby_call, 1, env);
+ if (nxt_slow_path(TYPE(result) != T_ARRAY)) {
+ nxt_log(nxt_ruby_run_ctx.task, NXT_LOG_ERR,
+ "Ruby: Invalid response format from application");
+
+ goto fail;
+ }
+
+ if (nxt_slow_path(RARRAY_LEN(result) != 3)) {
+ nxt_log(nxt_ruby_run_ctx.task, NXT_LOG_ERR,
+ "Ruby: Invalid response format from application. "
+ "Need 3 entries [Status, Headers, Body]");
+
+ goto fail;
+ }
+
+ rc = nxt_ruby_rack_result_status(result);
+ if (nxt_slow_path(rc != NXT_OK)) {
+ goto fail;
+ }
+
+ rc = nxt_ruby_rack_result_headers(result);
+ if (nxt_slow_path(rc != NXT_OK)) {
+ goto fail;
+ }
+
+ rc = nxt_ruby_rack_result_body(result);
+ if (nxt_slow_path(rc != NXT_OK)) {
+ goto fail;
+ }
+
+ rc = nxt_app_msg_flush(nxt_ruby_run_ctx.task, nxt_ruby_run_ctx.wmsg, 1);
+ if (nxt_slow_path(rc != NXT_OK)) {
+ goto fail;
+ }
+
+ rb_hash_delete(env, rb_obj_id(env));
+
+ return result;
+
+fail:
+
+ rb_hash_delete(env, rb_obj_id(env));
+
+ return Qnil;
+}
+
+
+static nxt_int_t
+nxt_ruby_read_request(nxt_ruby_run_ctx_t *run_ctx, VALUE hash_env)
+{
+ u_char *colon;
+ size_t query_size;
+ nxt_int_t rc;
+ nxt_str_t str, value, path, target;
+ nxt_str_t host, server_name, server_port;
+ nxt_task_t *task;
+ nxt_app_rmsg_t *rmsg;
+
+ static nxt_str_t def_host = nxt_string("localhost");
+ static nxt_str_t def_port = nxt_string("80");
+
+ task = run_ctx->task;
+ rmsg = run_ctx->rmsg;
+
+ rc = nxt_ruby_read_add_env(task, rmsg, hash_env, "REQUEST_METHOD", &str);
+ if (nxt_slow_path(rc != NXT_OK)) {
+ return NXT_ERROR;
+ }
+
+ rc = nxt_ruby_read_add_env(task, rmsg, hash_env, "REQUEST_URI", &target);
+ if (nxt_slow_path(rc != NXT_OK)) {
+ return NXT_ERROR;
+ }
+
+ rc = nxt_app_msg_read_str(task, rmsg, &path);
+ if (nxt_slow_path(rc != NXT_OK)) {
+ return NXT_ERROR;
+ }
+
+ rc = nxt_app_msg_read_size(task, rmsg, &query_size);
+ if (nxt_slow_path(rc != NXT_OK)) {
+ return NXT_ERROR;
+ }
+
+ if (path.start == NULL || path.length == 0) {
+ path = target;
+ }
+
+ rb_hash_aset(hash_env, rb_str_new2("PATH_INFO"),
+ rb_str_new((const char *) path.start, (long) path.length));
+
+ if (query_size > 0) {
+ query_size--;
+
+ if (nxt_slow_path(target.length < query_size)) {
+ return NXT_ERROR;
+ }
+
+ str.start = &target.start[query_size];
+ str.length = target.length - query_size;
+
+ rb_hash_aset(hash_env, rb_str_new2("QUERY_STRING"),
+ rb_str_new((const char *) str.start, (long) str.length));
+ }
+
+ rc = nxt_ruby_read_add_env(task, rmsg, hash_env, "SERVER_PROTOCOL", &str);
+ if (nxt_slow_path(rc != NXT_OK)) {
+ return NXT_ERROR;
+ }
+
+ rc = nxt_ruby_read_add_env(task, rmsg, hash_env, "REMOTE_ADDR", &str);
+ if (nxt_slow_path(rc != NXT_OK)) {
+ return NXT_ERROR;
+ }
+
+ rc = nxt_ruby_read_add_env(task, rmsg, hash_env, "SERVER_ADDR", &str);
+ if (nxt_slow_path(rc != NXT_OK)) {
+ return NXT_ERROR;
+ }
+
+ rc = nxt_app_msg_read_str(task, rmsg, &host);
+ if (nxt_slow_path(rc != NXT_OK)) {
+ return NXT_ERROR;
+ }
+
+ if (host.length == 0) {
+ host = def_host;
+ }
+
+ colon = nxt_memchr(host.start, ':', host.length);
+ server_name = host;
+
+ if (colon != NULL) {
+ server_name.length = colon - host.start;
+
+ server_port.start = colon + 1;
+ server_port.length = host.length - server_name.length - 1;
+
+ } else {
+ server_port = def_port;
+ }
+
+ rb_hash_aset(hash_env, rb_str_new2("SERVER_NAME"),
+ rb_str_new((const char *) server_name.start,
+ (long) server_name.length));
+
+ rb_hash_aset(hash_env, rb_str_new2("SERVER_PORT"),
+ rb_str_new((const char *) server_port.start,
+ (long) server_port.length));
+
+ rc = nxt_ruby_read_add_env(task, rmsg, hash_env, "CONTENT_TYPE", &str);
+ if (nxt_slow_path(rc != NXT_OK)) {
+ return NXT_ERROR;
+ }
+
+ rc = nxt_ruby_read_add_env(task, rmsg, hash_env, "CONTENT_LENGTH", &str);
+ if (nxt_slow_path(rc != NXT_OK)) {
+ return NXT_ERROR;
+ }
+
+ for ( ;; ) {
+ rc = nxt_app_msg_read_str(task, rmsg, &str);
+ if (nxt_slow_path(rc != NXT_OK)) {
+ return NXT_ERROR;
+ }
+
+ if (nxt_slow_path(str.length == 0)) {
+ break;
+ }
+
+ rc = nxt_app_msg_read_str(task, rmsg, &value);
+ if (nxt_slow_path(rc != NXT_OK)) {
+ return NXT_ERROR;
+ }
+
+ rb_hash_aset(hash_env,
+ rb_str_new((char *) str.start, (long) str.length),
+ rb_str_new((const char *) value.start,
+ (long) value.length));
+ }
+
+ rc = nxt_app_msg_read_size(task, rmsg, &run_ctx->body_preread_size);
+ if (nxt_slow_path(rc != NXT_OK)) {
+ return NXT_ERROR;
+ }
+
+ return NXT_OK;
+}
+
+
+nxt_inline nxt_int_t
+nxt_ruby_read_add_env(nxt_task_t *task, nxt_app_rmsg_t *rmsg, VALUE hash_env,
+ const char *name, nxt_str_t *str)
+{
+ nxt_int_t rc;
+
+ rc = nxt_app_msg_read_str(task, rmsg, str);
+ if (nxt_slow_path(rc != NXT_OK)) {
+ return rc;
+ }
+
+ if (str->start == NULL) {
+ rb_hash_aset(hash_env, rb_str_new2(name), Qnil);
+ return NXT_OK;
+ }
+
+ rb_hash_aset(hash_env, rb_str_new2(name),
+ rb_str_new((const char *) str->start, (long) str->length));
+
+ return NXT_OK;
+}
+
+
+static nxt_int_t
+nxt_ruby_rack_result_status(VALUE result)
+{
+ VALUE status;
+ u_char *p;
+ size_t len;
+ nxt_int_t rc;
+ u_char buf[3];
+
+ status = rb_ary_entry(result, 0);
+
+ if (TYPE(status) == T_FIXNUM) {
+ nxt_sprintf(buf, buf + 3, "%03d", FIX2INT(status));
+
+ p = buf;
+ len = 3;
+
+ } else if (TYPE(status) == T_STRING) {
+ p = (u_char *) RSTRING_PTR(status);
+ len = RSTRING_LEN(status);
+
+ } else {
+ nxt_log(nxt_ruby_run_ctx.task, NXT_LOG_ERR,
+ "Ruby: Invalid response 'status' format from application");
+
+ return NXT_ERROR;
+ }
+
+ rc = nxt_ruby_write(nxt_ruby_run_ctx.task, nxt_ruby_run_ctx.wmsg,
+ (u_char *) "Status: ", (sizeof("Status: ") - 1), 0, 0);
+ if (nxt_slow_path(rc != NXT_OK)) {
+ return NXT_ERROR;
+ }
+
+ rc = nxt_ruby_write(nxt_ruby_run_ctx.task, nxt_ruby_run_ctx.wmsg,
+ p, len, 0, 0);
+ if (nxt_slow_path(rc != NXT_OK)) {
+ return NXT_ERROR;
+ }
+
+ rc = nxt_ruby_write(nxt_ruby_run_ctx.task, nxt_ruby_run_ctx.wmsg,
+ (u_char *) "\r\n", (sizeof("\r\n") - 1), 0, 0);
+ if (nxt_slow_path(rc != NXT_OK)) {
+ return NXT_ERROR;
+ }
+
+ return NXT_OK;
+}
+
+
+nxt_inline nxt_int_t
+nxt_ruby_write(nxt_task_t *task, nxt_app_wmsg_t *wmsg,
+ const u_char *data, size_t len, nxt_bool_t flush, nxt_bool_t last)
+{
+ nxt_int_t rc;
+
+ rc = nxt_app_msg_write_raw(task, wmsg, data, len);
+ if (nxt_slow_path(rc != NXT_OK)) {
+ return rc;
+ }
+
+ if (flush || last) {
+ rc = nxt_app_msg_flush(task, wmsg, last);
+ }
+
+ return rc;
+}
+
+
+static nxt_int_t
+nxt_ruby_rack_result_headers(VALUE result)
+{
+ VALUE headers;
+ nxt_int_t rc;
+
+ headers = rb_ary_entry(result, 1);
+ if (nxt_slow_path(TYPE(headers) != T_HASH)) {
+ nxt_log(nxt_ruby_run_ctx.task, NXT_LOG_ERR,
+ "Ruby: Invalid response 'headers' format from application");
+
+ return NXT_ERROR;
+ }
+
+ rc = NXT_OK;
+
+ rb_hash_foreach(headers, nxt_ruby_hash_foreach, (VALUE) (uintptr_t) &rc);
+ if (nxt_slow_path(rc != NXT_OK)) {
+ return NXT_ERROR;
+ }
+
+ rc = nxt_ruby_write(nxt_ruby_run_ctx.task, nxt_ruby_run_ctx.wmsg,
+ (u_char *) "\r\n", (sizeof("\r\n") - 1), 0, 0);
+ if (nxt_slow_path(rc != NXT_OK)) {
+ return NXT_ERROR;
+ }
+
+ return NXT_OK;
+}
+
+
+static int
+nxt_ruby_hash_foreach(VALUE r_key, VALUE r_value, VALUE arg)
+{
+ nxt_int_t rc, *rc_p;
+ const char *value, *value_end, *pos;
+
+ rc_p = (nxt_int_t *) (uintptr_t) arg;
+
+ if (nxt_slow_path(TYPE(r_key) != T_STRING)) {
+ nxt_log(nxt_ruby_run_ctx.task, NXT_LOG_ERR,
+ "Ruby: Wrong header entry 'key' from application");
+
+ goto fail;
+ }
+
+ if (nxt_slow_path(TYPE(r_value) != T_STRING)) {
+ nxt_log(nxt_ruby_run_ctx.task, NXT_LOG_ERR,
+ "Ruby: Wrong header entry 'value' from application");
+
+ goto fail;
+ }
+
+ value = RSTRING_PTR(r_value);
+ value_end = value + RSTRING_LEN(r_value);
+
+ pos = value;
+
+ for ( ;; ) {
+ pos = strchr(pos, '\n');
+
+ if (pos == NULL) {
+ break;
+ }
+
+ rc = nxt_ruby_head_send_part(RSTRING_PTR(r_key), RSTRING_LEN(r_key),
+ value, pos - value);
+ if (nxt_slow_path(rc != NXT_OK)) {
+ goto fail;
+ }
+
+ pos++;
+ value = pos;
+ }
+
+ if (value <= value_end) {
+ rc = nxt_ruby_head_send_part(RSTRING_PTR(r_key), RSTRING_LEN(r_key),
+ value, value_end - value);
+ if (nxt_slow_path(rc != NXT_OK)) {
+ goto fail;
+ }
+ }
+
+ *rc_p = NXT_OK;
+
+ return ST_CONTINUE;
+
+fail:
+
+ *rc_p = NXT_ERROR;
+
+ return ST_STOP;
+}
+
+
+static nxt_int_t
+nxt_ruby_head_send_part(const char *key, size_t key_size,
+ const char *value, size_t value_size)
+{
+ nxt_int_t rc;
+
+ rc = nxt_app_msg_write_raw(nxt_ruby_run_ctx.task, nxt_ruby_run_ctx.wmsg,
+ (u_char *) key, key_size);
+ if (nxt_slow_path(rc != NXT_OK)) {
+ return rc;
+ }
+
+ rc = nxt_app_msg_write_raw(nxt_ruby_run_ctx.task, nxt_ruby_run_ctx.wmsg,
+ (u_char *) ": ", (sizeof(": ") - 1));
+ if (nxt_slow_path(rc != NXT_OK)) {
+ return rc;
+ }
+
+ rc = nxt_app_msg_write_raw(nxt_ruby_run_ctx.task, nxt_ruby_run_ctx.wmsg,
+ (u_char *) value, value_size);
+ if (nxt_slow_path(rc != NXT_OK)) {
+ return rc;
+ }
+
+ return nxt_app_msg_write_raw(nxt_ruby_run_ctx.task, nxt_ruby_run_ctx.wmsg,
+ (u_char *) "\r\n", (sizeof("\r\n") - 1));
+}
+
+
+static nxt_int_t
+nxt_ruby_rack_result_body(VALUE result)
+{
+ VALUE fn, body;
+ nxt_int_t rc;
+
+ body = rb_ary_entry(result, 2);
+
+ if (rb_respond_to(body, rb_intern("to_path"))) {
+
+ fn = rb_funcall(body, rb_intern("to_path"), 0);
+ if (nxt_slow_path(TYPE(fn) != T_STRING)) {
+ nxt_log(nxt_ruby_run_ctx.task, NXT_LOG_ERR,
+ "Ruby: Failed to get 'body' file path from application");
+
+ return NXT_ERROR;
+ }
+
+ rc = nxt_ruby_rack_result_body_file_write(fn);
+ if (nxt_slow_path(rc != NXT_OK)) {
+ return NXT_ERROR;
+ }
+
+ } else if (rb_respond_to(body, rb_intern("each"))) {
+ rb_iterate(rb_each, body, nxt_ruby_rack_result_body_each, 0);
+
+ } else {
+ nxt_log(nxt_ruby_run_ctx.task, NXT_LOG_ERR,
+ "Ruby: Invalid response 'body' format from application");
+
+ return NXT_ERROR;
+ }
+
+ if (rb_respond_to(body, rb_intern("close"))) {
+ rb_funcall(body, rb_intern("close"), 0);
+ }
+
+ return NXT_OK;
+}
+
+
+static nxt_int_t
+nxt_ruby_rack_result_body_file_write(VALUE filepath)
+{
+ size_t len;
+ ssize_t n;
+ nxt_off_t rest;
+ nxt_int_t rc;
+ nxt_file_t file;
+ nxt_file_info_t finfo;
+ u_char buf[8192];
+
+ nxt_memzero(&file, sizeof(nxt_file_t));
+
+ file.name = (nxt_file_name_t *) RSTRING_PTR(filepath);
+
+ rc = nxt_file_open(nxt_ruby_run_ctx.task, &file, NXT_FILE_RDONLY,
+ NXT_FILE_OPEN, 0);
+ if (nxt_slow_path(rc != NXT_OK)) {
+ nxt_log(nxt_ruby_run_ctx.task, NXT_LOG_ERR,
+ "Ruby: Failed to open 'body' file: %s",
+ (const char *) file.name);
+
+ return NXT_ERROR;
+ }
+
+ rc = nxt_file_info(&file, &finfo);
+ if (nxt_slow_path(rc != NXT_OK)) {
+ nxt_log(nxt_ruby_run_ctx.task, NXT_LOG_ERR,
+ "Ruby: Failed to get 'body' file information: %s",
+ (const char *) file.name);
+
+ goto fail;
+ }
+
+ rest = nxt_file_size(&finfo);
+
+ while (rest != 0) {
+ len = nxt_min(rest, (nxt_off_t) sizeof(buf));
+
+ n = nxt_file_read(&file, buf, len, nxt_file_size(&finfo) - rest);
+ if (nxt_slow_path(n != (ssize_t) len)) {
+ nxt_log(nxt_ruby_run_ctx.task, NXT_LOG_ERR,
+ "Ruby: Failed to read 'body' file");
+
+ goto fail;
+ }
+
+ rest -= len;
+
+ rc = nxt_app_msg_write_raw(nxt_ruby_run_ctx.task, nxt_ruby_run_ctx.wmsg,
+ buf, len);
+ if (nxt_slow_path(rc != NXT_OK)) {
+ nxt_log(nxt_ruby_run_ctx.task, NXT_LOG_ERR,
+ "Ruby: Failed to write 'body' from application");
+
+ goto fail;
+ }
+ }
+
+ nxt_file_close(nxt_ruby_run_ctx.task, &file);
+
+ return NXT_OK;
+
+fail:
+
+ nxt_file_close(nxt_ruby_run_ctx.task, &file);
+
+ return NXT_ERROR;
+}
+
+
+static VALUE
+nxt_ruby_rack_result_body_each(VALUE body)
+{
+ nxt_int_t rc;
+
+ if (TYPE(body) != T_STRING) {
+ return Qnil;
+ }
+
+ rc = nxt_app_msg_write_raw(nxt_ruby_run_ctx.task, nxt_ruby_run_ctx.wmsg,
+ (u_char *) RSTRING_PTR(body), RSTRING_LEN(body));
+ if (nxt_slow_path(rc != NXT_OK)) {
+ nxt_log(nxt_ruby_run_ctx.task, NXT_LOG_ERR,
+ "Ruby: Failed to write 'body' from application");
+ }
+
+ return Qnil;
+}
+
+
+static void
+nxt_ruby_exception_log(nxt_task_t *task, uint32_t level, const char *desc)
+{
+ int i;
+ VALUE err, ary, eclass, msg;
+
+ nxt_log(task, level, "Ruby: %s", desc);
+
+ err = rb_errinfo();
+ ary = rb_funcall(err, rb_intern("backtrace"), 0);
+
+ if (RARRAY_LEN(ary) == 0) {
+ return;
+ }
+
+ eclass = rb_class_name(rb_class_of(err));
+ msg = rb_funcall(err, rb_intern("message"), 0);
+
+ nxt_log(task, level, "Ruby: %s: %s (%s)",
+ RSTRING_PTR(RARRAY_PTR(ary)[0]),
+ RSTRING_PTR(msg), RSTRING_PTR(eclass));
+
+ for (i = 1; i < RARRAY_LEN(ary); i++) {
+ nxt_log(task, level, "from %s", RSTRING_PTR(RARRAY_PTR(ary)[i]));
+ }
+}
+
+
+static void
+nxt_ruby_atexit(nxt_task_t *task)
+{
+ rb_gc_unregister_address(&nxt_ruby_io_input);
+ rb_gc_unregister_address(&nxt_ruby_io_error);
+
+ rb_gc_unregister_address(&nxt_ruby_rackup);
+ rb_gc_unregister_address(&nxt_ruby_call);
+ rb_gc_unregister_address(&nxt_ruby_env);
+
+ ruby_cleanup(0);
+}
diff --git a/src/ruby/nxt_ruby.h b/src/ruby/nxt_ruby.h
new file mode 100644
index 00000000..9a6be0d4
--- /dev/null
+++ b/src/ruby/nxt_ruby.h
@@ -0,0 +1,34 @@
+
+/*
+ * Copyright (C) Alexander Borisov
+ * Copyright (C) NGINX, Inc.
+ */
+
+#ifndef _NXT_RUBY_H_INCLUDED_
+#define _NXT_RUBY_H_INCLUDED_
+
+
+#include <ruby.h>
+#include <ruby/io.h>
+#include <ruby/encoding.h>
+#include <ruby/version.h>
+
+#include <nxt_main.h>
+#include <nxt_router.h>
+#include <nxt_runtime.h>
+#include <nxt_application.h>
+
+
+typedef struct {
+ nxt_task_t *task;
+ nxt_app_rmsg_t *rmsg;
+ nxt_app_wmsg_t *wmsg;
+
+ size_t body_preread_size;
+} nxt_ruby_run_ctx_t;
+
+
+VALUE nxt_ruby_stream_io_input_init(void);
+VALUE nxt_ruby_stream_io_error_init(void);
+
+#endif /* _NXT_RUBY_H_INCLUDED_ */
diff --git a/src/ruby/nxt_ruby_stream_io.c b/src/ruby/nxt_ruby_stream_io.c
new file mode 100644
index 00000000..eee11c3c
--- /dev/null
+++ b/src/ruby/nxt_ruby_stream_io.c
@@ -0,0 +1,290 @@
+
+/*
+ * Copyright (C) Alexander Borisov
+ * Copyright (C) NGINX, Inc.
+ */
+
+#include <ruby/nxt_ruby.h>
+
+
+static VALUE nxt_ruby_stream_io_new(VALUE class, VALUE wrap);
+static VALUE nxt_ruby_stream_io_initialize(int argc, VALUE *argv, VALUE self);
+static VALUE nxt_ruby_stream_io_gets(VALUE obj, VALUE args);
+static size_t nxt_ruby_stream_io_read_line(nxt_app_rmsg_t *rmsg, VALUE str);
+static VALUE nxt_ruby_stream_io_each(VALUE obj, VALUE args);
+static VALUE nxt_ruby_stream_io_read(VALUE obj, VALUE args);
+static VALUE nxt_ruby_stream_io_rewind(VALUE obj, VALUE args);
+static VALUE nxt_ruby_stream_io_puts(VALUE obj, VALUE args);
+static VALUE nxt_ruby_stream_io_write(VALUE obj, VALUE args);
+nxt_inline long nxt_ruby_stream_io_s_write(nxt_ruby_run_ctx_t *run_ctx,
+ VALUE val);
+static VALUE nxt_ruby_stream_io_flush(VALUE obj, VALUE args);
+
+
+VALUE
+nxt_ruby_stream_io_input_init(void)
+{
+ VALUE stream_io;
+
+ stream_io = rb_define_class("NGINX_Unit_Stream_IO_Read", rb_cData);
+
+ rb_gc_register_address(&stream_io);
+
+ rb_define_singleton_method(stream_io, "new", nxt_ruby_stream_io_new, 1);
+ rb_define_method(stream_io, "initialize", nxt_ruby_stream_io_initialize, -1);
+ rb_define_method(stream_io, "gets", nxt_ruby_stream_io_gets, 0);
+ rb_define_method(stream_io, "each", nxt_ruby_stream_io_each, 0);
+ rb_define_method(stream_io, "read", nxt_ruby_stream_io_read, -2);
+ rb_define_method(stream_io, "rewind", nxt_ruby_stream_io_rewind, 0);
+
+ return stream_io;
+}
+
+
+VALUE
+nxt_ruby_stream_io_error_init(void)
+{
+ VALUE stream_io;
+
+ stream_io = rb_define_class("NGINX_Unit_Stream_IO_Error", rb_cData);
+
+ rb_gc_register_address(&stream_io);
+
+ rb_define_singleton_method(stream_io, "new", nxt_ruby_stream_io_new, 1);
+ rb_define_method(stream_io, "initialize", nxt_ruby_stream_io_initialize, -1);
+ rb_define_method(stream_io, "puts", nxt_ruby_stream_io_puts, -2);
+ rb_define_method(stream_io, "write", nxt_ruby_stream_io_write, -2);
+ rb_define_method(stream_io, "flush", nxt_ruby_stream_io_flush, 0);
+
+ return stream_io;
+}
+
+
+static VALUE
+nxt_ruby_stream_io_new(VALUE class, VALUE wrap)
+{
+ VALUE self;
+ nxt_ruby_run_ctx_t *run_ctx;
+
+ Data_Get_Struct(wrap, nxt_ruby_run_ctx_t, run_ctx);
+ self = Data_Wrap_Struct(class, 0, 0, run_ctx);
+
+ rb_obj_call_init(self, 0, NULL);
+
+ return self;
+}
+
+
+static VALUE
+nxt_ruby_stream_io_initialize(int argc, VALUE *argv, VALUE self)
+{
+ return self;
+}
+
+
+static VALUE
+nxt_ruby_stream_io_gets(VALUE obj, VALUE args)
+{
+ VALUE buf;
+ nxt_ruby_run_ctx_t *run_ctx;
+
+ Data_Get_Struct(obj, nxt_ruby_run_ctx_t, run_ctx);
+
+ if (run_ctx->body_preread_size == 0) {
+ return Qnil;
+ }
+
+ buf = rb_str_buf_new(1);
+
+ if (buf == Qnil) {
+ return Qnil;
+ }
+
+ run_ctx->body_preread_size -= nxt_ruby_stream_io_read_line(run_ctx->rmsg,
+ buf);
+
+ return buf;
+}
+
+
+static size_t
+nxt_ruby_stream_io_read_line(nxt_app_rmsg_t *rmsg, VALUE str)
+{
+ size_t len, size;
+ u_char *p;
+ nxt_buf_t *buf;
+
+ len = 0;
+
+ for (buf = rmsg->buf; buf != NULL; buf = buf->next) {
+
+ size = nxt_buf_mem_used_size(&buf->mem);
+ p = memchr(buf->mem.pos, '\n', size);
+
+ if (p != NULL) {
+ p++;
+ size = p - buf->mem.pos;
+
+ rb_str_cat(str, (const char *) buf->mem.pos, size);
+
+ len += size;
+ buf->mem.pos = p;
+
+ break;
+ }
+
+ rb_str_cat(str, (const char *) buf->mem.pos, size);
+
+ len += size;
+ buf->mem.pos = buf->mem.free;
+ }
+
+ rmsg->buf = buf;
+
+ return len;
+}
+
+
+static VALUE
+nxt_ruby_stream_io_each(VALUE obj, VALUE args)
+{
+ VALUE chunk;
+
+ if (rb_block_given_p() == 0) {
+ rb_raise(rb_eArgError, "Expected block on rack.input 'each' method");
+ }
+
+ for ( ;; ) {
+ chunk = nxt_ruby_stream_io_gets(obj, Qnil);
+
+ if (chunk == Qnil) {
+ return Qnil;
+ }
+
+ rb_yield(chunk);
+ }
+
+ return Qnil;
+}
+
+
+static VALUE
+nxt_ruby_stream_io_read(VALUE obj, VALUE args)
+{
+ VALUE buf;
+ long copy_size, u_size;
+ size_t len;
+ nxt_ruby_run_ctx_t *run_ctx;
+
+ Data_Get_Struct(obj, nxt_ruby_run_ctx_t, run_ctx);
+
+ copy_size = run_ctx->body_preread_size;
+
+ if (RARRAY_LEN(args) > 0 && TYPE(RARRAY_PTR(args)[0]) == T_FIXNUM) {
+ u_size = NUM2LONG(RARRAY_PTR(args)[0]);
+
+ if (u_size < 0 || copy_size == 0) {
+ return Qnil;
+ }
+
+ if (copy_size > u_size) {
+ copy_size = u_size;
+ }
+ }
+
+ if (copy_size == 0) {
+ return rb_str_new_cstr("");
+ }
+
+ buf = rb_str_buf_new(copy_size);
+
+ if (nxt_slow_path(buf == Qnil)) {
+ return Qnil;
+ }
+
+ len = nxt_app_msg_read_raw(run_ctx->task, run_ctx->rmsg,
+ RSTRING_PTR(buf), (size_t) copy_size);
+
+ if (RARRAY_LEN(args) > 1 && TYPE(RARRAY_PTR(args)[1]) == T_STRING) {
+
+ rb_str_set_len(RARRAY_PTR(args)[1], 0);
+ rb_str_cat(RARRAY_PTR(args)[1], RSTRING_PTR(buf), copy_size);
+ }
+
+ rb_str_set_len(buf, (long) len);
+
+ run_ctx->body_preread_size -= len;
+
+ return buf;
+}
+
+
+static VALUE
+nxt_ruby_stream_io_rewind(VALUE obj, VALUE args)
+{
+ return Qnil;
+}
+
+
+static VALUE
+nxt_ruby_stream_io_puts(VALUE obj, VALUE args)
+{
+ nxt_ruby_run_ctx_t *run_ctx;
+
+ if (RARRAY_LEN(args) != 1) {
+ return Qnil;
+ }
+
+ Data_Get_Struct(obj, nxt_ruby_run_ctx_t, run_ctx);
+
+ nxt_ruby_stream_io_s_write(run_ctx, RARRAY_PTR(args)[0]);
+
+ return Qnil;
+}
+
+
+static VALUE
+nxt_ruby_stream_io_write(VALUE obj, VALUE args)
+{
+ long len;
+ nxt_ruby_run_ctx_t *run_ctx;
+
+ if (RARRAY_LEN(args) != 1) {
+ return Qnil;
+ }
+
+ Data_Get_Struct(obj, nxt_ruby_run_ctx_t, run_ctx);
+
+ len = nxt_ruby_stream_io_s_write(run_ctx, RARRAY_PTR(args)[0]);
+
+ return LONG2FIX(len);
+}
+
+
+nxt_inline long
+nxt_ruby_stream_io_s_write(nxt_ruby_run_ctx_t *run_ctx, VALUE val)
+{
+ if (nxt_slow_path(val == Qnil)) {
+ return 0;
+ }
+
+ if (TYPE(val) != T_STRING) {
+ val = rb_funcall(val, rb_intern("to_s"), 0);
+
+ if (TYPE(val) != T_STRING) {
+ return 0;
+ }
+ }
+
+ nxt_log_error(NXT_LOG_ERR, run_ctx->task->log, "Ruby: %s",
+ RSTRING_PTR(val));
+
+ return RSTRING_LEN(val);
+}
+
+
+static VALUE
+nxt_ruby_stream_io_flush(VALUE obj, VALUE args)
+{
+ return Qnil;
+}
generated by cgit (git 2.34.1) at 2024-11-16 21:25:31 +0000