aboutsummaryrefslogtreecommitdiff
path: root/src/nvim/msgpack_rpc/channel.c
diff options
context:
space:
mode:
Diffstat (limited to 'src/nvim/msgpack_rpc/channel.c')
-rw-r--r--src/nvim/msgpack_rpc/channel.c81
1 files changed, 58 insertions, 23 deletions
diff --git a/src/nvim/msgpack_rpc/channel.c b/src/nvim/msgpack_rpc/channel.c
index 34ff7c6374..5b249ee1c7 100644
--- a/src/nvim/msgpack_rpc/channel.c
+++ b/src/nvim/msgpack_rpc/channel.c
@@ -7,8 +7,8 @@
#include "nvim/api/private/helpers.h"
#include "nvim/api/vim.h"
+#include "nvim/api/ui.h"
#include "nvim/msgpack_rpc/channel.h"
-#include "nvim/msgpack_rpc/remote_ui.h"
#include "nvim/event/loop.h"
#include "nvim/event/libuv_process.h"
#include "nvim/event/rstream.h"
@@ -16,6 +16,7 @@
#include "nvim/event/socket.h"
#include "nvim/msgpack_rpc/helpers.h"
#include "nvim/vim.h"
+#include "nvim/main.h"
#include "nvim/ascii.h"
#include "nvim/memory.h"
#include "nvim/os_unix.h"
@@ -119,7 +120,7 @@ void channel_teardown(void)
uint64_t channel_from_process(char **argv)
{
Channel *channel = register_channel(kChannelTypeProc);
- channel->data.process.uvproc = libuv_process_init(&loop, channel);
+ channel->data.process.uvproc = libuv_process_init(&main_loop, channel);
Process *proc = &channel->data.process.uvproc.process;
proc->argv = argv;
proc->in = &channel->data.process.in;
@@ -127,7 +128,7 @@ uint64_t channel_from_process(char **argv)
proc->err = &channel->data.process.err;
proc->cb = process_exit;
if (!process_spawn(proc)) {
- loop_poll_events(&loop, 0);
+ loop_poll_events(&main_loop, 0);
decref(channel);
return 0;
}
@@ -179,7 +180,7 @@ bool channel_send_event(uint64_t id, char *name, Array args)
// Pending request, queue the notification for later sending.
String method = cstr_as_string(name);
WBuffer *buffer = serialize_request(id, 0, method, args, &out_buffer, 1);
- kv_push(WBuffer *, channel->delayed_notifications, buffer);
+ kv_push(channel->delayed_notifications, buffer);
} else {
send_event(channel, name, args);
}
@@ -217,10 +218,10 @@ Object channel_send_call(uint64_t id,
send_request(channel, request_id, method_name, args);
// Push the frame
- ChannelCallFrame frame = {request_id, false, false, NIL};
- kv_push(ChannelCallFrame *, channel->call_stack, &frame);
+ ChannelCallFrame frame = { request_id, false, false, NIL };
+ kv_push(channel->call_stack, &frame);
channel->pending_requests++;
- LOOP_PROCESS_EVENTS_UNTIL(&loop, channel->events, -1, frame.returned);
+ LOOP_PROCESS_EVENTS_UNTIL(&main_loop, channel->events, -1, frame.returned);
(void)kv_pop(channel->call_stack);
channel->pending_requests--;
@@ -316,11 +317,11 @@ void channel_from_stdio(void)
Channel *channel = register_channel(kChannelTypeStdio);
incref(channel); // stdio channels are only closed on exit
// read stream
- rstream_init_fd(&loop, &channel->data.std.in, 0, CHANNEL_BUFFER_SIZE,
- channel);
+ rstream_init_fd(&main_loop, &channel->data.std.in, 0, CHANNEL_BUFFER_SIZE,
+ channel);
rstream_start(&channel->data.std.in, parse_msgpack);
// write stream
- wstream_init_fd(&loop, &channel->data.std.out, 1, 0, NULL);
+ wstream_init_fd(&main_loop, &channel->data.std.out, 1, 0, NULL);
}
static void forward_stderr(Stream *stream, RBuffer *rbuf, size_t count,
@@ -574,13 +575,12 @@ static void send_event(Channel *channel,
static void broadcast_event(char *name, Array args)
{
- kvec_t(Channel *) subscribed;
- kv_init(subscribed);
+ kvec_t(Channel *) subscribed = KV_INITIAL_VALUE;
Channel *channel;
map_foreach_value(channels, channel, {
if (pmap_has(cstr_t)(channel->subscribed_events, name)) {
- kv_push(Channel *, subscribed, channel);
+ kv_push(subscribed, channel);
}
});
@@ -600,7 +600,7 @@ static void broadcast_event(char *name, Array args)
for (size_t i = 0; i < kv_size(subscribed); i++) {
Channel *channel = kv_A(subscribed, i);
if (channel->pending_requests) {
- kv_push(WBuffer *, channel->delayed_notifications, buffer);
+ kv_push(channel->delayed_notifications, buffer);
} else {
channel_write(channel, buffer);
}
@@ -647,7 +647,7 @@ static void close_channel(Channel *channel)
case kChannelTypeStdio:
stream_close(&channel->data.std.in, NULL);
stream_close(&channel->data.std.out, NULL);
- queue_put(loop.fast_events, exit_event, 1, channel);
+ queue_put(main_loop.fast_events, exit_event, 1, channel);
return;
default:
abort();
@@ -692,7 +692,7 @@ static void close_cb(Stream *stream, void *data)
static Channel *register_channel(ChannelType type)
{
Channel *rv = xmalloc(sizeof(Channel));
- rv->events = queue_new_child(loop.events);
+ rv->events = queue_new_child(main_loop.events);
rv->type = type;
rv->refcount = 1;
rv->closed = false;
@@ -816,20 +816,55 @@ static void decref(Channel *channel)
#define REQ "[request] "
#define RES "[response] "
#define NOT "[notification] "
+#define ERR "[error] "
+
+// Cannot define array with negative offsets, so this one is needed to be added
+// to MSGPACK_UNPACK_\* values.
+#define MUR_OFF 2
+
+static const char *const msgpack_error_messages[] = {
+ [MSGPACK_UNPACK_EXTRA_BYTES + MUR_OFF] = "extra bytes found",
+ [MSGPACK_UNPACK_CONTINUE + MUR_OFF] = "incomplete string",
+ [MSGPACK_UNPACK_PARSE_ERROR + MUR_OFF] = "parse error",
+ [MSGPACK_UNPACK_NOMEM_ERROR + MUR_OFF] = "not enough memory",
+};
static void log_server_msg(uint64_t channel_id,
msgpack_sbuffer *packed)
{
msgpack_unpacked unpacked;
msgpack_unpacked_init(&unpacked);
- msgpack_unpack_next(&unpacked, packed->data, packed->size, NULL);
- uint64_t type = unpacked.data.via.array.ptr[0].via.u64;
DLOGN("[msgpack-rpc] nvim -> client(%" PRIu64 ") ", channel_id);
- log_lock();
- FILE *f = open_log_file();
- fprintf(f, type ? (type == 1 ? RES : NOT) : REQ);
- log_msg_close(f, unpacked.data);
- msgpack_unpacked_destroy(&unpacked);
+ const msgpack_unpack_return result =
+ msgpack_unpack_next(&unpacked, packed->data, packed->size, NULL);
+ switch (result) {
+ case MSGPACK_UNPACK_SUCCESS: {
+ uint64_t type = unpacked.data.via.array.ptr[0].via.u64;
+ log_lock();
+ FILE *f = open_log_file();
+ fprintf(f, type ? (type == 1 ? RES : NOT) : REQ);
+ log_msg_close(f, unpacked.data);
+ msgpack_unpacked_destroy(&unpacked);
+ break;
+ }
+ case MSGPACK_UNPACK_EXTRA_BYTES:
+ case MSGPACK_UNPACK_CONTINUE:
+ case MSGPACK_UNPACK_PARSE_ERROR:
+ case MSGPACK_UNPACK_NOMEM_ERROR: {
+ log_lock();
+ FILE *f = open_log_file();
+ fprintf(f, ERR);
+ log_msg_close(f, (msgpack_object) {
+ .type = MSGPACK_OBJECT_STR,
+ .via.str = {
+ .ptr = (char *)msgpack_error_messages[result + MUR_OFF],
+ .size = (uint32_t)strlen(
+ msgpack_error_messages[result + MUR_OFF]),
+ },
+ });
+ break;
+ }
+ }
}
static void log_client_msg(uint64_t channel_id,
generated by cgit (git 2.34.1) at 2025-07-23 22:00:07 +0000