Module: Agoo::Server

Defined in:

ext/agoo/rserver.c

Class Method Summary

• .add_mime(suffix, type) ⇒ Object

  call-seq: add_mime(suffix, type).

• .handle(method, pattern, handler) ⇒ Object

  call-seq: handle(method, pattern, handler).

• .handle_not_found(handler) ⇒ Object

  call-seq: not_found_handle(handler).

• .init(*args) ⇒ Object

  call-seq: init(port, root, options).

• .path_group(path, dirs) ⇒ Object

  call-seq: path_group(path, dirs).

• .shutdown ⇒ Object

  call-seq: shutdown().

• .start ⇒ Object

  call-seq: start().

Class Method Details

.add_mime(suffix, type) ⇒ Object

call-seq: add_mime(suffix, type)

Adds a mime type by associating a type string with a suffix. This is used for static files.

# File 'ext/agoo/rserver.c', line 980

static VALUE
add_mime(VALUE self, VALUE suffix, VALUE type) {
    struct _agooErr	err = AGOO_ERR_INIT;
    
    if (AGOO_ERR_OK != mime_set(&err, StringValuePtr(suffix), StringValuePtr(type))) {
	rb_raise(rb_eArgError, "%s", err.msg);
    }
    return Qnil;
}

.handle(method, pattern, handler) ⇒ Object

call-seq: handle(method, pattern, handler)

