Router: read configuration from port.

Controller: stub to send configuration from POST body "as is" to router.
This commit is contained in:
Max Romanov 2017-07-06 18:38:51 +03:00
parent e1219569f4
commit c56d2af3bc
6 changed files with 96 additions and 19 deletions

View file

@ -562,6 +562,38 @@ nxt_controller_process_request(nxt_task_t *task, nxt_conn_t *c,
nxt_memzero(&resp, sizeof(nxt_controller_response_t));
if (nxt_str_eq(&req->parser.method, "POST", 4)) {
nxt_port_t *port;
nxt_runtime_t *rt;
rt = task->thread->runtime;
nxt_runtime_port_each(rt, port) {
if (nxt_pid == port->pid) {
continue;
}
if (port->type == NXT_PROCESS_ROUTER) {
nxt_buf_t *b, *src;
src = c->read;
b = nxt_port_mmap_get_buf(task, port,
nxt_buf_mem_used_size(&src->mem));
nxt_memcpy(b->mem.pos, src->mem.pos,
nxt_buf_mem_used_size(&src->mem));
b->mem.free += nxt_buf_mem_used_size(&src->mem);
(void) nxt_port_socket_write(task, port, NXT_PORT_MSG_DATA,
-1, 0, 0, b);
break;
}
} nxt_runtime_port_loop;
}
if (nxt_str_eq(&req->parser.method, "GET", 3)) {
value = nxt_conf_get_path(nxt_controller_conf.root, &path);

View file

@ -166,7 +166,7 @@ nxt_master_start_router_process(nxt_task_t *task, nxt_runtime_t *rt)
init->start = nxt_router_start;
init->name = "router process";
init->user_cred = &rt->user_cred;
init->port_handlers = nxt_worker_process_port_handlers;
init->port_handlers = nxt_router_process_port_handlers;
init->signals = nxt_worker_process_signals;
init->type = NXT_PROCESS_ROUTER;

View file

@ -18,6 +18,7 @@ nxt_int_t nxt_router_start(nxt_task_t *task, nxt_runtime_t *rt);
extern nxt_port_handler_t nxt_worker_process_port_handlers[];
extern nxt_port_handler_t nxt_app_process_port_handlers[];
extern nxt_port_handler_t nxt_router_process_port_handlers[];
extern const nxt_sig_event_t nxt_master_process_signals[];
extern const nxt_sig_event_t nxt_worker_process_signals[];

View file

@ -80,6 +80,7 @@ static void nxt_router_conn_error(nxt_task_t *task, void *obj, void *data);
static void nxt_router_conn_timeout(nxt_task_t *task, void *obj, void *data);
static nxt_msec_t nxt_router_conn_timeout_value(nxt_conn_t *c, uintptr_t data);
static nxt_router_t *nxt_router;
nxt_int_t
nxt_router_start(nxt_task_t *task, nxt_runtime_t *rt)
@ -100,6 +101,8 @@ nxt_router_start(nxt_task_t *task, nxt_runtime_t *rt)
nxt_queue_init(&router->engines);
nxt_queue_init(&router->sockets);
nxt_router = router;
return NXT_OK;
}
@ -895,14 +898,14 @@ nxt_router_engine_post(nxt_router_engine_conf_t *recf)
static void
nxt_router_data_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg);
nxt_router_app_data_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg);
nxt_port_handler_t nxt_router_process_port_handlers[] = {
static nxt_port_handler_t nxt_router_app_port_handlers[] = {
NULL,
nxt_port_new_port_handler,
nxt_port_change_log_file_handler,
nxt_port_mmap_handler,
nxt_router_data_handler,
nxt_router_app_data_handler,
};
@ -930,7 +933,7 @@ nxt_router_thread_start(void *data)
thread->fiber = &engine->fibers->fiber;
engine->port->socket.task = task;
nxt_port_create(task, engine->port, nxt_router_process_port_handlers);
nxt_port_create(task, engine->port, nxt_router_app_port_handlers);
engine->mem_pool = nxt_mp_create(4096, 128, 1024, 64);
@ -1196,8 +1199,37 @@ static const nxt_conn_state_t nxt_router_conn_write_state
};
void
nxt_router_conf_data_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
{
size_t dump_size;
nxt_buf_t *b;
nxt_int_t ret;
b = msg->buf;
dump_size = nxt_buf_used_size(b);
if (dump_size > 300) {
dump_size = 300;
}
nxt_debug(task, "router conf data (%z): %*s",
msg->size, dump_size, b->mem.pos);
ret = nxt_router_new_conf(task, task->thread->runtime, nxt_router,
b->mem.pos, b->mem.free);
b->mem.pos = b->mem.free;
if (nxt_slow_path(ret != NXT_OK)) {
nxt_log_alert(task->log, "Failed to apply new conf");
}
}
static void
nxt_router_data_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
nxt_router_app_data_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
{
size_t dump_size;
nxt_buf_t *b, *i, *last;
@ -1229,7 +1261,7 @@ nxt_router_data_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
dump_size = 300;
}
nxt_debug(task, "%srouter data (%z): %*s",
nxt_debug(task, "%srouter app data (%z): %*s",
msg->port_msg.last ? "last " : "", msg->size, dump_size,
b->mem.pos);
@ -1268,24 +1300,24 @@ nxt_router_data_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
nxt_inline nxt_port_t *
nxt_router_app_port(nxt_task_t *task)
{
nxt_port_t *port;
nxt_runtime_t *rt;
nxt_port_t *port;
nxt_runtime_t *rt;
rt = task->thread->runtime;
rt = task->thread->runtime;
nxt_runtime_port_each(rt, port) {
nxt_runtime_port_each(rt, port) {
if (nxt_pid == port->pid) {
continue;
}
if (nxt_pid == port->pid) {
continue;
}
if (port->type == NXT_PROCESS_WORKER) {
return port;
}
if (port->type == NXT_PROCESS_WORKER) {
return port;
}
} nxt_runtime_port_loop;
} nxt_runtime_port_loop;
return NULL;
return NULL;
}

View file

@ -87,5 +87,7 @@ typedef struct {
nxt_int_t nxt_router_new_conf(nxt_task_t *task, nxt_runtime_t *rt,
nxt_router_t *router, u_char *start, u_char *end);
void nxt_router_conf_data_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg);
#endif /* _NXT_ROUTER_H_INCLUDED_ */

View file

@ -8,6 +8,7 @@
#include <nxt_runtime.h>
#include <nxt_port.h>
#include <nxt_master_process.h>
#include <nxt_router.h>
static void nxt_worker_process_quit(nxt_task_t *task);
@ -39,6 +40,15 @@ nxt_port_handler_t nxt_app_process_port_handlers[] = {
};
nxt_port_handler_t nxt_router_process_port_handlers[] = {
nxt_worker_process_quit_handler,
nxt_port_new_port_handler,
nxt_port_change_log_file_handler,
nxt_port_mmap_handler,
nxt_router_conf_data_handler,
};
const nxt_sig_event_t nxt_worker_process_signals[] = {
nxt_event_signal(SIGHUP, nxt_worker_process_signal_handler),
nxt_event_signal(SIGINT, nxt_worker_process_sigterm_handler),