src/ipc/ipc-connection.c

	ipc-connection.c revision e93184a9055c2530366dfe617e07199603c399dd
/* Copyright (c) 2011-2016 Dovecot authors, see the included COPYING file */

#include "lib.h"
#include "array.h"
#include "ioloop.h"
#include "istream.h"
#include "ostream.h"
#include "llist.h"
#include "master-service.h"
#include "ipc-group.h"
#include "ipc-connection.h"

#include <unistd.h>

#define IPC_SERVER_PROTOCOL_MAJOR_VERSION 1
#define IPC_SERVER_PROTOCOL_MINOR_VERSION 0

#define IPC_SERVER_HANDSHAKE "VERSION\tipc-proxy\t1\t0\n"

static unsigned int connection_id_counter;
static void ipc_connection_cmd_free(struct ipc_connection_cmd **cmd);

static void ipc_connection_cmd_free(struct ipc_connection_cmd **_cmd)
{
    struct ipc_connection_cmd *cmd = *_cmd;
    struct ipc_connection_cmd **cmds;
    unsigned int i, count;

    *_cmd = NULL;

    cmds = array_get_modifiable(&cmd->conn->cmds, &count);
    for (i = 0; i < count; i++) {
        if (cmds[i] == cmd) {
            array_delete(&cmd->conn->cmds, i, 1);
            break;
        }
    }
    if (cmd->callback != NULL) {
        cmd->callback(IPC_CMD_STATUS_ERROR,
                  "Connection to server died", cmd->context);
    }
    i_free(cmd);
}

static struct ipc_connection_cmd *
ipc_connection_cmd_find(struct ipc_connection *conn, unsigned int tag)
{
    struct ipc_connection_cmd *const *cmdp;

    array_foreach(&conn->cmds, cmdp) {
        if ((*cmdp)->tag == tag)
            return *cmdp;
    }
    return NULL;
}

static int
ipc_connection_input_line(struct ipc_connection *conn, char *line)
{
    struct ipc_connection_cmd *cmd;
    unsigned int tag;
    enum ipc_cmd_status status;
    char *data;

    /* <tag> [:+-]<data> */
    data = strchr(line, '\t');
    if (data == NULL)
        return -1;

    *data++ = '\0';
    if (str_to_uint(line, &tag) < 0)
        return -1;

    switch (data[0]) {
    case ':':
        status = IPC_CMD_STATUS_REPLY;
        break;
    case '+':
        status = IPC_CMD_STATUS_OK;
        break;
    case '-':
        status = IPC_CMD_STATUS_ERROR;
        break;
    default:
        return -1;
    }
    data++;

    cmd = ipc_connection_cmd_find(conn, tag);
    if (cmd == NULL) {
        i_error("IPC server: Input for unexpected command tag %u", tag);
        return 0;
    }
    cmd->callback(status, data, cmd->context);
    if (status != IPC_CMD_STATUS_REPLY) {
        cmd->callback = NULL;
        ipc_connection_cmd_free(&cmd);
    }
    return 0;
}

static void ipc_connection_input(struct ipc_connection *conn)
{
    const char *const *args;
    char *line;
    int ret;

    if (i_stream_read(conn->input) < 0) {
        ipc_connection_destroy(&conn);
        return;
    }

    if (!conn->version_received) {
        if ((line = i_stream_next_line(conn->input)) == NULL)
            return;

        if (!version_string_verify(line, "ipc-server",
                IPC_SERVER_PROTOCOL_MAJOR_VERSION)) {
            i_error("IPC server not compatible with this server "
                "(mixed old and new binaries?)");
            ipc_connection_destroy(&conn);
            return;
        }
        conn->version_received = TRUE;
    }
    if (!conn->handshake_received) {
        if ((line = i_stream_next_line(conn->input)) == NULL)
            return;

        args = t_strsplit_tab(line);
        if (str_array_length(args) < 3 ||
            strcmp(args[0], "HANDSHAKE") != 0) {
            i_error("IPC server sent invalid handshake");
            ipc_connection_destroy(&conn);
            return;
        }
        if (ipc_group_update_name(conn->group, args[1]) < 0) {
            i_error("IPC server named itself unexpectedly: %s "
                "(existing ones were %s)",
                args[1], conn->group->name);
            ipc_connection_destroy(&conn);
            return;
        }
        if (str_to_pid(args[2], &conn->pid) < 0) {
            i_error("IPC server gave broken PID: %s", args[2]);
            ipc_connection_destroy(&conn);
            return;
        }
        conn->handshake_received = TRUE;
    }

    while ((line = i_stream_next_line(conn->input)) != NULL) {
        T_BEGIN {
            ret = ipc_connection_input_line(conn, line);
        } T_END;
        if (ret < 0) {
            i_error("Invalid input from IPC server '%s': %s",
                conn->group->name, line);
            ipc_connection_destroy(&conn);
            break;
        }
    }
}

struct ipc_connection *ipc_connection_create(int listen_fd, int fd)
{
    struct ipc_connection *conn;

    conn = i_new(struct ipc_connection, 1);
    conn->group = ipc_group_lookup_listen_fd(listen_fd);
    if (conn->group == NULL)
        conn->group = ipc_group_alloc(listen_fd);
    conn->id = ++connection_id_counter;
    if (conn->id == 0)
        conn->id = ++connection_id_counter;
    conn->fd = fd;
    conn->io = io_add(fd, IO_READ, ipc_connection_input, conn);
    conn->input = i_stream_create_fd(fd, (size_t)-1);
    conn->output = o_stream_create_fd(fd, (size_t)-1);
    o_stream_set_no_error_handling(conn->output, TRUE);
    i_array_init(&conn->cmds, 8);
    o_stream_nsend_str(conn->output, IPC_SERVER_HANDSHAKE);

    DLLIST_PREPEND(&conn->group->connections, conn);
    return conn;
}

void ipc_connection_destroy(struct ipc_connection **_conn)
{
    struct ipc_connection *conn = *_conn;
    struct ipc_connection_cmd *const *cmdp, *cmd;

    *_conn = NULL;

    DLLIST_REMOVE(&conn->group->connections, conn);

    while (array_count(&conn->cmds) > 0) {
        cmdp = array_idx(&conn->cmds, 0);
        cmd = *cmdp;

        ipc_connection_cmd_free(&cmd);
    }
    array_free(&conn->cmds);

    io_remove(&conn->io);
    i_stream_destroy(&conn->input);
    o_stream_destroy(&conn->output);
    if (close(conn->fd) < 0)
        i_error("close(ipc connection) failed: %m");
    i_free(conn);

    master_service_client_connection_destroyed(master_service);
}

struct ipc_connection *
ipc_connection_lookup_id(struct ipc_group *group, unsigned int id)
{
    struct ipc_connection *conn;

    for (conn = group->connections; conn != NULL; conn = conn->next) {
        if (conn->id == id)
            return conn;
    }
    return NULL;
}

void ipc_connection_cmd(struct ipc_connection *conn, const char *cmd,
            ipc_cmd_callback_t *callback, void *context)
{
    struct ipc_connection_cmd *ipc_cmd;

    ipc_cmd = i_new(struct ipc_connection_cmd, 1);
    ipc_cmd->tag = ++conn->cmd_tag_counter;
    ipc_cmd->conn = conn;
    ipc_cmd->callback = callback;
    ipc_cmd->context = context;
    array_append(&conn->cmds, &ipc_cmd, 1);

    T_BEGIN {
        o_stream_nsend_str(conn->output,
            t_strdup_printf("%u\t%s\n", ipc_cmd->tag, cmd));
    } T_END;
}