diff options
Diffstat (limited to 'input')
| -rw-r--r-- | input/ipc.c | 723 |
1 files changed, 723 insertions, 0 deletions
diff --git a/input/ipc.c b/input/ipc.c new file mode 100644 index 0000000000..7ce8aa3424 --- /dev/null +++ b/input/ipc.c @@ -0,0 +1,723 @@ +/* + * This file is part of mpv. + * + * mpv is free software; you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation; either version 2 of the License, or + * (at your option) any later version. + * + * mpv is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License along + * with mpv. If not, see <http://www.gnu.org/licenses/>. + */ + +#include <pthread.h> +#include <assert.h> +#include <errno.h> +#include <unistd.h> + +#include <poll.h> +#include <sys/types.h> +#include <sys/socket.h> +#include <sys/stat.h> +#include <sys/un.h> + +#include "config.h" + +#include "osdep/io.h" + +#include "common/common.h" +#include "common/msg.h" +#include "input/input.h" +#include "libmpv/client.h" +#include "misc/bstr.h" +#include "misc/json.h" +#include "options/m_option.h" +#include "options/path.h" +#include "player/core.h" +#include "player/client.h" + +struct thread_arg { + struct MPContext *mpctx; + const char *path; + + pthread_t thread; + int death_pipe[2]; +}; + +struct client_arg { + struct mp_log *log; + struct mpv_handle *client; + + int client_fd; + + char *(*encode_event)(mpv_event *event); + char *(*execute_command)(struct client_arg *arg, bstr msg); +}; + +static mpv_node *mpv_node_map_get(mpv_node *src, const char *key) +{ + if (src->format != MPV_FORMAT_NODE_MAP) + return NULL; + + for (int i = 0; i < src->u.list->num; i++) + if (!strcmp(key, src->u.list->keys[i])) + return &src->u.list->values[i]; + + return NULL; +} + +static mpv_node *mpv_node_array_get(mpv_node *src, int index) +{ + if (src->format != MPV_FORMAT_NODE_ARRAY) + return NULL; + + if (src->u.list->num < (index + 1)) + return NULL; + + return &src->u.list->values[index]; +} + +static void mpv_node_array_add(void *ta_parent, mpv_node *src, mpv_node *val) +{ + if (src->format != MPV_FORMAT_NODE_ARRAY) + return; + + if (!src->u.list) + src->u.list = talloc_zero(ta_parent, mpv_node_list); + + MP_TARRAY_GROW(src->u.list, src->u.list->values, src->u.list->num); + + static const struct m_option type = { .type = CONF_TYPE_NODE }; + m_option_get_node(&type, ta_parent, &src->u.list->values[src->u.list->num], val); + + src->u.list->num++; +} + +static void mpv_node_array_add_string(void *ta_parent, mpv_node *src, const char *val) +{ + mpv_node val_node = {.format = MPV_FORMAT_STRING, .u.string = (char *)val}; + mpv_node_array_add(ta_parent, src, &val_node); +} + +static void mpv_node_map_add(void *ta_parent, mpv_node *src, const char *key, mpv_node *val) +{ + if (src->format != MPV_FORMAT_NODE_MAP) + return; + + if (!src->u.list) + src->u.list = talloc_zero(ta_parent, mpv_node_list); + + MP_TARRAY_GROW(src->u.list, src->u.list->keys, src->u.list->num); + MP_TARRAY_GROW(src->u.list, src->u.list->values, src->u.list->num); + + src->u.list->keys[src->u.list->num] = talloc_strdup(ta_parent, key); + + static const struct m_option type = { .type = CONF_TYPE_NODE }; + m_option_get_node(&type, ta_parent, &src->u.list->values[src->u.list->num], val); + + src->u.list->num++; +} + +static void mpv_node_map_add_null(void *ta_parent, mpv_node *src, const char *key) +{ + mpv_node val_node = {.format = MPV_FORMAT_NONE}; + mpv_node_map_add(ta_parent, src, key, &val_node); +} + +static void mpv_node_map_add_flag(void *ta_parent, mpv_node *src, const char *key, bool val) +{ + mpv_node val_node = {.format = MPV_FORMAT_FLAG, .u.flag = val}; + + mpv_node_map_add(ta_parent, src, key, &val_node); +} + +static void mpv_node_map_add_int64(void *ta_parent, mpv_node *src, const char *key, int64_t val) +{ + mpv_node val_node = {.format = MPV_FORMAT_INT64, .u.int64 = val}; + mpv_node_map_add(ta_parent, src, key, &val_node); +} + +static void mpv_node_map_add_double(void *ta_parent, mpv_node *src, const char *key, double val) +{ + mpv_node val_node = {.format = MPV_FORMAT_DOUBLE, .u.double_ = val}; + mpv_node_map_add(ta_parent, src, key, &val_node); +} + +static void mpv_node_map_add_string(void *ta_parent, mpv_node *src, const char *key, const char *val) +{ + mpv_node val_node = {.format = MPV_FORMAT_STRING, .u.string = (char*)val}; + mpv_node_map_add(ta_parent, src, key, &val_node); +} + +static void mpv_event_to_node(void *ta_parent, mpv_event *event, mpv_node *dst) +{ + mpv_node_map_add_string(ta_parent, dst, "event", mpv_event_name(event->event_id)); + + if (event->reply_userdata) + mpv_node_map_add_int64(ta_parent, dst, "id", event->reply_userdata); + + if (event->error < 0) + mpv_node_map_add_string(ta_parent, dst, "error", mpv_error_string(event->error)); + + switch (event->event_id) { + case MPV_EVENT_LOG_MESSAGE: { + mpv_event_log_message *msg = event->data; + + mpv_node_map_add_string(ta_parent, dst, "prefix", msg->prefix); + mpv_node_map_add_string(ta_parent, dst, "level", msg->level); + mpv_node_map_add_string(ta_parent, dst, "text", msg->text); + + break; + } + + case MPV_EVENT_SCRIPT_INPUT_DISPATCH: { + mpv_event_script_input_dispatch *msg = event->data; + + mpv_node_map_add_int64(ta_parent, dst, "arg0", msg->arg0); + mpv_node_map_add_string(ta_parent, dst, "type", msg->type); + + break; + } + + case MPV_EVENT_CLIENT_MESSAGE: { + mpv_event_client_message *msg = event->data; + + mpv_node args_node = {.format = MPV_FORMAT_NODE_ARRAY, .u.list = NULL}; + for (int n = 0; n < msg->num_args; n++) + mpv_node_array_add_string(ta_parent, &args_node, msg->args[n]); + mpv_node_map_add(ta_parent, dst, "args", &args_node); + break; + } + + case MPV_EVENT_PROPERTY_CHANGE: { + mpv_event_property *prop = event->data; + + mpv_node_map_add_string(ta_parent, dst, "name", prop->name); + + switch (prop->format) { + case MPV_FORMAT_NODE: + mpv_node_map_add(ta_parent, dst, "data", prop->data); + break; + case MPV_FORMAT_DOUBLE: + mpv_node_map_add_double(ta_parent, dst, "data", *(double *)prop->data); + break; + case MPV_FORMAT_FLAG: + mpv_node_map_add_flag(ta_parent, dst, "data", *(int *)prop->data); + break; + case MPV_FORMAT_STRING: + mpv_node_map_add_string(ta_parent, dst, "data", *(char **)prop->data); + break; + default: + mpv_node_map_add_null(ta_parent, dst, "data"); + } + break; + } + } +} + +static char *json_encode_event(mpv_event *event) +{ + void *ta_parent = talloc_new(NULL); + mpv_node event_node = {.format = MPV_FORMAT_NODE_MAP, .u.list = NULL}; + + mpv_event_to_node(ta_parent, event, &event_node); + + char *output = talloc_strdup(NULL, ""); + json_write(&output, &event_node); + output = ta_talloc_strdup_append(output, "\n"); + + talloc_free(ta_parent); + + return output; +} + +static char *json_execute_command(struct client_arg *arg, bstr msg) +{ + int rc; + const char *cmd = NULL; + + mpv_node msg_node; + mpv_node reply_node = {.format = MPV_FORMAT_NODE_MAP, .u.list = NULL}; + + void *ta_parent = talloc_new(NULL); + + char *src = bstrdup0(ta_parent, msg); + rc = json_parse(ta_parent, &msg_node, &src, 3); + if (rc < 0) { + rc = MPV_ERROR_INVALID_PARAMETER; + goto error; + } + + if (msg_node.format != MPV_FORMAT_NODE_MAP) { + rc = MPV_ERROR_INVALID_PARAMETER; + goto error; + } + + mpv_node *cmd_node = mpv_node_map_get(&msg_node, "command"); + if (!cmd_node || + (cmd_node->format != MPV_FORMAT_NODE_ARRAY) || + !cmd_node->u.list->num) + { + rc = MPV_ERROR_INVALID_PARAMETER; + goto error; + } + + mpv_node *cmd_str_node = mpv_node_array_get(cmd_node, 0); + if (!cmd_str_node || (cmd_str_node->format != MPV_FORMAT_STRING)) { + rc = MPV_ERROR_INVALID_PARAMETER; + goto error; + } + + cmd = cmd_str_node->u.string; + + if (!strcmp("client_name", cmd)) { + const char *client_name = mpv_client_name(arg->client); + mpv_node_map_add_string(ta_parent, &reply_node, "data", client_name); + rc = MPV_ERROR_SUCCESS; + } else if (!strcmp("get_time_us", cmd)) { + int64_t time_us = mpv_get_time_us(arg->client); + mpv_node_map_add_int64(ta_parent, &reply_node, "data", time_us); + rc = MPV_ERROR_SUCCESS; + } else if (!strcmp("get_property", cmd)) { + mpv_node result_node; + + if (cmd_node->u.list->num != 2) { + rc = MPV_ERROR_INVALID_PARAMETER; + goto error; + } + + if (cmd_node->u.list->values[1].format != MPV_FORMAT_STRING) { + rc = MPV_ERROR_INVALID_PARAMETER; + goto error; + } + + rc = mpv_get_property(arg->client, cmd_node->u.list->values[1].u.string, + MPV_FORMAT_NODE, &result_node); + if (rc >= 0) { + mpv_node_map_add(ta_parent, &reply_node, "data", &result_node); + mpv_free_node_contents(&result_node); + } + } else if (!strcmp("get_property_string", cmd)) { + if (cmd_node->u.list->num != 2) { + rc = MPV_ERROR_INVALID_PARAMETER; + goto error; + } + + if (cmd_node->u.list->values[1].format != MPV_FORMAT_STRING) { + rc = MPV_ERROR_INVALID_PARAMETER; + goto error; + } + + char *result = mpv_get_property_string(arg->client, + cmd_node->u.list->values[1].u.string); + if (!result) { + mpv_node_map_add_null(ta_parent, &reply_node, "data"); + } else { + mpv_node_map_add_string(ta_parent, &reply_node, "data", result); + mpv_free(result); + } + } else if (!strcmp("set_property", cmd)) { + if (cmd_node->u.list->num != 3) { + rc = MPV_ERROR_INVALID_PARAMETER; + goto error; + } + + if (cmd_node->u.list->values[1].format != MPV_FORMAT_STRING) { + rc = MPV_ERROR_INVALID_PARAMETER; + goto error; + } + + rc = mpv_set_property(arg->client, cmd_node->u.list->values[1].u.string, + MPV_FORMAT_NODE, &cmd_node->u.list->values[2]); + } else if (!strcmp("set_property_string", cmd)) { + if (cmd_node->u.list->num != 3) { + rc = MPV_ERROR_INVALID_PARAMETER; + goto error; + } + + if (cmd_node->u.list->values[1].format != MPV_FORMAT_STRING) { + rc = MPV_ERROR_INVALID_PARAMETER; + goto error; + } + + if (cmd_node->u.list->values[2].format != MPV_FORMAT_STRING) { + rc = MPV_ERROR_INVALID_PARAMETER; + goto error; + } + + rc = mpv_set_property_string(arg->client, + cmd_node->u.list->values[1].u.string, + cmd_node->u.list->values[2].u.string); + } else if (!strcmp("observe_property", cmd)) { + if (cmd_node->u.list->num != 3) { + rc = MPV_ERROR_INVALID_PARAMETER; + goto error; + } + + if (cmd_node->u.list->values[1].format != MPV_FORMAT_INT64) { + rc = MPV_ERROR_INVALID_PARAMETER; + goto error; + } + + if (cmd_node->u.list->values[2].format != MPV_FORMAT_STRING) { + rc = MPV_ERROR_INVALID_PARAMETER; + goto error; + } + + rc = mpv_observe_property(arg->client, + cmd_node->u.list->values[1].u.int64, + cmd_node->u.list->values[2].u.string, + MPV_FORMAT_NODE); + } else if (!strcmp("observe_property_string", cmd)) { + if (cmd_node->u.list->num != 3) { + rc = MPV_ERROR_INVALID_PARAMETER; + goto error; + } + + if (cmd_node->u.list->values[1].format != MPV_FORMAT_INT64) { + rc = MPV_ERROR_INVALID_PARAMETER; + goto error; + } + + if (cmd_node->u.list->values[2].format != MPV_FORMAT_STRING) { + rc = MPV_ERROR_INVALID_PARAMETER; + goto error; + } + + rc = mpv_observe_property(arg->client, + cmd_node->u.list->values[1].u.int64, + cmd_node->u.list->values[2].u.string, + MPV_FORMAT_STRING); + } else if (!strcmp("unobserve_property", cmd)) { + if (cmd_node->u.list->num != 2) { + rc = MPV_ERROR_INVALID_PARAMETER; + goto error; + } + + if (cmd_node->u.list->values[1].format != MPV_FORMAT_INT64) { + rc = MPV_ERROR_INVALID_PARAMETER; + goto error; + } + + rc = mpv_unobserve_property(arg->client, + cmd_node->u.list->values[1].u.int64); + } else if (!strcmp("suspend", cmd)) { + mpv_suspend(arg->client); + rc = MPV_ERROR_SUCCESS; + } else if (!strcmp("resume", cmd)) { + mpv_resume(arg->client); + rc = MPV_ERROR_SUCCESS; + } else { + mpv_node result_node; + + rc = mpv_command_node(arg->client, cmd_node, &result_node); + if (rc >= 0) + mpv_node_map_add(ta_parent, &reply_node, "data", &result_node); + } + +error: + mpv_node_map_add_string(ta_parent, &reply_node, "error", mpv_error_string(rc)); + + char *output = talloc_strdup(NULL, ""); + json_write(&output, &reply_node); + output = ta_talloc_strdup_append(output, "\n"); + + talloc_free(ta_parent); + + return output; +} + +static int ipc_write(int fd, const char *buf, size_t count) +{ + while (count > 0) { + ssize_t rc = write(fd, buf, count); + if (rc <= 0) { + if (rc == 0) + return ECONNRESET; + + if (rc == EINTR) + continue; + + if (rc == EAGAIN) + return 0; + + return rc; + } + + count -= rc; + buf += rc; + } + + return 0; +} + +static void *client_thread(void *p) +{ + pthread_detach(pthread_self()); + + int rc; + + struct client_arg *arg = p; + + int pipe_fd = mpv_get_wakeup_pipe(arg->client); + if (pipe_fd < 0) { + MP_ERR(arg, "Could not get wakeup pipe\n"); + goto done; + } + + MP_INFO(arg, "Client connected\n"); + + struct pollfd fds[2] = { + {.events = POLLIN, .fd = pipe_fd}, + {.events = POLLIN, .fd = arg->client_fd}, + }; + + fcntl(arg->client_fd, F_SETFL, fcntl(arg->client_fd, F_GETFL, 0) | O_NONBLOCK); + + bstr client_msg = { talloc_strdup(NULL, ""), 0 }; + while (1) { + rc = poll(fds, 2, -1); + if (rc < 0) { + MP_ERR(arg, "Poll error\n"); + continue; + } + + if (fds[0].revents & POLLIN) { + char discard[100]; + read(pipe_fd, discard, sizeof(discard)); + + while (1) { + mpv_event *event = mpv_wait_event(arg->client, 0); + + if (event->event_id == MPV_EVENT_NONE) + break; + + if (event->event_id == MPV_EVENT_SHUTDOWN) + goto done; + + if (!arg->encode_event) + continue; + + char *event_msg = arg->encode_event(event); + if (!event_msg) { + MP_ERR(arg, "Encoding error\n"); + goto done; + } + + rc = ipc_write(arg->client_fd, event_msg, strlen(event_msg)); + talloc_free(event_msg); + if (rc < 0) { + MP_ERR(arg, "Write error\n"); + goto done; + } + } + } + + if (fds[1].revents & POLLIN) { + while (1) { + char buf[128]; + bstr append = { buf, 0 }; + + ssize_t bytes = read(arg->client_fd, buf, sizeof(buf)); + if (bytes < 0) { + if (errno == EAGAIN) + break; + + MP_ERR(arg, "Read error\n"); + goto done; + } + + if (bytes == 0) { + MP_INFO(arg, "Client disconnected\n"); + goto done; + } + + append.len = bytes; + + bstr_xappend(NULL, &client_msg, append); + + while (bstrchr(client_msg, '\n') != -1) { + bstr rest, tmp; + bstr line = bstr_getline(client_msg, &rest); + + if (arg->execute_command) { + char *reply_msg = arg->execute_command(arg, line); + if (!reply_msg) + goto command_done; + + rc = ipc_write(arg->client_fd, reply_msg, + strlen(reply_msg)); + talloc_free(reply_msg); + if (rc < 0) { + MP_ERR(arg, "Write error\n"); + talloc_free(client_msg.start); + goto done; + } + } + +command_done: + tmp = bstrdup(NULL, rest); + talloc_free(client_msg.start); + client_msg = tmp; + } + } + } + } + +done: + close(arg->client_fd); + mpv_detach_destroy(arg->client); + talloc_free(arg); + return NULL; +} + +static void ipc_start_client(struct MPContext *mpctx, int id, int fd) +{ + struct client_arg *client = talloc_ptrtype(NULL, client); + char *client_name = talloc_asprintf(client, "ipc-%d", id); + *client = (struct client_arg){ + .client = mp_new_client(mpctx->clients, client_name), + .client_fd = fd, + + .encode_event = json_encode_event, + .execute_command = json_execute_command, + }; + + client->log = mp_client_get_log(client->client); + + pthread_t client_thr; + if (pthread_create(&client_thr, NULL, client_thread, client)) { + mpv_detach_destroy(client->client); + close(client->client_fd); + talloc_free(client); + } +} + +static void *ipc_thread(void *p) +{ + int rc; + + int ipc_fd; + struct sockaddr_un ipc_un; + + struct thread_arg *arg = p; + + MP_INFO(arg->mpctx, "Starting IPC master\n"); + + ipc_fd = socket(AF_UNIX, SOCK_STREAM, 0); + if (ipc_fd < 0) { + MP_ERR(arg->mpctx, "Could not create IPC socket\n"); + goto done; + } + + size_t path_len = strlen(arg->path); + if (path_len >= sizeof(ipc_un.sun_path) - 1) { + MP_ERR(arg->mpctx, "Could not create IPC socket\n"); + goto done; + } + + ipc_un.sun_family = AF_UNIX, + strncpy(ipc_un.sun_path, arg->path, sizeof(ipc_un.sun_path)); + + unlink(ipc_un.sun_path); + + if (ipc_un.sun_path[0] == '@') { + ipc_un.sun_path[0] = '\0'; + path_len--; + } + + size_t addr_len = offsetof(struct sockaddr_un, sun_path) + 1 + path_len; + rc = bind(ipc_fd, (struct sockaddr *) &ipc_un, addr_len); + if (rc < 0) { + MP_ERR(arg->mpctx, "Could not bind IPC socket\n"); + goto done; + } + + rc = listen(ipc_fd, 10); + if (rc < 0) { + MP_ERR(arg->mpctx, "Could not listen on IPC socket\n"); + goto done; + } + + int client_num = 0; + + struct pollfd fds[2] = { + {.events = POLLIN, .fd = arg->death_pipe[0]}, + {.events = POLLIN, .fd = ipc_fd}, + }; + + while (1) { + rc = poll(fds, 2, -1); + if (rc < 0) { + MP_ERR(arg->mpctx, "Poll error\n"); + continue; + } + + if (fds[0].revents & POLLIN) + goto done; + + if (fds[1].revents & POLLIN) { + int client_fd = accept(ipc_fd, NULL, NULL); + if (client_fd < 0) { + MP_ERR(arg->mpctx, "Could not accept IPC client\n"); + goto done; + } + + ipc_start_client(arg->mpctx, client_num++, client_fd); + } + } + +done: + if (ipc_fd >= 0) + close(ipc_fd); + + close(arg->death_pipe[0]); + arg->death_pipe[0] = -1; + close(arg->death_pipe[1]); + arg->death_pipe[1] = -1; + arg->mpctx->ipc_ctx = NULL; + + talloc_free(arg); + return NULL; +} + +void mp_init_ipc(struct MPContext *mpctx) +{ + if (!mpctx->opts->ipc_path || !*mpctx->opts->ipc_path) + return; + + struct thread_arg *arg = talloc_ptrtype(NULL, arg); + *arg = (struct thread_arg){ + .mpctx = mpctx, + .path = mp_get_user_path(arg, mpctx->global, mpctx->opts->ipc_path), + }; + + if (mp_make_wakeup_pipe(arg->death_pipe) < 0) { + talloc_free(arg); + return; + } + + mpctx->ipc_ctx = arg; + + if (pthread_create(&arg->thread, NULL, ipc_thread, arg)) + talloc_free(arg); +} + +void mp_uninit_ipc(struct MPContext *mpctx) +{ + struct thread_arg *arg = mpctx->ipc_ctx; + + if (!arg) + return; + + if (arg->death_pipe[1] != -1) + write(arg->death_pipe[1], &(char){0}, 1); + pthread_join(arg->thread, NULL); + + mpctx->ipc_ctx = NULL; +} |