Registers a handler for the HTTP method and path pattern specified. The path pattern follows glob like rules in that a single * matches a single token bounded by the ‘/` character and a double ** matches all remaining.

# File 'ext/agoo/rserver.c', line 872

static VALUE
handle(VALUE self, VALUE method, VALUE pattern, VALUE handler) {
    agooHook	hook;
    agooMethod	meth = AGOO_ALL;
    const char	*pat;
    ID		static_id = rb_intern("static?");

    rb_check_type(pattern, T_STRING);
    pat = StringValuePtr(pattern);

    if (connect_sym == method) {
	meth = AGOO_CONNECT;
    } else if (delete_sym == method) {
	meth = AGOO_DELETE;
    } else if (get_sym == method) {
	meth = AGOO_GET;
    } else if (head_sym == method) {
	meth = AGOO_HEAD;
    } else if (options_sym == method) {
	meth = AGOO_OPTIONS;
    } else if (post_sym == method) {
	meth = AGOO_POST;
    } else if (put_sym == method) {
	meth = AGOO_PUT;
    } else if (Qnil == method) {
	meth = AGOO_ALL;
    } else {
	rb_raise(rb_eArgError, "invalid method");
    }
    if (T_STRING == rb_type(handler)) {
	handler = resolve_classpath(StringValuePtr(handler), RSTRING_LEN(handler));
    }
    if (rb_respond_to(handler, static_id)) {
	if (Qtrue == rb_funcall(handler, static_id, 0, Qnil)) {
	    VALUE	res = rb_funcall(handler, call_id, 1, Qnil);
	    VALUE	bv;
	    
	    rb_check_type(res, T_ARRAY);
	    if (3 != RARRAY_LEN(res)) {
		rb_raise(rb_eArgError, "a rack call() response must be an array of 3 members.");
	    }
	    bv = rb_ary_entry(res, 2);
	    if (T_ARRAY == rb_type(bv)) {
		int		i;
		int		bcnt = (int)RARRAY_LEN(bv);
		agooText	t = agoo_text_allocate(1024);
		struct _agooErr	err = AGOO_ERR_INIT;
		VALUE		v;

		if (NULL == t) {
		    rb_raise(rb_eArgError, "failed to allocate response.");
		}
		for (i = 0; i < bcnt; i++) {
		    v = rb_ary_entry(bv, i);
		    t = agoo_text_append(t, StringValuePtr(v), (int)RSTRING_LEN(v));
		}
		if (NULL == agoo_page_immutable(&err, pat, t->text, t->len)) {
		    rb_raise(rb_eArgError, "%s", err.msg);
		}
		agoo_text_release(t);

		return Qnil;
	    }
	}
    }
    if (NULL == (hook = rhook_create(meth, pat, handler, &agoo_server.eval_queue))) {
	rb_raise(rb_eStandardError, "out of memory.");
    } else {
	agooHook	h;
	agooHook	prev = NULL;

	for (h = agoo_server.hooks; NULL != h; h = h->next) {
	    prev = h;
	}
	if (NULL != prev) {
	    prev->next = hook;
	} else {
	    agoo_server.hooks = hook;
	}
	rb_gc_register_address((VALUE*)&hook->handler);
    }
    return Qnil;
}

.handle_not_found(handler) ⇒ Object

call-seq: not_found_handle(handler)

Registers a handler to be called when no other hook is found and no static file is found.

# File 'ext/agoo/rserver.c', line 963

static VALUE
handle_not_found(VALUE self, VALUE handler) {
    if (NULL == (agoo_server.hook404 = rhook_create(AGOO_GET, "/", handler, &agoo_server.eval_queue))) {
	rb_raise(rb_eStandardError, "out of memory.");
    }
    rb_gc_register_address((VALUE*)&agoo_server.hook404->handler);
    
    return Qnil;
}

.init(*args) ⇒ Object

call-seq: init(port, root, options)

Configures the server that will listen on the designated port and using the root as the root of the static resources. Logging is feature based and not level based and the options reflect that approach. If bind option is to be used instead of the port then set the port to zero.

• options [Hash] server options

  • :pedantic [true|false] if true response header and status codes are checked and an exception raised if they violate the rack spec at github.com/rack/rack/blob/master/SPEC, tools.ietf.org/html/rfc3875#section-4.1.18, or tools.ietf.org/html/rfc7230.

  • :thread_count [Integer] number of ruby worker threads. Defaults to one. If zero then the start function will not return but instead will proess using the thread that called start. Usually the default is best unless the workers are making IO calls.

  • :worker_count [Integer] number of workers to fork. Defaults to one which is not to fork.

  • :bind [String|Array] a binding or array of binds. Examples are: “http ://127.0.0.1:6464”, “unix:///tmp/agoo.socket”, “http ://[::1]:6464, or to not restrict the address ”http ://:6464“.

# File 'ext/agoo/rserver.c', line 251

static VALUE
rserver_init(int argc, VALUE *argv, VALUE self) {
    struct _agooErr	err = AGOO_ERR_INIT;
    int			port;
    const char		*root;
    VALUE		options = Qnil;

    if (argc < 2 || 3 < argc) {
	rb_raise(rb_eArgError, "Wrong number of arguments to Agoo::Server.configure.");
    }
    port = FIX2INT(argv[0]);
    rb_check_type(argv[1], T_STRING);
    root = StringValuePtr(argv[1]);
    if (3 <= argc) {
	options = argv[2];
    }
    agoo_server_setup();
    agoo_server.ctx_nil_value = (void*)Qnil;
    agoo_server.env_nil_value = (void*)Qnil;

    if (AGOO_ERR_OK != configure(&err, port, root, options)) {
	rb_raise(rb_eArgError, "%s", err.msg);
    }
    agoo_server.inited = true;

    return Qnil;
}

.path_group(path, dirs) ⇒ Object

call-seq: path_group(path, dirs)

Sets up a path group where the path defines a group of directories to search for a file. For example a path of ‘/assets’ could be mapped to a set of [ ‘home/user/images’, ‘/home/user/app/assets/images’ ].

# File 'ext/agoo/rserver.c', line 998

static VALUE
path_group(VALUE self, VALUE path, VALUE dirs) {
    agooGroup	g;

    rb_check_type(path, T_STRING);
    rb_check_type(dirs, T_ARRAY);

    if (NULL != (g = group_create(StringValuePtr(path)))) {
	int	i;
	int	dcnt = (int)RARRAY_LEN(dirs);
	VALUE	entry;
	
	for (i = dcnt - 1; 0 <= i; i--) {
	    entry = rb_ary_entry(dirs, i);
	    if (T_STRING != rb_type(entry)) {
		entry = rb_funcall(entry, rb_intern("to_s"), 0);
	    }
	    group_add(g, StringValuePtr(entry));
	}
    }
    return Qnil;
}

.shutdown ⇒ Object

call-seq: shutdown()

Shutdown the server. Logs and queues are flushed before shutting down.

# File 'ext/agoo/rserver.c', line 820

VALUE
rserver_shutdown(VALUE self) {
    if (agoo_server.inited) {
	agoo_server_shutdown("Agoo", stop_runners);

	if (1 < the_rserver.worker_cnt && getpid() == *the_rserver.worker_pids) {
	    int	i;
	    int	status;
	    int	exit_cnt = 1;
	    int	j;
	    
	    for (i = 1; i < the_rserver.worker_cnt; i++) {
		kill(the_rserver.worker_pids[i], SIGKILL);
	    }
	    for (j = 0; j < 20; j++) {
		for (i = 1; i < the_rserver.worker_cnt; i++) {
		    if (0 == the_rserver.worker_pids[i]) {
			continue;
		    }
		    if (0 < waitpid(the_rserver.worker_pids[i], &status, WNOHANG)) {
			if (WIFEXITED(status)) {
			    //printf("exited, status=%d for %d\n", agoo_server.worker_pids[i], WEXITSTATUS(status));
			    the_rserver.worker_pids[i] = 0;
			    exit_cnt++;
			} else if (WIFSIGNALED(status)) {
			    //printf("*** killed by signal %d for %d\n", agoo_server.worker_pids[i], WTERMSIG(status));
			    the_rserver.worker_pids[i] = 0;
			    exit_cnt++;
			}
		    }
		}
		if (the_rserver.worker_cnt <= exit_cnt) {
		    break;
		}
		dsleep(0.2);
	    }
	    if (exit_cnt < the_rserver.worker_cnt) {
		printf("*-*-* Some workers did not exit.\n");
	    }
	}
    }
    return Qnil;
}

.start ⇒ Object

call-seq: start()

Start the server.

# File 'ext/agoo/rserver.c', line 712

static VALUE
rserver_start(VALUE self) {
    VALUE		*vp;
    int			i;
    int			pid;
    double		giveup;
    struct _agooErr	err = AGOO_ERR_INIT;
    VALUE		agoo = rb_const_get_at(rb_cObject, rb_intern("Agoo"));
    VALUE		v = rb_const_get_at(agoo, rb_intern("VERSION"));
    
    *the_rserver.worker_pids = getpid();
    
    // If workers then set the loop_max based on the expected number of
    // threads per worker.
    if (1 < the_rserver.worker_cnt) {
	agoo_server.loop_max /= the_rserver.worker_cnt;
	if (agoo_server.loop_max < 1) {
	    agoo_server.loop_max = 1;
	}
    }
    if (AGOO_ERR_OK != setup_listen(&err)) {
	rb_raise(rb_eIOError, "%s", err.msg);
    }
    for (i = 1; i < the_rserver.worker_cnt; i++) {
	VALUE	rpid = rb_funcall(rb_cObject, rb_intern("fork"), 0);

	if (Qnil == rpid) {
	    pid = 0;
	} else {
	    pid = NUM2INT(rpid);
	}
	if (0 > pid) { // error, use single process
	    agoo_log_cat(&agoo_error_cat, "Failed to fork. %s.", strerror(errno));
	    break;
	} else if (0 == pid) {
	    if (AGOO_ERR_OK != agoo_log_start(&err, true)) {
		rb_raise(rb_eStandardError, "%s", err.msg);
	    }
	    break;
	} else {
	    the_rserver.worker_pids[i] = pid;
	}
    }
    if (AGOO_ERR_OK != agoo_server_start(&err, "Agoo", StringValuePtr(v))) {
	rb_raise(rb_eStandardError, "%s", err.msg);
    }
    if (0 >= agoo_server.thread_cnt) {
	agooReq		req;

	while (agoo_server.active) {
	    if (NULL != (req = (agooReq)agoo_queue_pop(&agoo_server.eval_queue, 0.1))) {
		handle_protected(req, false);
		agoo_req_destroy(req);
	    } else {
		rb_thread_schedule();
	    }

	}
    } else {
	the_rserver.eval_threads = (VALUE*)malloc(sizeof(VALUE) * (agoo_server.thread_cnt + 1));
	DEBUG_ALLOC(mem_eval_threads, agoo_server.eval_threads);

	for (i = agoo_server.thread_cnt, vp = the_rserver.eval_threads; 0 < i; i--, vp++) {
	    *vp = rb_thread_create(wrap_process_loop, NULL);
	}
	*vp = Qnil;

	giveup = dtime() + 1.0;
	while (dtime() < giveup) {
	    // The processing threads will not start until this thread
	    // releases ownership so do that and then see if the threads has
	    // been started yet.
	    rb_thread_schedule();
	    if (2 + agoo_server.thread_cnt <= (long)atomic_load(&agoo_server.running)) {
		break;
	    }
	}
    }
    return Qnil;
}