patch 8.0.0107
Problem: When reading channel output in a timer, messages may go missing.
(Skywind)
Solution: Add the "drop" option. Write error messages in the channel log.
Don't have ch_canread() check for the channel being open.
diff --git a/src/channel.c b/src/channel.c
index e5f2800..32f3527 100644
--- a/src/channel.c
+++ b/src/channel.c
@@ -1195,6 +1195,7 @@
if (opt->jo_set & JO_CLOSE_CALLBACK)
set_callback(&channel->ch_close_cb, &channel->ch_close_partial,
opt->jo_close_cb, opt->jo_close_partial);
+ channel->ch_drop_never = opt->jo_drop_never;
if ((opt->jo_set & JO_OUT_IO) && opt->jo_io[PART_OUT] == JIO_BUFFER)
{
@@ -1918,6 +1919,7 @@
clear_tv(&listtv);
else
{
+ item->jq_no_callback = FALSE;
item->jq_value = alloc_tv();
if (item->jq_value == NULL)
{
@@ -2050,11 +2052,17 @@
* When "id" is positive it must match the first number in the list.
* When "id" is zero or negative jut get the first message. But not the one
* with id ch_block_id.
+ * When "without_callback" is TRUE also get messages that were pushed back.
* Return OK when found and return the value in "rettv".
* Return FAIL otherwise.
*/
static int
-channel_get_json(channel_T *channel, ch_part_T part, int id, typval_T **rettv)
+channel_get_json(
+ channel_T *channel,
+ ch_part_T part,
+ int id,
+ int without_callback,
+ typval_T **rettv)
{
jsonq_T *head = &channel->ch_part[part].ch_json_head;
jsonq_T *item = head->jq_next;
@@ -2064,10 +2072,11 @@
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)
+ if ((without_callback || !item->jq_no_callback)
+ && ((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 != channel->ch_part[part].ch_block_id)))
+ || tv->vval.v_number != channel->ch_part[part].ch_block_id))))
{
*rettv = item->jq_value;
if (tv->v_type == VAR_NUMBER)
@@ -2080,6 +2089,65 @@
return FAIL;
}
+/*
+ * Put back "rettv" into the JSON queue, there was no callback for it.
+ * Takes over the values in "rettv".
+ */
+ static void
+channel_push_json(channel_T *channel, ch_part_T part, typval_T *rettv)
+{
+ jsonq_T *head = &channel->ch_part[part].ch_json_head;
+ jsonq_T *item = head->jq_next;
+ jsonq_T *newitem;
+
+ if (head->jq_prev != NULL && head->jq_prev->jq_no_callback)
+ /* last item was pushed back, append to the end */
+ item = NULL;
+ else while (item != NULL && item->jq_no_callback)
+ /* append after the last item that was pushed back */
+ item = item->jq_next;
+
+ newitem = (jsonq_T *)alloc((unsigned)sizeof(jsonq_T));
+ if (newitem == NULL)
+ clear_tv(rettv);
+ else
+ {
+ newitem->jq_value = alloc_tv();
+ if (newitem->jq_value == NULL)
+ {
+ vim_free(newitem);
+ clear_tv(rettv);
+ }
+ else
+ {
+ newitem->jq_no_callback = FALSE;
+ *newitem->jq_value = *rettv;
+ if (item == NULL)
+ {
+ /* append to the end */
+ newitem->jq_prev = head->jq_prev;
+ head->jq_prev = newitem;
+ newitem->jq_next = NULL;
+ if (newitem->jq_prev == NULL)
+ head->jq_next = newitem;
+ else
+ newitem->jq_prev->jq_next = newitem;
+ }
+ else
+ {
+ /* append after "item" */
+ newitem->jq_prev = item;
+ newitem->jq_next = item->jq_next;
+ item->jq_next = newitem;
+ if (newitem->jq_next == NULL)
+ head->jq_prev = newitem;
+ else
+ newitem->jq_next->jq_prev = newitem;
+ }
+ }
+ }
+}
+
#define CH_JSON_MAX_ARGS 4
/*
@@ -2410,11 +2478,11 @@
int argc = 0;
/* Get any json message in the queue. */
- if (channel_get_json(channel, part, -1, &listtv) == FAIL)
+ if (channel_get_json(channel, part, -1, FALSE, &listtv) == FAIL)
{
/* Parse readahead, return when there is still no message. */
channel_parse_json(channel, part);
- if (channel_get_json(channel, part, -1, &listtv) == FAIL)
+ if (channel_get_json(channel, part, -1, FALSE, &listtv) == FAIL)
return FALSE;
}
@@ -2454,7 +2522,7 @@
{
/* If there is a close callback it may use ch_read() to get the
* messages. */
- if (channel->ch_close_cb == NULL)
+ if (channel->ch_close_cb == NULL && !channel->ch_drop_never)
drop_messages(channel, part);
return FALSE;
}
@@ -2531,7 +2599,7 @@
{
int done = FALSE;
- /* invoke the one-time callback with the matching nr */
+ /* JSON or JS mode: invoke the one-time callback with the matching nr */
for (cbitem = cbhead->cq_next; cbitem != NULL; cbitem = cbitem->cq_next)
if (cbitem->cq_seq_nr == seq_nr)
{
@@ -2540,7 +2608,17 @@
break;
}
if (!done)
- ch_logn(channel, "Dropping message %d without callback", seq_nr);
+ {
+ if (channel->ch_drop_never)
+ {
+ /* message must be read with ch_read() */
+ channel_push_json(channel, part, listtv);
+ listtv = NULL;
+ }
+ else
+ ch_logn(channel, "Dropping message %d without callback",
+ seq_nr);
+ }
}
else if (callback != NULL || buffer != NULL)
{
@@ -2567,7 +2645,7 @@
}
}
else
- ch_log(channel, "Dropping message");
+ ch_logn(channel, "Dropping message %d", seq_nr);
if (listtv != NULL)
free_tv(listtv);
@@ -2792,9 +2870,10 @@
redraw_after_callback();
}
- /* any remaining messages are useless now */
- for (part = PART_SOCK; part < PART_IN; ++part)
- drop_messages(channel, part);
+ if (!channel->ch_drop_never)
+ /* any remaining messages are useless now */
+ for (part = PART_SOCK; part < PART_IN; ++part)
+ drop_messages(channel, part);
}
channel->ch_nb_close_cb = NULL;
@@ -3091,9 +3170,9 @@
channel_close_now(channel_T *channel)
{
ch_log(channel, "Closing channel because all readable fds are closed");
- channel_close(channel, TRUE);
if (channel->ch_nb_close_cb != NULL)
(*channel->ch_nb_close_cb)();
+ channel_close(channel, TRUE);
}
/*
@@ -3243,7 +3322,7 @@
* When "id" is -1 accept any message;
* Blocks until the message is received or the timeout is reached.
*/
- int
+ static int
channel_read_json_block(
channel_T *channel,
ch_part_T part,
@@ -3264,7 +3343,7 @@
more = channel_parse_json(channel, part);
/* search for message "id" */
- if (channel_get_json(channel, part, id, rettv) == OK)
+ if (channel_get_json(channel, part, id, TRUE, rettv) == OK)
{
chanpart->ch_block_id = 0;
return OK;
@@ -4290,6 +4369,20 @@
return FAIL;
}
}
+ else if (STRCMP(hi->hi_key, "drop") == 0)
+ {
+ int never = FALSE;
+ val = get_tv_string(item);
+
+ if (STRCMP(val, "never") == 0)
+ never = TRUE;
+ else if (STRCMP(val, "auto") != 0)
+ {
+ EMSG2(_(e_invarg2), "drop");
+ return FAIL;
+ }
+ opt->jo_drop_never = never;
+ }
else if (STRCMP(hi->hi_key, "exit_cb") == 0)
{
if (!(supported & JO_EXIT_CB))
diff --git a/src/evalfunc.c b/src/evalfunc.c
index 88e4852..6721173 100644
--- a/src/evalfunc.c
+++ b/src/evalfunc.c
@@ -1786,7 +1786,7 @@
static void
f_ch_canread(typval_T *argvars, typval_T *rettv)
{
- channel_T *channel = get_channel_arg(&argvars[0], TRUE, TRUE, 0);
+ channel_T *channel = get_channel_arg(&argvars[0], FALSE, FALSE, 0);
rettv->vval.v_number = 0;
if (channel != NULL)
diff --git a/src/message.c b/src/message.c
index 32360a9..7dda6fa 100644
--- a/src/message.c
+++ b/src/message.c
@@ -42,6 +42,9 @@
static char_u *confirm_msg = NULL; /* ":confirm" message */
static char_u *confirm_msg_tail; /* tail of confirm_msg */
#endif
+#ifdef FEAT_JOB_CHANNEL
+static int emsg_to_channel_log = FALSE;
+#endif
struct msg_hist
{
@@ -166,6 +169,14 @@
&& STRCMP(s, last_msg_hist->msg)))
add_msg_hist(s, -1, attr);
+#ifdef FEAT_JOB_CHANNEL
+ if (emsg_to_channel_log)
+ {
+ /* Write message in the channel log. */
+ ch_logs(NULL, "ERROR: %s", (char *)s);
+ }
+#endif
+
/* When displaying keep_msg, don't let msg_start() free it, caller must do
* that. */
if (s == keep_msg)
@@ -556,6 +567,7 @@
{
int attr;
char_u *p;
+ int r;
#ifdef FEAT_EVAL
int ignore = FALSE;
int severe;
@@ -624,6 +636,9 @@
}
redir_write(s, -1);
}
+#ifdef FEAT_JOB_CHANNEL
+ ch_logs(NULL, "ERROR: %s", (char *)s);
+#endif
return TRUE;
}
@@ -650,6 +665,9 @@
* and a redraw is expected because
* msg_scrolled is non-zero */
+#ifdef FEAT_JOB_CHANNEL
+ emsg_to_channel_log = TRUE;
+#endif
/*
* Display name and line number for the source of the error.
*/
@@ -659,7 +677,12 @@
* Display the error message itself.
*/
msg_nowait = FALSE; /* wait for this msg */
- return msg_attr(s, attr);
+ r = msg_attr(s, attr);
+
+#ifdef FEAT_JOB_CHANNEL
+ emsg_to_channel_log = FALSE;
+#endif
+ return r;
}
diff --git a/src/proto/channel.pro b/src/proto/channel.pro
index f127268..a54d387 100644
--- a/src/proto/channel.pro
+++ b/src/proto/channel.pro
@@ -33,7 +33,6 @@
void channel_clear(channel_T *channel);
void channel_free_all(void);
char_u *channel_read_block(channel_T *channel, ch_part_T part, int timeout);
-int channel_read_json_block(channel_T *channel, ch_part_T part, int timeout_arg, int id, typval_T **rettv);
void common_channel_read(typval_T *argvars, typval_T *rettv, int raw);
channel_T *channel_fd2channel(sock_T fd, ch_part_T *partp);
void channel_handle_events(void);
diff --git a/src/structs.h b/src/structs.h
index 1b73e37..3fdfb5f 100644
--- a/src/structs.h
+++ b/src/structs.h
@@ -1474,6 +1474,7 @@
typval_T *jq_value;
jsonq_T *jq_next;
jsonq_T *jq_prev;
+ int jq_no_callback; /* TRUE when no callback was found */
};
struct cbq_S
@@ -1597,6 +1598,7 @@
partial_T *ch_partial;
char_u *ch_close_cb; /* call when channel is closed */
partial_T *ch_close_partial;
+ int ch_drop_never;
job_T *ch_job; /* Job that uses this channel; this does not
* count as a reference to avoid a circular
@@ -1684,6 +1686,7 @@
partial_T *jo_close_partial; /* not referenced! */
char_u *jo_exit_cb; /* not allocated! */
partial_T *jo_exit_partial; /* not referenced! */
+ int jo_drop_never;
int jo_waittime;
int jo_timeout;
int jo_out_timeout;
diff --git a/src/version.c b/src/version.c
index 152d4d8..c683ed5 100644
--- a/src/version.c
+++ b/src/version.c
@@ -765,6 +765,8 @@
static int included_patches[] =
{ /* Add new patch number below this line */
/**/
+ 107,
+/**/
106,
/**/
105,