aboutsummaryrefslogtreecommitdiff
path: root/src/nvim/msgpack_rpc
diff options
context:
space:
mode:
Diffstat (limited to 'src/nvim/msgpack_rpc')
-rw-r--r--src/nvim/msgpack_rpc/channel.c49
-rw-r--r--src/nvim/msgpack_rpc/defs.h5
-rw-r--r--src/nvim/msgpack_rpc/remote_ui.c23
3 files changed, 42 insertions, 35 deletions
diff --git a/src/nvim/msgpack_rpc/channel.c b/src/nvim/msgpack_rpc/channel.c
index 861614f147..0e3b8200c9 100644
--- a/src/nvim/msgpack_rpc/channel.c
+++ b/src/nvim/msgpack_rpc/channel.c
@@ -68,6 +68,7 @@ typedef struct {
uint64_t next_request_id;
kvec_t(ChannelCallFrame *) call_stack;
kvec_t(WBuffer *) delayed_notifications;
+ Queue *events;
} Channel;
typedef struct {
@@ -123,14 +124,14 @@ void channel_teardown(void)
uint64_t channel_from_process(char **argv)
{
Channel *channel = register_channel(kChannelTypeProc);
- channel->data.process.uvproc = uv_process_init(channel);
+ channel->data.process.uvproc = uv_process_init(&loop, channel);
Process *proc = &channel->data.process.uvproc.process;
proc->argv = argv;
proc->in = &channel->data.process.in;
proc->out = &channel->data.process.out;
proc->err = &channel->data.process.err;
proc->cb = process_exit;
- if (!process_spawn(&loop, proc)) {
+ if (!process_spawn(proc)) {
loop_poll_events(&loop, 0);
decref(channel);
return 0;
@@ -153,6 +154,9 @@ void channel_from_connection(SocketWatcher *watcher)
{
Channel *channel = register_channel(kChannelTypeSocket);
socket_watcher_accept(watcher, &channel->data.stream, channel);
+ incref(channel); // close channel only after the stream is closed
+ channel->data.stream.internal_close_cb = close_cb;
+ channel->data.stream.internal_data = channel;
wstream_init(&channel->data.stream, 0);
rstream_init(&channel->data.stream, CHANNEL_BUFFER_SIZE);
rstream_start(&channel->data.stream, parse_msgpack);
@@ -221,7 +225,7 @@ Object channel_send_call(uint64_t id,
ChannelCallFrame frame = {request_id, false, false, NIL};
kv_push(ChannelCallFrame *, channel->call_stack, &frame);
channel->pending_requests++;
- LOOP_POLL_EVENTS_UNTIL(&loop, -1, frame.returned);
+ LOOP_PROCESS_EVENTS_UNTIL(&loop, channel->events, -1, frame.returned);
(void)kv_pop(channel->call_stack);
channel->pending_requests--;
@@ -324,7 +328,8 @@ static void channel_from_stdio(void)
wstream_init_fd(&loop, &channel->data.std.out, 1, 0, NULL);
}
-static void forward_stderr(Stream *stream, RBuffer *rbuf, void *data, bool eof)
+static void forward_stderr(Stream *stream, RBuffer *rbuf, size_t count,
+ void *data, bool eof)
{
while (rbuffer_size(rbuf)) {
char buf[256];
@@ -339,7 +344,8 @@ static void process_exit(Process *proc, int status, void *data)
decref(data);
}
-static void parse_msgpack(Stream *stream, RBuffer *rbuf, void *data, bool eof)
+static void parse_msgpack(Stream *stream, RBuffer *rbuf, size_t c, void *data,
+ bool eof)
{
Channel *channel = data;
incref(channel);
@@ -447,31 +453,31 @@ static void handle_request(Channel *channel, msgpack_object *request)
method->via.bin.size);
} else {
handler.fn = msgpack_rpc_handle_missing_method;
- handler.defer = false;
+ handler.async = true;
}
Array args = ARRAY_DICT_INIT;
if (!msgpack_rpc_to_array(msgpack_rpc_args(request), &args)) {
handler.fn = msgpack_rpc_handle_invalid_arguments;
- handler.defer = false;
+ handler.async = true;
}
- bool defer = (!kv_size(channel->call_stack) && handler.defer);
RequestEvent *event_data = xmalloc(sizeof(RequestEvent));
event_data->channel = channel;
event_data->handler = handler;
event_data->args = args;
event_data->request_id = request_id;
incref(channel);
- loop_push_event(&loop, (Event) {
- .handler = on_request_event,
- .data = event_data
- }, defer);
+ if (handler.async) {
+ on_request_event((void **)&event_data);
+ } else {
+ queue_put(channel->events, on_request_event, 1, event_data);
+ }
}
-static void on_request_event(Event event)
+static void on_request_event(void **argv)
{
- RequestEvent *e = event.data;
+ RequestEvent *e = argv[0];
Channel *channel = e->channel;
MsgpackRpcRequestHandler handler = e->handler;
Array args = e->args;
@@ -636,7 +642,7 @@ static void close_channel(Channel *channel)
switch (channel->type) {
case kChannelTypeSocket:
- stream_close(&channel->data.stream, close_cb);
+ stream_close(&channel->data.stream, NULL);
break;
case kChannelTypeProc:
if (!channel->data.process.uvproc.process.closed) {
@@ -646,9 +652,8 @@ static void close_channel(Channel *channel)
case kChannelTypeStdio:
stream_close(&channel->data.std.in, NULL);
stream_close(&channel->data.std.out, NULL);
- loop_push_event(&loop,
- (Event) { .handler = on_stdio_close, .data = channel }, false);
- break;
+ queue_put(loop.fast_events, exit_event, 1, channel);
+ return;
default:
abort();
}
@@ -656,9 +661,9 @@ static void close_channel(Channel *channel)
decref(channel);
}
-static void on_stdio_close(Event e)
+static void exit_event(void **argv)
{
- decref(e.data);
+ decref(argv[0]);
if (!exiting) {
mch_exit(0);
@@ -680,17 +685,19 @@ static void free_channel(Channel *channel)
pmap_free(cstr_t)(channel->subscribed_events);
kv_destroy(channel->call_stack);
kv_destroy(channel->delayed_notifications);
+ queue_free(channel->events);
xfree(channel);
}
static void close_cb(Stream *stream, void *data)
{
- xfree(data);
+ decref(data);
}
static Channel *register_channel(ChannelType type)
{
Channel *rv = xmalloc(sizeof(Channel));
+ rv->events = queue_new_child(loop.events);
rv->type = type;
rv->refcount = 1;
rv->closed = false;
diff --git a/src/nvim/msgpack_rpc/defs.h b/src/nvim/msgpack_rpc/defs.h
index 0492a65290..d97cf28ca1 100644
--- a/src/nvim/msgpack_rpc/defs.h
+++ b/src/nvim/msgpack_rpc/defs.h
@@ -11,9 +11,8 @@ typedef struct {
uint64_t request_id,
Array args,
Error *error);
- bool defer; // Should the call be deferred to the main loop? This should
- // be true if the function mutates editor data structures such
- // as buffers, windows, tabs, or if it executes vimscript code.
+ bool async; // function is always safe to run immediately instead of being
+ // put in a request queue for handling when nvim waits for input.
} MsgpackRpcRequestHandler;
/// Initializes the msgpack-rpc method table
diff --git a/src/nvim/msgpack_rpc/remote_ui.c b/src/nvim/msgpack_rpc/remote_ui.c
index 07d78dd9d7..3334b0e6af 100644
--- a/src/nvim/msgpack_rpc/remote_ui.c
+++ b/src/nvim/msgpack_rpc/remote_ui.c
@@ -28,7 +28,7 @@ void remote_ui_init(void)
connected_uis = pmap_new(uint64_t)();
// Add handler for "attach_ui"
String method = cstr_as_string("ui_attach");
- MsgpackRpcRequestHandler handler = {.fn = remote_ui_attach, .defer = false};
+ MsgpackRpcRequestHandler handler = {.fn = remote_ui_attach, .async = true};
msgpack_rpc_add_method_handler(method, handler);
method = cstr_as_string("ui_detach");
handler.fn = remote_ui_detach;
@@ -86,8 +86,7 @@ static Object remote_ui_attach(uint64_t channel_id, uint64_t request_id,
ui->busy_stop = remote_ui_busy_stop;
ui->mouse_on = remote_ui_mouse_on;
ui->mouse_off = remote_ui_mouse_off;
- ui->insert_mode = remote_ui_insert_mode;
- ui->normal_mode = remote_ui_normal_mode;
+ ui->mode_change = remote_ui_mode_change;
ui->set_scroll_region = remote_ui_set_scroll_region;
ui->scroll = remote_ui_scroll;
ui->highlight_set = remote_ui_highlight_set;
@@ -214,16 +213,18 @@ static void remote_ui_mouse_off(UI *ui)
push_call(ui, "mouse_off", args);
}
-static void remote_ui_insert_mode(UI *ui)
+static void remote_ui_mode_change(UI *ui, int mode)
{
Array args = ARRAY_DICT_INIT;
- push_call(ui, "insert_mode", args);
-}
-
-static void remote_ui_normal_mode(UI *ui)
-{
- Array args = ARRAY_DICT_INIT;
- push_call(ui, "normal_mode", args);
+ if (mode == INSERT) {
+ ADD(args, STRING_OBJ(cstr_to_string("insert")));
+ } else if (mode == REPLACE) {
+ ADD(args, STRING_OBJ(cstr_to_string("replace")));
+ } else {
+ assert(mode == NORMAL);
+ ADD(args, STRING_OBJ(cstr_to_string("normal")));
+ }
+ push_call(ui, "mode_change", args);
}
static void remote_ui_set_scroll_region(UI *ui, int top, int bot, int left,
generated by cgit (git 2.34.1) at 2025-09-26 09:19:36 +0000