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 909

static VALUE
add_mime(VALUE self, VALUE suffix, VALUE type) {
    struct _Err	err = ERR_INIT;
    
    if (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 838

static VALUE
handle(VALUE self, VALUE method, VALUE pattern, VALUE handler) {
    Hook	hook;
    Method	meth = ALL;
    const char	*pat;

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

    if (connect_sym == method) {
	meth = CONNECT;
    } else if (delete_sym == method) {
	meth = DELETE;
    } else if (get_sym == method) {
	meth = GET;
    } else if (head_sym == method) {
	meth = HEAD;
    } else if (options_sym == method) {
	meth = OPTIONS;
    } else if (post_sym == method) {
	meth = POST;
    } else if (put_sym == method) {
	meth = PUT;
    } else if (Qnil == method) {
	meth = ALL;
    } else {
	rb_raise(rb_eArgError, "invalid method");
    }
    if (NULL == (hook = rhook_create(meth, pat, handler, &the_rserver.eval_queue))) {
	rb_raise(rb_eStandardError, "out of memory.");
    } else {
	Hook	h;
	Hook	prev = NULL;

	for (h = the_server.hooks; NULL != h; h = h->next) {
	    prev = h;
	}
	if (NULL != prev) {
	    prev->next = hook;
	} else {
	    the_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 892

static VALUE
handle_not_found(VALUE self, VALUE handler) {
    if (NULL == (the_server.hook404 = rhook_create(GET, "/", handler, &the_rserver.eval_queue))) {
	rb_raise(rb_eStandardError, "out of memory.");
    }
    rb_gc_register_address((VALUE*)&the_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: “127.0.0.1:6464”, “unix:///tmp/agoo.socket”, “[::1]:6464, or to not restrict the address ”:6464“.

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

static VALUE
rserver_init(int argc, VALUE *argv, VALUE self) {
    struct _Err	err = 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];
    }
    server_setup();
    sub_init(&the_rserver.sub_cache);

    if (ERR_OK != configure(&err, port, root, options)) {
	rb_raise(rb_eArgError, "%s", err.msg);
    }
    queue_multi_init(&the_rserver.pub_queue, 256, true, false);
    queue_multi_init(&the_rserver.eval_queue, 1024, false, true);

    pthread_mutex_init(&the_rserver.up_lock, 0);
    the_rserver.up_list = NULL;
    
    the_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 927

static VALUE
path_group(VALUE self, VALUE path, VALUE dirs) {
    Group	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 783

VALUE
rserver_shutdown(VALUE self) {
    if (the_server.inited) {
	server_shutdown("Agoo", stop_runners);

	queue_cleanup(&the_rserver.pub_queue);
	queue_cleanup(&the_rserver.eval_queue);

	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", the_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", the_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 684

static VALUE
rserver_start(VALUE self) {
    VALUE	*vp;
    int		i;
    int		pid;
    double	giveup;
    struct _Err	err = 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 (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
	    log_cat(&error_cat, "Failed to fork. %s.", strerror(errno));
	    break;
	} else if (0 == pid) {
	    log_start(true);
	    break;
	} else {
	    the_rserver.worker_pids[i] = pid;
	}
    }
    if (ERR_OK != server_start(&err, "Agoo", StringValuePtr(v))) {
	rb_raise(rb_eStandardError, "%s", err.msg);
    }
    if (0 >= the_server.thread_cnt) {
	Req		req;

	while (the_server.active) {
	    if (NULL != (req = (Req)queue_pop(&the_rserver.eval_queue, 0.1))) {
		handle_protected(req, false);
		request_destroy(req);
	    } else {
		rb_thread_schedule();
	    }

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

	for (i = the_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 + the_server.thread_cnt <= atomic_load(&the_server.running)) {
		break;
	    }
	}
    }
    return Qnil;
}