Extruded nxt_port_send_port() to send port to one process.

1 files changed, 34 insertions, 26 deletions

diff --git a/src/nxt_port.c b/src/nxt_port.c
index e9553bd6..eb320640 100644
--- a/src/nxt_port.c
+++ b/src/nxt_port.c
@@ -81,13 +81,10 @@ void
 nxt_port_send_new_port(nxt_task_t *task, nxt_runtime_t *rt,
     nxt_port_t *new_port)
 {
-    nxt_buf_t                *b;
-    nxt_port_t               *port;
-    nxt_process_t            *process;
-    nxt_port_msg_new_port_t  *msg;
+    nxt_process_t  *process;
 
     nxt_debug(task, "new port %d for process %PI engine %uD",
-              new_port->socket.fd, new_port->pid, new_port->engine);
+              new_port->pair[1], new_port->pid, new_port->engine);
 
     nxt_runtime_process_each(rt, process)
     {
@@ -95,33 +92,44 @@ nxt_port_send_new_port(nxt_task_t *task, nxt_runtime_t *rt,
             continue;
         }
 
-        port = nxt_process_port_first(process);
+        (void) nxt_port_send_port(task, nxt_process_port_first(process),
+                                  new_port);
+    }
+    nxt_runtime_process_loop;
+}
 
-        b = nxt_buf_mem_alloc(port->mem_pool, sizeof(nxt_port_data_t), 0);
 
-        if (nxt_slow_path(b == NULL)) {
-            continue;
-        }
+nxt_int_t
+nxt_port_send_port(nxt_task_t *task, nxt_port_t *port, nxt_port_t *new_port)
+{
+    nxt_mp_t                 *mp;
+    nxt_buf_t                *b;
+    nxt_port_msg_new_port_t  *msg;
 
-        nxt_debug(task, "send new port %FD to process %PI",
-                  new_port->socket.fd, process->pid);
+    mp = port->mem_pool;
 
-        b->data = port->mem_pool;
-        b->completion_handler = nxt_port_new_port_buf_completion;
-        b->mem.free += sizeof(nxt_port_msg_new_port_t);
-        msg = (nxt_port_msg_new_port_t *) b->mem.pos;
+    b = nxt_buf_mem_alloc(mp, sizeof(nxt_port_data_t), 0);
+    if (nxt_slow_path(b == NULL)) {
+        return NXT_ERROR;
+    }
 
-        msg->id = new_port->id;
-        msg->pid = new_port->pid;
-        msg->engine = new_port->engine;
-        msg->max_size = port->max_size;
-        msg->max_share = port->max_share;
-        msg->type = new_port->type;
+    nxt_debug(task, "send port %FD to process %PI",
+              new_port->pair[1], port->pid);
 
-        (void) nxt_port_socket_write(task, port, NXT_PORT_MSG_NEW_PORT,
-                                     new_port->socket.fd, 0, 0, b);
-    }
-    nxt_runtime_process_loop;
+    b->data = mp;
+    b->completion_handler = nxt_port_new_port_buf_completion;
+    b->mem.free += sizeof(nxt_port_msg_new_port_t);
+    msg = (nxt_port_msg_new_port_t *) b->mem.pos;
+
+    msg->id = new_port->id;
+    msg->pid = new_port->pid;
+    msg->engine = new_port->engine;
+    msg->max_size = port->max_size;
+    msg->max_share = port->max_share;
+    msg->type = new_port->type;
+
+    return nxt_port_socket_write(task, port, NXT_PORT_MSG_NEW_PORT,
+                                 new_port->pair[1], 0, 0, b);
 }