Merge pull request #6594 from bfredl/sockopen
connect to socket (RPC only for the moment)
This commit is contained in:
commit
1b7a9bf4d2
|
@ -6863,6 +6863,21 @@ sinh({expr}) *sinh()*
|
|||
:echo sinh(-0.9)
|
||||
< -1.026517
|
||||
|
||||
sockconnect({mode}, {address}, {opts}) *sockconnect()*
|
||||
Connect a socket to an address. If {mode} is "pipe" then
|
||||
{address} should be the path of a named pipe. If {mode} is
|
||||
"tcp" then {address} should be of the form "host:port" where
|
||||
the host should be an ip adderess or host name, and port the
|
||||
port number. Currently only rpc sockets are supported, so
|
||||
{opts} must be passed with "rpc" set to |TRUE|.
|
||||
|
||||
{opts} is a dictionary with these keys:
|
||||
rpc : If set, |msgpack-rpc| will be used to communicate
|
||||
over the socket.
|
||||
Returns:
|
||||
- The channel ID on success, which is used by
|
||||
|rpcnotify()| and |rpcrequest()| and |rpcstop()|.
|
||||
- 0 on invalid arguments or connection failure.
|
||||
|
||||
sort({list} [, {func} [, {dict}]]) *sort()* *E702*
|
||||
Sort the items in {list} in-place. Returns {list}.
|
||||
|
|
|
@ -15058,6 +15058,54 @@ static void f_simplify(typval_T *argvars, typval_T *rettv, FunPtr fptr)
|
|||
rettv->v_type = VAR_STRING;
|
||||
}
|
||||
|
||||
/// "sockconnect()" function
|
||||
static void f_sockconnect(typval_T *argvars, typval_T *rettv, FunPtr fptr)
|
||||
{
|
||||
if (argvars[0].v_type != VAR_STRING || argvars[1].v_type != VAR_STRING) {
|
||||
EMSG(_(e_invarg));
|
||||
return;
|
||||
}
|
||||
if (argvars[2].v_type != VAR_DICT && argvars[2].v_type != VAR_UNKNOWN) {
|
||||
// Wrong argument types
|
||||
EMSG2(_(e_invarg2), "expected dictionary");
|
||||
return;
|
||||
}
|
||||
|
||||
const char *mode = tv_get_string(&argvars[0]);
|
||||
const char *address = tv_get_string(&argvars[1]);
|
||||
|
||||
bool tcp;
|
||||
if (strcmp(mode, "tcp") == 0) {
|
||||
tcp = true;
|
||||
} else if (strcmp(mode, "pipe") == 0) {
|
||||
tcp = false;
|
||||
} else {
|
||||
EMSG2(_(e_invarg2), "invalid mode");
|
||||
return;
|
||||
}
|
||||
|
||||
bool rpc = false;
|
||||
if (argvars[2].v_type == VAR_DICT) {
|
||||
dict_T *opts = argvars[2].vval.v_dict;
|
||||
rpc = tv_dict_get_number(opts, "rpc") != 0;
|
||||
}
|
||||
|
||||
if (!rpc) {
|
||||
EMSG2(_(e_invarg2), "rpc option must be true");
|
||||
return;
|
||||
}
|
||||
|
||||
const char *error = NULL;
|
||||
uint64_t id = channel_connect(tcp, address, 50, &error);
|
||||
|
||||
if (error) {
|
||||
EMSG2(_("connection failed: %s"), error);
|
||||
}
|
||||
|
||||
rettv->vval.v_number = (varnumber_T)id;
|
||||
rettv->v_type = VAR_NUMBER;
|
||||
}
|
||||
|
||||
/// struct used in the array that's given to qsort()
|
||||
typedef struct {
|
||||
listitem_T *item;
|
||||
|
|
|
@ -268,6 +268,7 @@ return {
|
|||
simplify={args=1},
|
||||
sin={args=1, func="float_op_wrapper", data="&sin"},
|
||||
sinh={args=1, func="float_op_wrapper", data="&sinh"},
|
||||
sockconnect={args={2,3}},
|
||||
sort={args={1, 3}},
|
||||
soundfold={args=1},
|
||||
spellbadword={args={0, 1}},
|
||||
|
|
|
@ -15,6 +15,7 @@
|
|||
#include "nvim/vim.h"
|
||||
#include "nvim/strings.h"
|
||||
#include "nvim/path.h"
|
||||
#include "nvim/main.h"
|
||||
#include "nvim/memory.h"
|
||||
#include "nvim/macros.h"
|
||||
#include "nvim/charset.h"
|
||||
|
@ -189,3 +190,76 @@ static void close_cb(uv_handle_t *handle)
|
|||
watcher->close_cb(watcher, watcher->data);
|
||||
}
|
||||
}
|
||||
|
||||
static void connect_cb(uv_connect_t *req, int status)
|
||||
{
|
||||
int *ret_status = req->data;
|
||||
*ret_status = status;
|
||||
if (status != 0) {
|
||||
uv_close((uv_handle_t *)req->handle, NULL);
|
||||
}
|
||||
}
|
||||
|
||||
bool socket_connect(Loop *loop, Stream *stream,
|
||||
bool is_tcp, const char *address,
|
||||
int timeout, const char **error)
|
||||
{
|
||||
bool success = false;
|
||||
int status;
|
||||
uv_connect_t req;
|
||||
req.data = &status;
|
||||
uv_stream_t *uv_stream;
|
||||
|
||||
uv_tcp_t *tcp = &stream->uv.tcp;
|
||||
uv_getaddrinfo_t addr_req;
|
||||
addr_req.addrinfo = NULL;
|
||||
const struct addrinfo *addrinfo = NULL;
|
||||
char *addr = NULL;
|
||||
if (is_tcp) {
|
||||
addr = xstrdup(address);
|
||||
char *host_end = strrchr(addr, ':');
|
||||
if (!host_end) {
|
||||
*error = _("tcp address must be host:port");
|
||||
goto cleanup;
|
||||
}
|
||||
*host_end = NUL;
|
||||
|
||||
const struct addrinfo hints = { .ai_family = AF_UNSPEC,
|
||||
.ai_socktype = SOCK_STREAM,
|
||||
.ai_flags = AI_NUMERICSERV };
|
||||
int retval = uv_getaddrinfo(&loop->uv, &addr_req, NULL,
|
||||
addr, host_end+1, &hints);
|
||||
if (retval != 0) {
|
||||
*error = _("failed to lookup host or port");
|
||||
goto cleanup;
|
||||
}
|
||||
addrinfo = addr_req.addrinfo;
|
||||
|
||||
tcp_retry:
|
||||
uv_tcp_init(&loop->uv, tcp);
|
||||
uv_tcp_connect(&req, tcp, addrinfo->ai_addr, connect_cb);
|
||||
uv_stream = (uv_stream_t *)tcp;
|
||||
|
||||
} else {
|
||||
uv_pipe_t *pipe = &stream->uv.pipe;
|
||||
uv_pipe_init(&loop->uv, pipe, 0);
|
||||
uv_pipe_connect(&req, pipe, address, connect_cb);
|
||||
uv_stream = (uv_stream_t *)pipe;
|
||||
}
|
||||
status = 1;
|
||||
LOOP_PROCESS_EVENTS_UNTIL(&main_loop, NULL, timeout, status != 1);
|
||||
if (status == 0) {
|
||||
stream_init(NULL, stream, -1, uv_stream);
|
||||
success = true;
|
||||
} else if (is_tcp && addrinfo->ai_next) {
|
||||
addrinfo = addrinfo->ai_next;
|
||||
goto tcp_retry;
|
||||
} else {
|
||||
*error = _("connection refused");
|
||||
}
|
||||
|
||||
cleanup:
|
||||
xfree(addr);
|
||||
uv_freeaddrinfo(addr_req.addrinfo);
|
||||
return success;
|
||||
}
|
||||
|
|
|
@ -12,6 +12,7 @@
|
|||
#include "nvim/api/vim.h"
|
||||
#include "nvim/api/ui.h"
|
||||
#include "nvim/msgpack_rpc/channel.h"
|
||||
#include "nvim/msgpack_rpc/server.h"
|
||||
#include "nvim/event/loop.h"
|
||||
#include "nvim/event/libuv_process.h"
|
||||
#include "nvim/event/rstream.h"
|
||||
|
@ -28,6 +29,7 @@
|
|||
#include "nvim/map.h"
|
||||
#include "nvim/log.h"
|
||||
#include "nvim/misc1.h"
|
||||
#include "nvim/path.h"
|
||||
#include "nvim/lib/kvec.h"
|
||||
#include "nvim/os/input.h"
|
||||
|
||||
|
@ -41,7 +43,8 @@
|
|||
typedef enum {
|
||||
kChannelTypeSocket,
|
||||
kChannelTypeProc,
|
||||
kChannelTypeStdio
|
||||
kChannelTypeStdio,
|
||||
kChannelTypeInternal
|
||||
} ChannelType;
|
||||
|
||||
typedef struct {
|
||||
|
@ -125,7 +128,7 @@ uint64_t channel_from_process(Process *proc, uint64_t id)
|
|||
|
||||
wstream_init(proc->in, 0);
|
||||
rstream_init(proc->out, 0);
|
||||
rstream_start(proc->out, parse_msgpack, channel);
|
||||
rstream_start(proc->out, receive_msgpack, channel);
|
||||
|
||||
return channel->id;
|
||||
}
|
||||
|
@ -142,7 +145,36 @@ void channel_from_connection(SocketWatcher *watcher)
|
|||
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, channel);
|
||||
rstream_start(&channel->data.stream, receive_msgpack, channel);
|
||||
}
|
||||
|
||||
uint64_t channel_connect(bool tcp, const char *address,
|
||||
int timeout, const char **error)
|
||||
{
|
||||
if (!tcp) {
|
||||
char *path = fix_fname(address);
|
||||
if (server_owns_pipe_address(path)) {
|
||||
// avoid deadlock
|
||||
xfree(path);
|
||||
return channel_create_internal();
|
||||
}
|
||||
xfree(path);
|
||||
}
|
||||
|
||||
Channel *channel = register_channel(kChannelTypeSocket, 0, NULL);
|
||||
if (!socket_connect(&main_loop, &channel->data.stream,
|
||||
tcp, address, timeout, error)) {
|
||||
decref(channel);
|
||||
return 0;
|
||||
}
|
||||
|
||||
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, receive_msgpack, channel);
|
||||
return channel->id;
|
||||
}
|
||||
|
||||
/// Sends event/arguments to channel
|
||||
|
@ -305,11 +337,20 @@ void channel_from_stdio(void)
|
|||
incref(channel); // stdio channels are only closed on exit
|
||||
// read stream
|
||||
rstream_init_fd(&main_loop, &channel->data.std.in, 0, CHANNEL_BUFFER_SIZE);
|
||||
rstream_start(&channel->data.std.in, parse_msgpack, channel);
|
||||
rstream_start(&channel->data.std.in, receive_msgpack, channel);
|
||||
// write stream
|
||||
wstream_init_fd(&main_loop, &channel->data.std.out, 1, 0);
|
||||
}
|
||||
|
||||
/// Creates a loopback channel. This is used to avoid deadlock
|
||||
/// when an instance connects to its own named pipe.
|
||||
uint64_t channel_create_internal(void)
|
||||
{
|
||||
Channel *channel = register_channel(kChannelTypeInternal, 0, NULL);
|
||||
incref(channel); // internal channel lives until process exit
|
||||
return channel->id;
|
||||
}
|
||||
|
||||
void channel_process_exit(uint64_t id, int status)
|
||||
{
|
||||
Channel *channel = pmap_get(uint64_t)(channels, id);
|
||||
|
@ -318,8 +359,8 @@ void channel_process_exit(uint64_t id, int status)
|
|||
decref(channel);
|
||||
}
|
||||
|
||||
static void parse_msgpack(Stream *stream, RBuffer *rbuf, size_t c, void *data,
|
||||
bool eof)
|
||||
static void receive_msgpack(Stream *stream, RBuffer *rbuf, size_t c,
|
||||
void *data, bool eof)
|
||||
{
|
||||
Channel *channel = data;
|
||||
incref(channel);
|
||||
|
@ -341,6 +382,14 @@ static void parse_msgpack(Stream *stream, RBuffer *rbuf, size_t c, void *data,
|
|||
rbuffer_read(rbuf, msgpack_unpacker_buffer(channel->unpacker), count);
|
||||
msgpack_unpacker_buffer_consumed(channel->unpacker, count);
|
||||
|
||||
parse_msgpack(channel);
|
||||
|
||||
end:
|
||||
decref(channel);
|
||||
}
|
||||
|
||||
static void parse_msgpack(Channel *channel)
|
||||
{
|
||||
msgpack_unpacked unpacked;
|
||||
msgpack_unpacked_init(&unpacked);
|
||||
msgpack_unpack_return result;
|
||||
|
@ -364,7 +413,7 @@ static void parse_msgpack(Stream *stream, RBuffer *rbuf, size_t c, void *data,
|
|||
}
|
||||
msgpack_unpacked_destroy(&unpacked);
|
||||
// Bail out from this event loop iteration
|
||||
goto end;
|
||||
return;
|
||||
}
|
||||
|
||||
handle_request(channel, &unpacked.data);
|
||||
|
@ -388,11 +437,9 @@ static void parse_msgpack(Stream *stream, RBuffer *rbuf, size_t c, void *data,
|
|||
"This error can also happen when deserializing "
|
||||
"an object with high level of nesting");
|
||||
}
|
||||
|
||||
end:
|
||||
decref(channel);
|
||||
}
|
||||
|
||||
|
||||
static void handle_request(Channel *channel, msgpack_object *request)
|
||||
FUNC_ATTR_NONNULL_ALL
|
||||
{
|
||||
|
@ -502,8 +549,11 @@ static bool channel_write(Channel *channel, WBuffer *buffer)
|
|||
case kChannelTypeStdio:
|
||||
success = wstream_write(&channel->data.std.out, buffer);
|
||||
break;
|
||||
default:
|
||||
abort();
|
||||
case kChannelTypeInternal:
|
||||
incref(channel);
|
||||
CREATE_EVENT(channel->events, internal_read_event, 2, channel, buffer);
|
||||
success = true;
|
||||
break;
|
||||
}
|
||||
|
||||
if (!success) {
|
||||
|
@ -520,6 +570,22 @@ static bool channel_write(Channel *channel, WBuffer *buffer)
|
|||
return success;
|
||||
}
|
||||
|
||||
static void internal_read_event(void **argv)
|
||||
{
|
||||
Channel *channel = argv[0];
|
||||
WBuffer *buffer = argv[1];
|
||||
|
||||
msgpack_unpacker_reserve_buffer(channel->unpacker, buffer->size);
|
||||
memcpy(msgpack_unpacker_buffer(channel->unpacker),
|
||||
buffer->data, buffer->size);
|
||||
msgpack_unpacker_buffer_consumed(channel->unpacker, buffer->size);
|
||||
|
||||
parse_msgpack(channel);
|
||||
|
||||
decref(channel);
|
||||
wstream_release_wbuffer(buffer);
|
||||
}
|
||||
|
||||
static void send_error(Channel *channel, uint64_t id, char *err)
|
||||
{
|
||||
Error e = ERROR_INIT;
|
||||
|
@ -636,8 +702,9 @@ static void close_channel(Channel *channel)
|
|||
stream_close(&channel->data.std.out, NULL, NULL);
|
||||
multiqueue_put(main_loop.fast_events, exit_event, 1, channel);
|
||||
return;
|
||||
default:
|
||||
abort();
|
||||
case kChannelTypeInternal:
|
||||
// nothing to free.
|
||||
break;
|
||||
}
|
||||
|
||||
decref(channel);
|
||||
|
|
|
@ -97,6 +97,18 @@ char *server_address_new(void)
|
|||
#endif
|
||||
}
|
||||
|
||||
/// Check if this instance owns a pipe address.
|
||||
/// The argument must already be resolved to an absolute path!
|
||||
bool server_owns_pipe_address(const char *path)
|
||||
{
|
||||
for (int i = 0; i < watchers.ga_len; i++) {
|
||||
if (!strcmp(path, ((SocketWatcher **)watchers.ga_data)[i]->addr)) {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
/// Starts listening for API calls.
|
||||
///
|
||||
/// The socket type is determined by parsing `endpoint`: If it's a valid IPv4
|
||||
|
|
|
@ -1715,7 +1715,7 @@ int vim_FullName(const char *fname, char *buf, size_t len, bool force)
|
|||
///
|
||||
/// @param fname is the filename to expand
|
||||
/// @return [allocated] Full path (NULL for failure).
|
||||
char *fix_fname(char *fname)
|
||||
char *fix_fname(const char *fname)
|
||||
{
|
||||
#ifdef UNIX
|
||||
return FullName_save(fname, true);
|
||||
|
|
|
@ -9,6 +9,8 @@ local nvim_prog, command, funcs = helpers.nvim_prog, helpers.command, helpers.fu
|
|||
local source, next_message = helpers.source, helpers.next_message
|
||||
local ok = helpers.ok
|
||||
local meths = helpers.meths
|
||||
local spawn, nvim_argv = helpers.spawn, helpers.nvim_argv
|
||||
local set_session = helpers.set_session
|
||||
|
||||
describe('server -> client', function()
|
||||
local cid
|
||||
|
@ -225,4 +227,75 @@ describe('server -> client', function()
|
|||
end)
|
||||
end)
|
||||
|
||||
describe('when connecting to another nvim instance', function()
|
||||
local function connect_test(server, mode, address)
|
||||
local serverpid = funcs.getpid()
|
||||
local client = spawn(nvim_argv)
|
||||
set_session(client, true)
|
||||
local clientpid = funcs.getpid()
|
||||
neq(serverpid, clientpid)
|
||||
local id = funcs.sockconnect(mode, address, {rpc=true})
|
||||
ok(id > 0)
|
||||
|
||||
funcs.rpcrequest(id, 'nvim_set_current_line', 'hello')
|
||||
local client_id = funcs.rpcrequest(id, 'nvim_get_api_info')[1]
|
||||
|
||||
set_session(server, true)
|
||||
eq(serverpid, funcs.getpid())
|
||||
eq('hello', meths.get_current_line())
|
||||
|
||||
-- method calls work both ways
|
||||
funcs.rpcrequest(client_id, 'nvim_set_current_line', 'howdy!')
|
||||
eq(id, funcs.rpcrequest(client_id, 'nvim_get_api_info')[1])
|
||||
|
||||
set_session(client, true)
|
||||
eq(clientpid, funcs.getpid())
|
||||
eq('howdy!', meths.get_current_line())
|
||||
|
||||
server:close()
|
||||
client:close()
|
||||
end
|
||||
|
||||
it('over a named pipe', function()
|
||||
local server = spawn(nvim_argv)
|
||||
set_session(server)
|
||||
local address = funcs.serverlist()[1]
|
||||
local first = string.sub(address,1,1)
|
||||
ok(first == '/' or first == '\\')
|
||||
connect_test(server, 'pipe', address)
|
||||
end)
|
||||
|
||||
it('to an ip adress', function()
|
||||
local server = spawn(nvim_argv)
|
||||
set_session(server)
|
||||
local address = funcs.serverstart("127.0.0.1:")
|
||||
eq('127.0.0.1:', string.sub(address,1,10))
|
||||
connect_test(server, 'tcp', address)
|
||||
end)
|
||||
|
||||
it('to a hostname', function()
|
||||
local server = spawn(nvim_argv)
|
||||
set_session(server)
|
||||
local address = funcs.serverstart("localhost:")
|
||||
eq('localhost:', string.sub(address,1,10))
|
||||
connect_test(server, 'tcp', address)
|
||||
end)
|
||||
end)
|
||||
|
||||
describe('when connecting to its own pipe adress', function()
|
||||
it('it does not deadlock', function()
|
||||
local address = funcs.serverlist()[1]
|
||||
local first = string.sub(address,1,1)
|
||||
ok(first == '/' or first == '\\')
|
||||
local serverpid = funcs.getpid()
|
||||
|
||||
local id = funcs.sockconnect('pipe', address, {rpc=true})
|
||||
|
||||
funcs.rpcrequest(id, 'nvim_set_current_line', 'hello')
|
||||
eq('hello', meths.get_current_line())
|
||||
eq(serverpid, funcs.rpcrequest(id, "nvim_eval", "getpid()"))
|
||||
|
||||
eq(id, funcs.rpcrequest(id, 'nvim_get_api_info')[1])
|
||||
end)
|
||||
end)
|
||||
end)
|
||||
|
|
|
@ -76,8 +76,8 @@ end
|
|||
|
||||
local session, loop_running, last_error
|
||||
|
||||
local function set_session(s)
|
||||
if session then
|
||||
local function set_session(s, keep)
|
||||
if session and not keep then
|
||||
session:close()
|
||||
end
|
||||
session = s
|
||||
|
@ -609,6 +609,7 @@ local module = {
|
|||
nvim = nvim,
|
||||
nvim_async = nvim_async,
|
||||
nvim_prog = nvim_prog,
|
||||
nvim_argv = nvim_argv,
|
||||
nvim_set = nvim_set,
|
||||
nvim_dir = nvim_dir,
|
||||
buffer = buffer,
|
||||
|
|
Loading…
Reference in New Issue