Port message fragmentation supported.

- Each sendmsg() transmits no more than port->max_size payload data.
- Longer buffers are fragmented and send using multiple sendmsg() calls.
- On receive side, buffers are connected in chain.
- Number of handler calls is the same as number of nxt_port_socket_write()
  calls.
- nxt_buf_make_plain() function introduced to make single plain buffer from
  the chain.

1 files changed, 9 insertions, 15 deletions

diff --git a/src/nxt_router.c b/src/nxt_router.c
index fddc3582..d46dae72 100644
--- a/src/nxt_router.c
+++ b/src/nxt_router.c
@@ -592,27 +592,19 @@ nxt_router_new_port_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
 void
 nxt_router_conf_data_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
 {
-    size_t                  dump_size;
     nxt_int_t               ret;
     nxt_buf_t               *b;
     nxt_router_temp_conf_t  *tmcf;
 
-    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);
-
     tmcf = nxt_router_temp_conf(task);
     if (nxt_slow_path(tmcf == NULL)) {
         return;
     }
 
+    b = nxt_buf_chk_make_plain(tmcf->conf->mem_pool, msg->buf, msg->size);
+
+    nxt_assert(b != NULL);
+
     tmcf->conf->router = nxt_router;
     tmcf->stream = msg->port_msg.stream;
     tmcf->port = nxt_runtime_port_find(task->thread->runtime,
@@ -1442,8 +1434,12 @@ nxt_router_listen_socket_error(nxt_task_t *task, nxt_port_recv_msg_t *msg,
 
     rpc = data;
     sa = rpc->socket_conf->sockaddr;
+    tmcf = rpc->temp_conf;
+
+    in = nxt_buf_chk_make_plain(tmcf->mem_pool, msg->buf, msg->size);
+
+    nxt_assert(in != NULL);
 
-    in = msg->buf;
     p = in->mem.pos;
 
     error = *p++;
@@ -1452,8 +1448,6 @@ nxt_router_listen_socket_error(nxt_task_t *task, nxt_port_recv_msg_t *msg,
            + sizeof("{listener: \"\", code:\"\", message: \"\"}") - 1
            + sa->length + socket_errors[error].length + (in->mem.free - p);
 
-    tmcf = rpc->temp_conf;
-
     out = nxt_buf_mem_alloc(tmcf->mem_pool, size, 0);
     if (nxt_slow_path(out == NULL)) {
         return;