vim: src/channel.c comparison

comparison src/channel.c @ 8151:aa845d10c6fb v7.4.1369

commit https://github.com/vim/vim/commit/42d38a2db17e70312d073095257555c27a5f9443 Author: Bram Moolenaar <Bram@vim.org> Date: Sat Feb 20 18:18:59 2016 +0100 patch 7.4.1369 Problem: Channels don't have a queue for stderr. Solution: Have a queue for each part of the channel.

author	Christian Brabandt <cb@256bit.org>
date	Sat, 20 Feb 2016 18:30:04 +0100
parents	8abef552ece7
children	6ee6fb27dcea

comparison

equal deleted inserted replaced

-:85b476dac933
+:aa845d10c6fb
 * Returns NULL if out of memory.
 */
 channel_T *
 add_channel(void)
 {
-int		which;
+int		part;
 channel_T	*channel = (channel_T *)alloc_clear((int)sizeof(channel_T));
 if (channel == NULL)
 	return NULL;
 channel->ch_id = next_ch_id++;
 ch_log(channel, "Created channel");
 #ifdef CHANNEL_PIPES
-for (which = CHAN_SOCK; which <= CHAN_IN; ++which)
+for (part = PART_SOCK; part <= PART_IN; ++part)
 #else
-which = CHAN_SOCK;
+part = PART_SOCK;
 #endif
 {
-	channel->ch_pfd[which].ch_fd = CHAN_FD_INVALID;
+	channel->ch_part[part].ch_fd = INVALID_FD;
 #ifdef FEAT_GUI_X11
-	channel->ch_pfd[which].ch_inputHandler = (XtInputId)NULL;
+	channel->ch_part[part].ch_inputHandler = (XtInputId)NULL;
 #endif
 #ifdef FEAT_GUI_GTK
-	channel->ch_pfd[which].ch_inputHandler = 0;
+	channel->ch_part[part].ch_inputHandler = 0;
 #endif
 #ifdef FEAT_GUI_W32
-	channel->ch_pfd[which].ch_inputHandler = -1;
+	channel->ch_part[part].ch_inputHandler = -1;
 #endif
-}
+	channel->ch_part[part].ch_timeout = 2000;
+}
-channel->ch_timeout = 2000;
 if (first_channel != NULL)
 {
 	first_channel->ch_prev = channel;
 	channel->ch_next = first_channel;
 #if defined(FEAT_GUI_X11) || defined(FEAT_GUI_GTK)
 static void
 channel_read_netbeans(int id)
 {
-channel_T *channel = channel_from_id(id);
+channel_T	*channel = channel_from_id(id);
+int		part;
 if (channel == NULL)
 	ch_errorn(NULL, "Channel %d not found", id);
 else
-	channel_read(channel, -1, "messageFromNetbeans");
+{
+	/* TODO: check stderr */
+	if (channel->CH_SOCK_FD != INVALID_FD)
+	    part = PART_SOCK;
+	else
+	    part = PART_OUT;
+	channel_read(channel, part, "messageFromNetbeans");
+}
 }
 #endif
 /*
 * Read a command from netbeans.
+* TODO: instead of channel ID use the FD.
 */
 #ifdef FEAT_GUI_X11
 static void
 messageFromNetbeans(XtPointer clientData,
 		    int *unused1 UNUSED,
 channel_read_netbeans((int)(long)clientData);
 }
 #endif
 static void
-channel_gui_register_one(channel_T *channel, int which)
+channel_gui_register_one(channel_T *channel, int part)
 {
 # ifdef FEAT_GUI_X11
 /* Tell notifier we are interested in being called
 * when there is input on the editor connection socket. */
-if (channel->ch_pfd[which].ch_inputHandler == (XtInputId)NULL)
+if (channel->ch_part[part].ch_inputHandler == (XtInputId)NULL)
-	channel->ch_pfd[which].ch_inputHandler = XtAppAddInput(
+	channel->ch_part[part].ch_inputHandler = XtAppAddInput(
 		(XtAppContext)app_context,
-		channel->ch_pfd[which].ch_fd,
+		channel->ch_part[part].ch_fd,
 		(XtPointer)(XtInputReadMask + XtInputExceptMask),
 		messageFromNetbeans,
 		(XtPointer)(long)channel->ch_id);
 # else
 #  ifdef FEAT_GUI_GTK
 /* Tell gdk we are interested in being called when there
 * is input on the editor connection socket. */
-if (channel->ch_pfd[which].ch_inputHandler == 0)
+if (channel->ch_part[part].ch_inputHandler == 0)
-	channel->ch_pfd[which].ch_inputHandler = gdk_input_add(
+	channel->ch_part[part].ch_inputHandler = gdk_input_add(
-		(gint)channel->ch_pfd[which].ch_fd,
+		(gint)channel->ch_part[part].ch_fd,
 		(GdkInputCondition)
 			     ((int)GDK_INPUT_READ + (int)GDK_INPUT_EXCEPTION),
 		messageFromNetbeans,
 		(gpointer)(long)channel->ch_id);
 #  else
 #   ifdef FEAT_GUI_W32
 /* Tell Windows we are interested in receiving message when there
 * is input on the editor connection socket.  */
-if (channel->ch_pfd[which].ch_inputHandler == -1)
+if (channel->ch_part[part].ch_inputHandler == -1)
-	channel->ch_pfd[which].ch_inputHandler = WSAAsyncSelect(
+	channel->ch_part[part].ch_inputHandler = WSAAsyncSelect(
-		channel->ch_pfd[which].ch_fd,
+		channel->ch_part[part].ch_fd,
 		s_hwnd, WM_NETBEANS, FD_READ);
 #   endif
 #  endif
 # endif
 }
 channel_gui_register(channel_T *channel)
 {
 if (!CH_HAS_GUI)
 	return;
-if (channel->CH_SOCK != CHAN_FD_INVALID)
+if (channel->CH_SOCK_FD != INVALID_FD)
-	channel_gui_register_one(channel, CHAN_SOCK);
+	channel_gui_register_one(channel, PART_SOCK);
 # ifdef CHANNEL_PIPES
-if (channel->CH_OUT != CHAN_FD_INVALID)
+if (channel->CH_OUT_FD != INVALID_FD)
-	channel_gui_register_one(channel, CHAN_OUT);
+	channel_gui_register_one(channel, PART_OUT);
-if (channel->CH_ERR != CHAN_FD_INVALID)
+if (channel->CH_ERR_FD != INVALID_FD)
-	channel_gui_register_one(channel, CHAN_ERR);
+	channel_gui_register_one(channel, PART_ERR);
 # endif
 }
 /*
 * Register any of our file descriptors with the GUI event handling system.
 }
 static void
 channel_gui_unregister(channel_T *channel)
 {
-int	    which;
+int	    part;
 #ifdef CHANNEL_PIPES
-for (which = CHAN_SOCK; which < CHAN_IN; ++which)
+for (part = PART_SOCK; part < PART_IN; ++part)
 #else
-which = CHAN_SOCK;
+part = PART_SOCK;
 #endif
 {
 # ifdef FEAT_GUI_X11
-	if (channel->ch_pfd[which].ch_inputHandler != (XtInputId)NULL)
+	if (channel->ch_part[part].ch_inputHandler != (XtInputId)NULL)
 	{
-	    XtRemoveInput(channel->ch_pfd[which].ch_inputHandler);
+	    XtRemoveInput(channel->ch_part[part].ch_inputHandler);
-	    channel->ch_pfd[which].ch_inputHandler = (XtInputId)NULL;
+	    channel->ch_part[part].ch_inputHandler = (XtInputId)NULL;
 	}
 # else
 #  ifdef FEAT_GUI_GTK
-	if (channel->ch_pfd[which].ch_inputHandler != 0)
+	if (channel->ch_part[part].ch_inputHandler != 0)
 	{
-	    gdk_input_remove(channel->ch_pfd[which].ch_inputHandler);
+	    gdk_input_remove(channel->ch_part[part].ch_inputHandler);
-	    channel->ch_pfd[which].ch_inputHandler = 0;
+	    channel->ch_part[part].ch_inputHandler = 0;
 	}
 #  else
 #   ifdef FEAT_GUI_W32
-	if (channel->ch_pfd[which].ch_inputHandler == 0)
+	if (channel->ch_part[part].ch_inputHandler == 0)
 	{
-	    WSAAsyncSelect(channel->ch_pfd[which].ch_fd, s_hwnd, 0, 0);
+	    WSAAsyncSelect(channel->ch_part[part].ch_fd, s_hwnd, 0, 0);
-	    channel->ch_pfd[which].ch_inputHandler = -1;
+	    channel->ch_part[part].ch_inputHandler = -1;
 	}
 #   endif
 #  endif
 # endif
 }
 #else
 	(void)fcntl(sd, F_SETFL, 0);
 #endif
 }
-channel->CH_SOCK = (sock_T)sd;
+channel->CH_SOCK_FD = (sock_T)sd;
 channel->ch_close_cb = close_cb;
 #ifdef FEAT_GUI
 channel_gui_register(channel);
 #endif
 #if defined(CHANNEL_PIPES) || defined(PROTO)
 void
 channel_set_pipes(channel_T *channel, sock_T in, sock_T out, sock_T err)
 {
-channel->CH_IN = in;
+channel->CH_IN_FD = in;
-channel->CH_OUT = out;
+channel->CH_OUT_FD = out;
-channel->CH_ERR = err;
+channel->CH_ERR_FD = err;
 }
 #endif
 void
 channel_set_job(channel_T *channel, job_T *job)
 * Set various properties from an "options" argument.
 */
 void
 channel_set_options(channel_T *channel, jobopt_T *options)
 {
+int part;
 if (options->jo_set & JO_MODE)
-	channel->ch_mode = options->jo_mode;
+	for (part = PART_SOCK; part <= PART_IN; ++part)
+	    channel->ch_part[part].ch_mode = options->jo_mode;
 if (options->jo_set & JO_TIMEOUT)
-	channel->ch_timeout = options->jo_timeout;
+	for (part = PART_SOCK; part <= PART_IN; ++part)
+	    channel->ch_part[part].ch_timeout = options->jo_timeout;
 if (options->jo_set & JO_CALLBACK)
 {
 	vim_free(channel->ch_callback);
 	if (options->jo_callback != NULL && *options->jo_callback != NUL)
 	    channel->ch_callback = NULL;
 }
 }
 /*
-* Set the callback for channel "channel" for the response with "id".
+* Set the callback for "channel"/"part" for the response with "id".
 */
 void
-channel_set_req_callback(channel_T *channel, char_u *callback, int id)
+channel_set_req_callback(
-{
+	channel_T *channel,
-cbq_T *head = &channel->ch_cb_head;
+	int part,
+	char_u *callback,
+	int id)
+{
+cbq_T *head = &channel->ch_part[part].ch_cb_head;
 cbq_T *item = (cbq_T *)alloc((int)sizeof(cbq_T));
 if (item != NULL)
 {
 	item->cq_callback = vim_strsave(callback);
 cursor_on();
 out_flush();
 }
 /*
-* Return the first buffer from the channel and remove it.
+* Return the first buffer from channel "channel"/"part" and remove it.
 * The caller must free it.
 * Returns NULL if there is nothing.
 */
 char_u *
-channel_get(channel_T *channel)
+channel_get(channel_T *channel, int part)
 {
-readq_T *head = &channel->ch_head;
+readq_T *head = &channel->ch_part[part].ch_head;
 readq_T *node = head->rq_next;
 char_u *p;
 if (node == NULL)
 	return NULL;
 vim_free(node);
 return p;
 }
 /*
-* Returns the whole buffer contents concatenated.
+* Returns the whole buffer contents concatenated for "channel"/"part".
 */
 static char_u *
-channel_get_all(channel_T *channel)
+channel_get_all(channel_T *channel, int part)
 {
 /* Concatenate everything into one buffer.
 * TODO: avoid multiple allocations. */
-while (channel_collapse(channel) == OK)
+while (channel_collapse(channel, part) == OK)
 	;
-return channel_get(channel);
+return channel_get(channel, part);
 }
 /*
-* Collapses the first and second buffer in the channel "channel".
+* Collapses the first and second buffer for "channel"/"part".
 * Returns FAIL if that is not possible.
 */
 int
-channel_collapse(channel_T *channel)
+channel_collapse(channel_T *channel, int part)
 {
-readq_T *head = &channel->ch_head;
+readq_T *head = &channel->ch_part[part].ch_head;
 readq_T *node = head->rq_next;
 char_u  *p;
 if (node == NULL || node->rq_next == NULL)
 	return FAIL;
 vim_free(node);
 return OK;
 }
 /*
-* Use the read buffer of channel "channel" and parse a JSON messages that is
+* Store "buf[len]" on "channel"/"part".
+* Returns OK or FAIL.
+*/
+static int
+channel_save(channel_T *channel, int part, char_u *buf, int len)
+{
+readq_T *node;
+readq_T *head = &channel->ch_part[part].ch_head;
+char_u  *p;
+int	    i;
+node = (readq_T *)alloc(sizeof(readq_T));
+if (node == NULL)
+	return FAIL;	    /* out of memory */
+node->rq_buffer = alloc(len + 1);
+if (node->rq_buffer == NULL)
+{
+	vim_free(node);
+	return FAIL;	    /* out of memory */
+}
+if (channel->ch_part[part].ch_mode == MODE_NL)
+{
+	/* Drop any CR before a NL. */
+	p = node->rq_buffer;
+	for (i = 0; i < len; ++i)
+	    if (buf[i] != CAR || i + 1 >= len || buf[i + 1] != NL)
+		*p++ = buf[i];
+	*p = NUL;
+}
+else
+{
+	mch_memmove(node->rq_buffer, buf, len);
+	node->rq_buffer[len] = NUL;
+}
+/* append node to the tail of the queue */
+node->rq_next = NULL;
+node->rq_prev = head->rq_prev;
+if (head->rq_prev == NULL)
+	head->rq_next = node;
+else
+	head->rq_prev->rq_next = node;
+head->rq_prev = node;
+if (log_fd != NULL)
+{
+	ch_log_lead("RECV ", channel);
+	fprintf(log_fd, "'");
+	if (fwrite(buf, len, 1, log_fd) != 1)
+	    return FAIL;
+	fprintf(log_fd, "'\n");
+}
+return OK;
+}
+/*
+* Use the read buffer of "channel"/"part" and parse a JSON messages that is
 * complete.  The messages are added to the queue.
 * Return TRUE if there is more to read.
 */
 static int
-channel_parse_json(channel_T *channel)
+channel_parse_json(channel_T *channel, int part)
 {
 js_read_T	reader;
 typval_T	listtv;
 jsonq_T	*item;
-jsonq_T	*head = &channel->ch_json_head;
+jsonq_T	*head = &channel->ch_part[part].ch_json_head;
 int		ret;
-if (channel_peek(channel) == NULL)
+if (channel_peek(channel, part) == NULL)
 	return FALSE;
 /* TODO: make reader work properly */
-/* reader.js_buf = channel_peek(channel); */
+/* reader.js_buf = channel_peek(channel, part); */
-reader.js_buf = channel_get_all(channel);
+reader.js_buf = channel_get_all(channel, part);
 reader.js_used = 0;
 reader.js_fill = NULL;
 /* reader.js_fill = channel_fill; */
 reader.js_cookie = channel;
 ret = json_decode(&reader, &listtv,
-				   channel->ch_mode == MODE_JS ? JSON_JS : 0);
+		     channel->ch_part[part].ch_mode == MODE_JS ? JSON_JS : 0);
 if (ret == OK)
 {
 	/* Only accept the response when it is a list with at least two
 	 * items. */
 	if (listtv.v_type != VAR_LIST || listtv.vval.v_list->lv_len < 2)
 /* Put the unread part back into the channel.
 * TODO: insert in front */
 if (reader.js_buf[reader.js_used] != NUL)
 {
-	channel_save(channel, reader.js_buf + reader.js_used,
+	channel_save(channel, part, reader.js_buf + reader.js_used,
 		(int)(reader.js_end - reader.js_buf) - reader.js_used);
 	ret = TRUE;
 }
 else
 	ret = FALSE;
 * with id ch_block_id.
 * Return OK when found and return the value in "rettv".
 * Return FAIL otherwise.
 */
 static int
-channel_get_json(channel_T *channel, int id, typval_T **rettv)
+channel_get_json(channel_T *channel, int part, int id, typval_T **rettv)
 {
-jsonq_T   *head = &channel->ch_json_head;
+jsonq_T   *head = &channel->ch_part[part].ch_json_head;
 jsonq_T   *item = head->jq_next;
 while (item != NULL)
 {
 	list_T	    *l = item->jq_value->vval.v_list;
 	typval_T    *tv = &l->lv_first->li_tv;
 	if ((id > 0 && tv->v_type == VAR_NUMBER && tv->vval.v_number == id)
 	      || (id <= 0 && (tv->v_type != VAR_NUMBER
-			       || tv->vval.v_number == 0
+		 || tv->vval.v_number == 0
-			       || tv->vval.v_number != channel->ch_block_id)))
+		 || tv->vval.v_number != channel->ch_part[part].ch_block_id)))
 	{
 	    *rettv = item->jq_value;
 	    remove_json_node(head, item);
 	    return OK;
 	}
 }
 return FAIL;
 }
 /*
-* Execute a command received over channel "channel".
+* Execute a command received over "channel"/"part"
 * "cmd" is the command string, "arg2" the second argument.
 * "arg3" is the third argument, NULL if missing.
 */
 static void
-channel_exe_cmd(channel_T *channel, char_u *cmd, typval_T *arg2, typval_T *arg3)
+channel_exe_cmd(
+	channel_T *channel,
+	int part,
+	char_u *cmd,
+	typval_T *arg2,
+	typval_T *arg3)
 {
 char_u *arg;
 if (arg2->v_type != VAR_STRING)
 {
 	else
 	{
 	    typval_T	*tv;
 	    typval_T	err_tv;
 	    char_u	*json = NULL;
-	    int		options = channel->ch_mode == MODE_JS ? JSON_JS : 0;
+	    int		options = channel->ch_part[part].ch_mode == MODE_JS
+								? JSON_JS : 0;
 	    /* Don't pollute the display with errors. */
 	    ++emsg_skip;
 	    tv = eval_expr(arg, NULL);
 	    if (is_eval)
 		    json = json_encode_nr_expr(arg3->vval.v_number, tv,
 								     options);
 		}
 		if (json != NULL)
 		{
-		    channel_send(channel, json, "eval");
+		    channel_send(channel, part, json, "eval");
 		    vim_free(json);
 		}
 	    }
 	    --emsg_skip;
 	    if (tv != &err_tv)
 else if (p_verbose > 2)
 	EMSG2("E905: received unknown command: %s", cmd);
 }
 /*
-* Invoke a callback for channel "channel" if needed.
+* Invoke a callback for "channel"/"part" if needed.
-* TODO: add "which" argument, read stderr.
 * Return TRUE when a message was handled, there might be another one.
 */
 static int
-may_invoke_callback(channel_T *channel)
+may_invoke_callback(channel_T *channel, int part)
 {
 char_u	*msg = NULL;
 typval_T	*listtv = NULL;
 list_T	*list;
 typval_T	*typetv;
 typval_T	argv[3];
 int		seq_nr = -1;
-ch_mode_T	ch_mode = channel->ch_mode;
+ch_mode_T	ch_mode = channel->ch_part[part].ch_mode;
+char_u	*callback = NULL;
 if (channel->ch_close_cb != NULL)
 	/* this channel is handled elsewhere (netbeans) */
 	return FALSE;
+if (channel->ch_part[part].ch_callback != NULL)
+	callback = channel->ch_part[part].ch_callback;
+else
+	callback = channel->ch_callback;
 if (ch_mode == MODE_JSON || ch_mode == MODE_JS)
 {
 	/* Get any json message in the queue. */
-	if (channel_get_json(channel, -1, &listtv) == FAIL)
+	if (channel_get_json(channel, part, -1, &listtv) == FAIL)
 	{
 	    /* Parse readahead, return when there is still no message. */
-	    channel_parse_json(channel);
+	    channel_parse_json(channel, part);
-	    if (channel_get_json(channel, -1, &listtv) == FAIL)
+	    if (channel_get_json(channel, part, -1, &listtv) == FAIL)
 		return FALSE;
 	}
 	list = listtv->vval.v_list;
 	argv[1] = list->lv_first->li_next->li_tv;
 	    /* ["cmd", arg] or ["cmd", arg, arg] */
 	    if (list->lv_len == 3)
 		arg3 = &list->lv_last->li_tv;
 	    ch_logs(channel, "Executing %s command", (char *)cmd);
-	    channel_exe_cmd(channel, cmd, &argv[1], arg3);
+	    channel_exe_cmd(channel, part, cmd, &argv[1], arg3);
 	    free_tv(listtv);
 	    return TRUE;
 	}
 	if (typetv->v_type != VAR_NUMBER)
 	    free_tv(listtv);
 	    return FALSE;
 	}
 	seq_nr = typetv->vval.v_number;
 }
-else if (channel_peek(channel) == NULL)
+else if (channel_peek(channel, part) == NULL)
 {
 	/* nothing to read on RAW or NL channel */
 	return FALSE;
 }
 else
 {
 	/* If there is no callback drop the message. */
-	if (channel->ch_callback == NULL)
+	if (callback == NULL)
 	{
-	    while ((msg = channel_get(channel)) != NULL)
+	    while ((msg = channel_get(channel, part)) != NULL)
 		vim_free(msg);
 	    return FALSE;
 	}
 	if (ch_mode == MODE_NL)
 	    /* See if we have a message ending in NL in the first buffer.  If
 	     * not try to concatenate the first and the second buffer. */
 	    while (TRUE)
 	    {
-		buf = channel_peek(channel);
+		buf = channel_peek(channel, part);
 		nl = vim_strchr(buf, NL);
 		if (nl != NULL)
 		    break;
-		if (channel_collapse(channel) == FAIL)
+		if (channel_collapse(channel, part) == FAIL)
 		    return FALSE; /* incomplete message */
 	    }
 	    if (nl[1] == NUL)
 		/* get the whole buffer */
-		msg = channel_get(channel);
+		msg = channel_get(channel, part);
 	    else
 	    {
 		/* Copy the message into allocated memory and remove it from
 		 * the buffer. */
 		msg = vim_strnsave(buf, (int)(nl - buf));
 	    }
 	}
 	else
 	    /* For a raw channel we don't know where the message ends, just
 	     * get everything we have. */
-	    msg = channel_get_all(channel);
+	    msg = channel_get_all(channel, part);
 	argv[1].v_type = VAR_STRING;
 	argv[1].vval.v_string = msg;
 }
 if (seq_nr > 0)
 {
-	cbq_T	*head = &channel->ch_cb_head;
+	cbq_T	*head = &channel->ch_part[part].ch_cb_head;
 	cbq_T	*item = head->cq_next;
 	int	done = FALSE;
 	/* invoke the one-time callback with the matching nr */
 	while (item != NULL)
 	    item = item->cq_next;
 	}
 	if (!done)
 	    ch_log(channel, "Dropping message without callback");
 }
-else if (channel->ch_callback != NULL)
+else if (callback != NULL)
 {
 	/* invoke the channel callback */
 	ch_log(channel, "Invoking channel callback");
-	invoke_callback(channel, channel->ch_callback, argv);
+	invoke_callback(channel, callback, argv);
 }
 else
 	ch_log(channel, "Dropping message");
 if (listtv != NULL)
 * Also returns FALSE or invalid "channel".
 */
 int
 channel_can_write_to(channel_T *channel)
 {
-return channel != NULL && (channel->CH_SOCK != CHAN_FD_INVALID
+return channel != NULL && (channel->CH_SOCK_FD != INVALID_FD
 #ifdef CHANNEL_PIPES
-			  || channel->CH_IN != CHAN_FD_INVALID
+			  || channel->CH_IN_FD != INVALID_FD
 #endif
 			  );
 }
 /*
 * Also returns FALSE for invalid "channel".
 */
 int
 channel_is_open(channel_T *channel)
 {
-return channel != NULL && (channel->CH_SOCK != CHAN_FD_INVALID
+return channel != NULL && (channel->CH_SOCK_FD != INVALID_FD
 #ifdef CHANNEL_PIPES
-			  || channel->CH_IN != CHAN_FD_INVALID
+			  || channel->CH_IN_FD != INVALID_FD
-			  || channel->CH_OUT != CHAN_FD_INVALID
+			  || channel->CH_OUT_FD != INVALID_FD
-			  || channel->CH_ERR != CHAN_FD_INVALID
+			  || channel->CH_ERR_FD != INVALID_FD
 #endif
 			  );
 }
 /*
 #ifdef FEAT_GUI
 channel_gui_unregister(channel);
 #endif
-if (channel->CH_SOCK != CHAN_FD_INVALID)
+if (channel->CH_SOCK_FD != INVALID_FD)
 {
-	sock_close(channel->CH_SOCK);
+	sock_close(channel->CH_SOCK_FD);
-	channel->CH_SOCK = CHAN_FD_INVALID;
+	channel->CH_SOCK_FD = INVALID_FD;
 }
 #if defined(CHANNEL_PIPES)
-if (channel->CH_IN != CHAN_FD_INVALID)
+if (channel->CH_IN_FD != INVALID_FD)
 {
-	fd_close(channel->CH_IN);
+	fd_close(channel->CH_IN_FD);
-	channel->CH_IN = CHAN_FD_INVALID;
+	channel->CH_IN_FD = INVALID_FD;
 }
-if (channel->CH_OUT != CHAN_FD_INVALID)
+if (channel->CH_OUT_FD != INVALID_FD)
 {
-	fd_close(channel->CH_OUT);
+	fd_close(channel->CH_OUT_FD);
-	channel->CH_OUT = CHAN_FD_INVALID;
+	channel->CH_OUT_FD = INVALID_FD;
 }
-if (channel->CH_ERR != CHAN_FD_INVALID)
+if (channel->CH_ERR_FD != INVALID_FD)
 {
-	fd_close(channel->CH_ERR);
+	fd_close(channel->CH_ERR_FD);
-	channel->CH_ERR = CHAN_FD_INVALID;
+	channel->CH_ERR_FD = INVALID_FD;
 }
 #endif
 channel->ch_close_cb = NULL;
 channel_clear(channel);
 }
 /*
-* Store "buf[len]" on channel "channel".
+* Return the first buffer from "channel"/"part" without removing it.
-* Returns OK or FAIL.
-*/
-int
-channel_save(channel_T *channel, char_u *buf, int len)
-{
-readq_T *node;
-readq_T *head = &channel->ch_head;
-char_u  *p;
-int	    i;
-node = (readq_T *)alloc(sizeof(readq_T));
-if (node == NULL)
-	return FAIL;	    /* out of memory */
-node->rq_buffer = alloc(len + 1);
-if (node->rq_buffer == NULL)
-{
-	vim_free(node);
-	return FAIL;	    /* out of memory */
-}
-if (channel->ch_mode == MODE_NL)
-{
-	/* Drop any CR before a NL. */
-	p = node->rq_buffer;
-	for (i = 0; i < len; ++i)
-	    if (buf[i] != CAR || i + 1 >= len || buf[i + 1] != NL)
-		*p++ = buf[i];
-	*p = NUL;
-}
-else
-{
-	mch_memmove(node->rq_buffer, buf, len);
-	node->rq_buffer[len] = NUL;
-}
-/* append node to the tail of the queue */
-node->rq_next = NULL;
-node->rq_prev = head->rq_prev;
-if (head->rq_prev == NULL)
-	head->rq_next = node;
-else
-	head->rq_prev->rq_next = node;
-head->rq_prev = node;
-if (log_fd != NULL)
-{
-	ch_log_lead("RECV ", channel);
-	fprintf(log_fd, "'");
-	if (fwrite(buf, len, 1, log_fd) != 1)
-	    return FAIL;
-	fprintf(log_fd, "'\n");
-}
-return OK;
-}
-/*
-* Return the first buffer from the channel without removing it.
 * Returns NULL if there is nothing.
 */
 char_u *
-channel_peek(channel_T *channel)
+channel_peek(channel_T *channel, int part)
 {
-readq_T *head = &channel->ch_head;
+readq_T *head = &channel->ch_part[part].ch_head;
 if (head->rq_next == NULL)
 	return NULL;
 return head->rq_next->rq_buffer;
 }
 /*
-* Clear the read buffer on channel "channel".
+* Clear the read buffer on "channel"/"part".
 */
-void
+static void
-channel_clear(channel_T *channel)
+channel_clear_one(channel_T *channel, int part)
 {
-jsonq_T *json_head = &channel->ch_json_head;
+jsonq_T *json_head = &channel->ch_part[part].ch_json_head;
-cbq_T   *cb_head = &channel->ch_cb_head;
+cbq_T   *cb_head = &channel->ch_part[part].ch_cb_head;
-while (channel_peek(channel) != NULL)
+while (channel_peek(channel, part) != NULL)
-	vim_free(channel_get(channel));
+	vim_free(channel_get(channel, part));
 while (cb_head->cq_next != NULL)
 {
 	cbq_T *node = cb_head->cq_next;
 {
 	free_tv(json_head->jq_next->jq_value);
 	remove_json_node(json_head, json_head->jq_next);
 }
+vim_free(channel->ch_part[part].ch_callback);
+channel->ch_part[part].ch_callback = NULL;
+}
+/*
+* Clear all the read buffers on "channel".
+*/
+void
+channel_clear(channel_T *channel)
+{
+channel_clear_one(channel, PART_SOCK);
+#ifdef CHANNEL_PIPES
+channel_clear_one(channel, PART_OUT);
+channel_clear_one(channel, PART_ERR);
+#endif
 vim_free(channel->ch_callback);
 channel->ch_callback = NULL;
 }
 #if defined(EXITFREE) || defined(PROTO)
 {
 if (timeout > 0)
 	ch_logn(channel, "Waiting for up to %d msec", timeout);
 # ifdef WIN32
-if (fd != channel->CH_SOCK)
+if (fd != channel->CH_SOCK_FD)
 {
 	DWORD	nread;
 	int	diff;
 	DWORD	deadline = GetTickCount() + timeout;
 return next_id++;
 }
 /*
-* Get the file descriptor to read from, either the socket or stdout.
-* TODO: should have a way to read stderr.
-*/
-static sock_T
-get_read_fd(channel_T *channel)
-{
-if (channel->CH_SOCK != CHAN_FD_INVALID)
-	return channel->CH_SOCK;
-#if defined(CHANNEL_PIPES)
-if (channel->CH_OUT != CHAN_FD_INVALID)
-	return channel->CH_OUT;
-#endif
-ch_error(channel, "channel_read() called while socket is closed");
-return CHAN_FD_INVALID;
-}
-/*
 * Read from channel "channel" for as long as there is something to read.
-* "which" is CHAN_SOCK, CHAN_OUT or CHAN_ERR.  When -1 use CHAN_SOCK or
+* "part" is PART_SOCK, PART_OUT or PART_ERR.
-* CHAN_OUT, the one that is open.
 * The data is put in the read queue.
 */
 void
-channel_read(channel_T *channel, int which, char *func)
+channel_read(channel_T *channel, int part, char *func)
 {
 static char_u	*buf = NULL;
 int			len = 0;
 int			readlen = 0;
 sock_T		fd;
 int			use_socket = FALSE;
-if (which < 0)
+fd = channel->ch_part[part].ch_fd;
-	fd = get_read_fd(channel);
+if (fd == INVALID_FD)
-else
+{
-	fd = channel->ch_pfd[which].ch_fd;
+	ch_error(channel, "channel_read() called while socket is closed");
-if (fd == CHAN_FD_INVALID)
 	return;
-use_socket = fd == channel->CH_SOCK;
+}
+use_socket = fd == channel->CH_SOCK_FD;
 /* Allocate a buffer to read into. */
 if (buf == NULL)
 {
 	buf = alloc(MAXMSGSIZE);
 	    len = fd_read(fd, (char *)buf, MAXMSGSIZE);
 	if (len <= 0)
 	    break;	/* error or nothing more to read */
 	/* Store the read message in the queue. */
-	channel_save(channel, buf, len);
+	channel_save(channel, part, buf, len);
 	readlen += len;
 	if (len < MAXMSGSIZE)
 	    break;	/* did read everything that's available */
 }
 #ifdef FEAT_GUI_W32
 	 *		-> ui_breakcheck
 	 *		    -> gui event loop or select loop
 	 *			-> channel_read()
 	 */
 	ch_errors(channel, "%s(): Cannot read", func);
-	channel_save(channel, (char_u *)DETACH_MSG, (int)STRLEN(DETACH_MSG));
+	channel_save(channel, part,
+			       (char_u *)DETACH_MSG, (int)STRLEN(DETACH_MSG));
 	/* TODO: When reading from stdout is not possible, should we try to
 	 * keep stdin and stderr open?  Probably not, assume the other side
 	 * has died. */
 	channel_close(channel);
 	gtk_main_quit();
 #endif
 }
 /*
-* Read from RAW or NL channel "channel".  Blocks until there is something to
+* Read from RAW or NL "channel"/"part".  Blocks until there is something to
 * read or the timeout expires.
-* TODO: add "which" argument and read from stderr.
 * Returns what was read in allocated memory.
 * Returns NULL in case of error or timeout.
 */
 char_u *
-channel_read_block(channel_T *channel)
+channel_read_block(channel_T *channel, int part)
 {
 char_u	*buf;
 char_u	*msg;
-ch_mode_T	mode = channel->ch_mode;
+ch_mode_T	mode = channel->ch_part[part].ch_mode;
-sock_T	fd = get_read_fd(channel);
+int		timeout = channel->ch_part[part].ch_timeout;
+sock_T	fd = channel->ch_part[part].ch_fd;
 char_u	*nl;
 ch_logsn(channel, "Blocking %s read, timeout: %d msec",
-			mode == MODE_RAW ? "RAW" : "NL", channel->ch_timeout);
+				    mode == MODE_RAW ? "RAW" : "NL", timeout);
 while (TRUE)
 {
-	buf = channel_peek(channel);
+	buf = channel_peek(channel, part);
 	if (buf != NULL && (mode == MODE_RAW
 			 || (mode == MODE_NL && vim_strchr(buf, NL) != NULL)))
 	    break;
-	if (buf != NULL && channel_collapse(channel) == OK)
+	if (buf != NULL && channel_collapse(channel, part) == OK)
 	    continue;
 	/* Wait for up to the channel timeout. */
-	if (fd == CHAN_FD_INVALID
+	if (fd == INVALID_FD
-		|| channel_wait(channel, fd, channel->ch_timeout) == FAIL)
+		|| channel_wait(channel, fd, timeout) == FAIL)
 	    return NULL;
-	channel_read(channel, -1, "channel_read_block");
+	channel_read(channel, part, "channel_read_block");
 }
 if (mode == MODE_RAW)
 {
-	msg = channel_get_all(channel);
+	msg = channel_get_all(channel, part);
 }
 else
 {
 	nl = vim_strchr(buf, NL);
 	if (nl[1] == NUL)
 	{
 	    /* get the whole buffer */
-	    msg = channel_get(channel);
+	    msg = channel_get(channel, part);
 	    *nl = NUL;
 	}
 	else
 	{
 	    /* Copy the message into allocated memory and remove it from the
 	ch_logn(channel, "Returning %d bytes", (int)STRLEN(msg));
 return msg;
 }
 /*
-* Read one JSON message with ID "id" from channel "channel" and store the
+* Read one JSON message with ID "id" from "channel"/"part" and store the
 * result in "rettv".
 * Blocks until the message is received or the timeout is reached.
 */
 int
-channel_read_json_block(channel_T *channel, int id, typval_T **rettv)
+channel_read_json_block(channel_T *channel, int part, int id, typval_T **rettv)
 {
 int		more;
 sock_T	fd;
 ch_log(channel, "Reading JSON");
-channel->ch_block_id = id;
+channel->ch_part[part].ch_block_id = id;
 for (;;)
 {
-	more = channel_parse_json(channel);
+	more = channel_parse_json(channel, part);
 	/* search for messsage "id" */
-	if (channel_get_json(channel, id, rettv) == OK)
+	if (channel_get_json(channel, part, id, rettv) == OK)
 	{
-	    channel->ch_block_id = 0;
+	    channel->ch_part[part].ch_block_id = 0;
 	    return OK;
 	}
 	if (!more)
 	{
 	     * messages may have arrived. */
 	    if (channel_parse_messages())
 		continue;
 	    /* Wait for up to the channel timeout. */
-	    fd = get_read_fd(channel);
+	    fd = channel->ch_part[part].ch_fd;
-	    if (fd == CHAN_FD_INVALID
+	    if (fd == INVALID_FD || channel_wait(channel, fd,
-		    || channel_wait(channel, fd, channel->ch_timeout) == FAIL)
+				   channel->ch_part[part].ch_timeout) == FAIL)
 		break;
-	    channel_read(channel, -1, "channel_read_json_block");
+	    channel_read(channel, part, "channel_read_json_block");
 	}
 }
-channel->ch_block_id = 0;
+channel->ch_part[part].ch_block_id = 0;
 return FAIL;
 }
 # if defined(WIN32) || defined(PROTO)
 /*
-* Lookup the channel from the socket.  Set "which" to the fd index.
+* Lookup the channel from the socket.  Set "part" to the fd index.
 * Returns NULL when the socket isn't found.
 */
 channel_T *
-channel_fd2channel(sock_T fd, int *whichp)
+channel_fd2channel(sock_T fd, int *part)
 {
 channel_T	*channel;
-int		i;
+int		part;
-if (fd != CHAN_FD_INVALID)
+if (fd != INVALID_FD)
 	for (channel = first_channel; channel != NULL;
 						   channel = channel->ch_next)
 	{
 #  ifdef CHANNEL_PIPES
-	    for (i = CHAN_SOCK; i < CHAN_IN; ++i)
+	    for (part = PART_SOCK; part < PART_IN; ++part)
 #  else
-	    i = CHAN_SOCK;
+	    part = PART_SOCK;
 #  endif
-		if (channel->ch_pfd[i].ch_fd == fd)
+		if (channel->ch_part[part].ch_fd == fd)
 		{
-		    *whichp = i;
+		    *part = part;
 		    return channel;
 		}
 	}
 return NULL;
 }
 void
 channel_handle_events(void)
 {
 channel_T	*channel;
-int		which;
+int		part;
 static int	loop = 0;
 /* Skip heavily polling */
 if (loop++ % 2)
 	return;
 for (channel = first_channel; channel != NULL; channel = channel->ch_next)
 {
 #  ifdef FEAT_GUI_W32
 	/* only check the pipes */
-	for (which = CHAN_OUT; which < CHAN_ERR; ++which)
+	for (part = PART_OUT; part <= PART_ERR; ++part)
 #  else
 #   ifdef CHANNEL_PIPES
 	/* check the socket and pipes */
-	for (which = CHAN_SOCK; which < CHAN_ERR; ++which)
+	for (part = PART_SOCK; part <= PART_ERR; ++part)
 #   else
 	/* only check the socket */
-	which = CHAN_SOCK;
+	part = PART_SOCK;
 #   endif
 #  endif
-	channel_read(channel, which, "channel_handle_events");
+	channel_read(channel, part, "channel_handle_events");
 }
 }
 # endif
 /*
-* Write "buf" (NUL terminated string) to channel "channel".
+* Write "buf" (NUL terminated string) to "channel"/"part".
 * When "fun" is not NULL an error message might be given.
 * Return FAIL or OK.
 */
 int
-channel_send(channel_T *channel, char_u *buf, char *fun)
+channel_send(channel_T *channel, int part, char_u *buf, char *fun)
 {
 int		len = (int)STRLEN(buf);
 int		res;
-sock_T	fd = CHAN_FD_INVALID;
+sock_T	fd;
-int		use_socket = FALSE;
+fd = channel->ch_part[part].ch_fd;
-if (channel->CH_SOCK != CHAN_FD_INVALID)
+if (fd == INVALID_FD)
-{
-	fd = channel->CH_SOCK;
-	use_socket = TRUE;
-}
-#if defined(CHANNEL_PIPES)
-else if (channel->CH_IN != CHAN_FD_INVALID)
-	fd = channel->CH_IN;
-#endif
-if (fd == CHAN_FD_INVALID)
 {
 	if (!channel->ch_error && fun != NULL)
 	{
 	    ch_errors(channel, "%s(): write while not connected", fun);
 	    EMSG2("E630: %s(): write while not connected", fun);
 	ignored = (int)fwrite(buf, len, 1, log_fd);
 	fprintf(log_fd, "'\n");
 	fflush(log_fd);
 }
-if (use_socket)
+if (part == PART_SOCK)
 	res = sock_write(fd, (char *)buf, len);
 else
 	res = fd_write(fd, (char *)buf, len);
 if (res != len)
 {
 channel_poll_setup(int nfd_in, void *fds_in)
 {
 int		nfd = nfd_in;
 channel_T	*channel;
 struct	pollfd *fds = fds_in;
-int		which;
+int		part;
 for (channel = first_channel; channel != NULL; channel = channel->ch_next)
 {
 #  ifdef CHANNEL_PIPES
-	for (which = CHAN_SOCK; which < CHAN_IN; ++which)
+	for (part = PART_SOCK; part < PART_IN; ++part)
 #  else
-	which = CHAN_SOCK;
+	part = PART_SOCK;
 #  endif
 	{
-	    if (channel->ch_pfd[which].ch_fd != CHAN_FD_INVALID)
+	    if (channel->ch_part[part].ch_fd != INVALID_FD)
 	    {
-		channel->ch_pfd[which].ch_poll_idx = nfd;
+		channel->ch_part[part].ch_poll_idx = nfd;
-		fds[nfd].fd = channel->ch_pfd[which].ch_fd;
+		fds[nfd].fd = channel->ch_part[part].ch_fd;
 		fds[nfd].events = POLLIN;
 		nfd++;
 	    }
 	    else
-		channel->ch_pfd[which].ch_poll_idx = -1;
+		channel->ch_part[part].ch_poll_idx = -1;
 	}
 }
 return nfd;
 }
 channel_poll_check(int ret_in, void *fds_in)
 {
 int		ret = ret_in;
 channel_T	*channel;
 struct	pollfd *fds = fds_in;
-int		which;
+int		part;
 for (channel = first_channel; channel != NULL; channel = channel->ch_next)
 {
 #  ifdef CHANNEL_PIPES
-	for (which = CHAN_SOCK; which < CH_IN; ++which)
+	for (part = PART_SOCK; part < PART_IN; ++part)
 #  else
-	which = CHAN_SOCK;
+	part = PART_SOCK;
 #  endif
 	{
-	    int idx = channel->ch_pfd[which].ch_poll_idx;
+	    int idx = channel->ch_part[part].ch_poll_idx;
 	    if (ret > 0 && idx != -1 && fds[idx].revents & POLLIN)
 	    {
-		channel_read(channel, which, "channel_poll_check");
+		channel_read(channel, part, "channel_poll_check");
 		--ret;
 	    }
 	}
 }
 channel_select_setup(int maxfd_in, void *rfds_in)
 {
 int		maxfd = maxfd_in;
 channel_T	*channel;
 fd_set	*rfds = rfds_in;
-int		which;
+int		part;
 for (channel = first_channel; channel != NULL; channel = channel->ch_next)
 {
 #  ifdef CHANNEL_PIPES
-	for (which = CHAN_SOCK; which < CHAN_IN; ++which)
+	for (part = PART_SOCK; part < PART_IN; ++part)
 #  else
-	which = CHAN_SOCK;
+	part = PART_SOCK;
 #  endif
 	{
-	    sock_T fd = channel->ch_pfd[which].ch_fd;
+	    sock_T fd = channel->ch_part[part].ch_fd;
-	    if (fd != CHAN_FD_INVALID)
+	    if (fd != INVALID_FD)
 	    {
 		FD_SET((int)fd, rfds);
 		if (maxfd < (int)fd)
 		    maxfd = (int)fd;
 	    }
 channel_select_check(int ret_in, void *rfds_in)
 {
 int		ret = ret_in;
 channel_T	*channel;
 fd_set	*rfds = rfds_in;
-int		which;
+int		part;
 for (channel = first_channel; channel != NULL; channel = channel->ch_next)
 {
 #  ifdef CHANNEL_PIPES
-	for (which = CHAN_SOCK; which < CHAN_IN; ++which)
+	for (part = PART_SOCK; part < PART_IN; ++part)
 #  else
-	which = CHAN_SOCK;
+	part = PART_SOCK;
 #  endif
 	{
-	    sock_T fd = channel->ch_pfd[which].ch_fd;
+	    sock_T fd = channel->ch_part[part].ch_fd;
-	    if (ret > 0 && fd != CHAN_FD_INVALID && FD_ISSET(fd, rfds))
+	    if (ret > 0 && fd != INVALID_FD && FD_ISSET(fd, rfds))
 	    {
-		channel_read(channel, which, "channel_select_check");
+		channel_read(channel, part, "channel_select_check");
 		--ret;
 	    }
 	}
 }
 channel_parse_messages(void)
 {
 channel_T	*channel = first_channel;
 int		ret = FALSE;
 int		r;
+int		part = PART_SOCK;
 while (channel != NULL)
 {
-	/* Increase the refcount, in case the handler causes the channel to be
+	if (channel->ch_part[part].ch_fd != INVALID_FD)
-	 * unreferenced or closed. */
+	{
-	++channel->ch_refcount;
+	    /* Increase the refcount, in case the handler causes the channel
-	r = may_invoke_callback(channel);
+	     * to be unreferenced or closed. */
-	if (channel_unref(channel))
+	    ++channel->ch_refcount;
-	    /* channel was freed, start over */
+	    r = may_invoke_callback(channel, part);
-	    channel = first_channel;
+	    if (r == OK)
+		ret = TRUE;
-	if (r == OK)
+	    if (channel_unref(channel) || r == OK)
-	{
+	    {
-	    channel = first_channel;  /* something was done, start over */
+		/* channel was freed or something was done, start over */
-	    ret = TRUE;
+		channel = first_channel;
-	}
+		part = PART_SOCK;
+		continue;
+	    }
+	}
+#ifdef CHANNEL_PIPES
+	if (part < PART_ERR)
+	    ++part;
 	else
+#endif
+	{
 	    channel = channel->ch_next;
+	    part = PART_SOCK;
+	}
 }
 return ret;
 }
 /*
 int
 set_ref_in_channel(int copyID)
 {
 int		abort = FALSE;
 channel_T	*channel;
+int		part;
 for (channel = first_channel; channel != NULL; channel = channel->ch_next)
 {
-	jsonq_T *head = &channel->ch_json_head;
+#ifdef CHANNEL_PIPES
-	jsonq_T *item = head->jq_next;
+	for (part = PART_SOCK; part < PART_IN; ++part)
+#else
-	while (item != NULL)
+	part = PART_SOCK;
-	{
+#endif
-	    list_T	*l = item->jq_value->vval.v_list;
+	{
+	    jsonq_T *head = &channel->ch_part[part].ch_json_head;
-	    if (l->lv_copyID != copyID)
+	    jsonq_T *item = head->jq_next;
+	    while (item != NULL)
 	    {
-		l->lv_copyID = copyID;
+		list_T	*l = item->jq_value->vval.v_list;
-		abort = abort || set_ref_in_list(l, copyID, NULL);
+		if (l->lv_copyID != copyID)
+		{
+		    l->lv_copyID = copyID;
+		    abort = abort || set_ref_in_list(l, copyID, NULL);
+		}
+		item = item->jq_next;
 	    }
-	    item = item->jq_next;
 	}
 }
 return abort;
 }
 /*
-* Return the mode of channel "channel".
+* Return the "part" to write to for "channel".
+*/
+int
+channel_part_send(channel_T *channel)
+{
+#ifdef CHANNEL_PIPES
+if (channel->CH_SOCK_FD == INVALID_FD)
+	return PART_IN;
+#endif
+return PART_SOCK;
+}
+/*
+* Return the default "part" to read from for "channel".
+*/
+int
+channel_part_read(channel_T *channel)
+{
+#ifdef CHANNEL_PIPES
+if (channel->CH_SOCK_FD == INVALID_FD)
+	return PART_OUT;
+#endif
+return PART_SOCK;
+}
+/*
+* Return the mode of "channel"/"part"
 * If "channel" is invalid returns MODE_JSON.
 */
 ch_mode_T
-channel_get_mode(channel_T *channel)
+channel_get_mode(channel_T *channel, int part)
 {
 if (channel == NULL)
 	return MODE_JSON;
-return channel->ch_mode;
+return channel->ch_part[part].ch_mode;
 }
 #endif /* FEAT_CHANNEL */

Mercurial > vim

comparison src/channel.c @ 8151:aa845d10c6fb v7.4.1369