Router: read configuration from port.

Controller: stub to send configuration from POST body "as is" to router.

6 files changed, 96 insertions, 19 deletions

diff --git a/src/nxt_controller.c b/src/nxt_controller.c
index 757c294d..6403c98a 100644
--- a/src/nxt_controller.c
+++ b/src/nxt_controller.c
@@ -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);
diff --git a/src/nxt_master_process.c b/src/nxt_master_process.c
index 54b5c11c..d2262a1b 100644
--- a/src/nxt_master_process.c
+++ b/src/nxt_master_process.c
@@ -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;
 
diff --git a/src/nxt_master_process.h b/src/nxt_master_process.h
index ff84ca9f..417c0aa5 100644
--- a/src/nxt_master_process.h
+++ b/src/nxt_master_process.h
@@ -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[];
 
diff --git a/src/nxt_router.c b/src/nxt_router.c
index 1f3a0dfe..f063d87d 100644
--- a/src/nxt_router.c
+++ b/src/nxt_router.c
@@ -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;
 }
 
 
diff --git a/src/nxt_router.h b/src/nxt_router.h
index 9c6f75b4..a7fb54a9 100644
--- a/src/nxt_router.h
+++ b/src/nxt_router.h
@@ -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_ */
diff --git a/src/nxt_worker_process.c b/src/nxt_worker_process.c
index 4c5b6888..b0c68160 100644
--- a/src/nxt_worker_process.c
+++ b/src/nxt_worker_process.c
@@ -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),