mirrors/neovim
1
0
Fork 0
mirror of https://github.com/neovim/neovim.git synced 2025-09-06 03:18:16 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge #33043 redundar channel_close()

Browse Source
This commit is contained in:
Justin M. Keyes
2025-03-24 10:15:35 -07:00
committed by GitHub
parent c4e9ff30a6 c49162be59
commit 76eca62137
4 changed files with 13 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
src/nvim/event/rstream.c
View File

@@ -184,6 +184,7 @@ static void read_event(void **argv)
} }
stream->s.pending_reqs--; stream->s.pending_reqs--;
if (stream->s.closed && !stream->s.pending_reqs) { if (stream->s.closed && !stream->s.pending_reqs) {
// Last pending read; free the stream.
stream_close_handle(&stream->s, true); stream_close_handle(&stream->s, true);
} }
} }

2
src/nvim/event/stream.c
View File

@@ -120,7 +120,7 @@ void stream_may_close(Stream *stream, bool rstream)
if (!stream->pending_reqs) { if (!stream->pending_reqs) {
stream_close_handle(stream, rstream); stream_close_handle(stream, rstream);
} } // Else: rstream.c:read_event() or wstream.c:write_cb() will call stream_close_handle().
} }
void stream_close_handle(Stream *stream, bool rstream) void stream_close_handle(Stream *stream, bool rstream)

2
src/nvim/event/wstream.c
View File

@@ -156,7 +156,7 @@ static void write_cb(uv_write_t *req, int status)
data->stream->pending_reqs--; data->stream->pending_reqs--;
if (data->stream->closed && data->stream->pending_reqs == 0) { if (data->stream->closed && data->stream->pending_reqs == 0) {
// Last pending write, free the stream; // Last pending write; free the stream.
stream_close_handle(data->stream, false); stream_close_handle(data->stream, false);
} }

19
src/nvim/msgpack_rpc/channel.c
View File

@@ -222,10 +222,9 @@ static size_t receive_msgpack(RStream *stream, const char *rbuf, size_t c, void
} }
if (eof) { if (eof) {
channel_close(channel->id, kChannelPartRpc, NULL);
char buf[256]; char buf[256];
snprintf(buf, sizeof(buf), "ch %" PRIu64 " was closed by the peer", channel->id); snprintf(buf, sizeof(buf), "ch %" PRIu64 " was closed by the peer", channel->id);
chan_close_with_error(channel, buf, LOGLVL_INF); chan_close_on_err(channel, buf, LOGLVL_INF);
} }
channel_decref(channel); channel_decref(channel);
@@ -268,7 +267,7 @@ static void parse_msgpack(Channel *channel)
"ch %" PRIu64 " (type=%" PRIu32 ") returned a response with an unknown request " "ch %" PRIu64 " (type=%" PRIu32 ") returned a response with an unknown request "
"id %" PRIu32 ". Ensure the client is properly synchronized", "id %" PRIu32 ". Ensure the client is properly synchronized",
channel->id, (unsigned)channel->rpc.client_type, p->request_id); channel->id, (unsigned)channel->rpc.client_type, p->request_id);
chan_close_with_error(channel, buf, LOGLVL_ERR); chan_close_on_err(channel, buf, LOGLVL_ERR);
return; return;
} }
frame->returned = true; frame->returned = true;
@@ -292,7 +291,7 @@ static void parse_msgpack(Channel *channel)
Object res = p->result; Object res = p->result;
if (p->result.type != kObjectTypeArray) { if (p->result.type != kObjectTypeArray) {
chan_close_with_error(channel, "msgpack-rpc request args must be an array", LOGLVL_ERR); chan_close_on_err(channel, "msgpack-rpc request args must be an array", LOGLVL_ERR);
return; return;
} }
Array arg = res.data.array; Array arg = res.data.array;
@@ -301,7 +300,7 @@ static void parse_msgpack(Channel *channel)
} }
if (unpacker_closed(p)) { if (unpacker_closed(p)) {
chan_close_with_error(channel, p->unpack_error.msg, LOGLVL_INF); chan_close_on_err(channel, p->unpack_error.msg, LOGLVL_INF);
api_clear_error(&p->unpack_error); api_clear_error(&p->unpack_error);
} }
} }
@@ -419,7 +418,7 @@ static bool channel_write(Channel *channel, WBuffer *buffer)
"ch %" PRIu64 ": stream write failed. " "ch %" PRIu64 ": stream write failed. "
"RPC canceled; closing channel", "RPC canceled; closing channel",
channel->id); channel->id);
chan_close_with_error(channel, buf, LOGLVL_ERR); chan_close_on_err(channel, buf, LOGLVL_ERR);
} }
return success; return success;
@@ -438,7 +437,7 @@ static void internal_read_event(void **argv)
if (p->read_size) { if (p->read_size) {
// This should not happen, as WBuffer is one single serialized message. // This should not happen, as WBuffer is one single serialized message.
if (!channel->rpc.closed) { if (!channel->rpc.closed) {
chan_close_with_error(channel, "internal channel: internal error", LOGLVL_ERR); chan_close_on_err(channel, "internal channel: internal error", LOGLVL_ERR);
} }
} }
@@ -519,9 +518,9 @@ void rpc_free(Channel *channel)
api_free_dict(channel->rpc.info); api_free_dict(channel->rpc.info);
} }
static void chan_close_with_error(Channel *channel, char *msg, int loglevel) /// Logs a fatal error received from a channel, then closes the channel.
static void chan_close_on_err(Channel *channel, char *msg, int loglevel)
{ {
LOG(loglevel, "RPC: %s", msg);
for (size_t i = 0; i < kv_size(channel->rpc.call_stack); i++) { for (size_t i = 0; i < kv_size(channel->rpc.call_stack); i++) {
ChannelCallFrame *frame = kv_A(channel->rpc.call_stack, i); ChannelCallFrame *frame = kv_A(channel->rpc.call_stack, i);
frame->returned = true; frame->returned = true;
@@ -530,6 +529,8 @@ static void chan_close_with_error(Channel *channel, char *msg, int loglevel)
} }
channel_close(channel->id, kChannelPartRpc, NULL); channel_close(channel->id, kChannelPartRpc, NULL);
LOG(loglevel, "RPC: %s", msg);
} }
static void serialize_request(Channel **chans, size_t nchans, uint32_t request_id, static void serialize_request(Channel **chans, size_t nchans, uint32_t request_id,