path: root/propagate/src/listener.c

#include <stdlib.h>
#include <stdio.h>
#include <unistd.h>
#include <errno.h>

#include <sys/types.h>
#include <sys/socket.h>
#include <sys/un.h>
#include <signal.h>

#include <poll.h>

#include "pg.h"

#define SOCK_BASENAME "/tmp/propagate_sock"
#define SOCK_NAME_MAXLEN 255

static int listener_start(struct listener *);
static int unix_start(struct listener *);
static int inet_start(struct listener *);

int
listener_add(int type, char *path) {
    struct listener *l;

    if (listeners_count >= LISTENERS_MAX) {
        log_warn("listeners max reached");
        return -1;
    }
    l = xcalloc(1, sizeof(struct listener));
    if (!l)
        return -1;

    l->type = type;
    switch (type) {
    case LISTENER_UNIX:
        l->sock_unix.path = path;
        break;
    case LISTENER_INET:
        log_warn("listener_add: XXX inet not implemented yet");
        break;
    }
    LIST_INIT(&(l->conns));
    l->conns_count = 0;

    LIST_INSERT_HEAD(&listeners, l, entry);
    listeners_count++;

    listener_start(l);

    return 0;
}

struct listener *
listener_find(int fd) {
    struct listener *l;

    LIST_FOREACH(l, &listeners, entry) {
        if (l->sock == fd)
            return l;
    }
    return NULL;
}

int
listener_conn_add(struct listener *l, int fd) {
    struct conn *c;

    if (l->conns_count >= LISTENER_CONN_MAX) {
        log_warn("conn max reached");
        return -1;
    }
    c = xcalloc(1, sizeof(struct conn));
    if (!c)
        return -1;

    c->listener = l;
    c->fd = fd;
    c->state = CONN_OPEN;

    LIST_INSERT_HEAD(&(l->conns), c, entry);
    l->conns_count++;

    setnonblock(fd);

    return 0;
}

void
listener_conn_del(struct conn *c) {
    close(c->fd);
    if (c->exec.cmd)
        listener_conn_exec_kill(c);
    LIST_REMOVE(c, entry);
    c->listener->conns_count--;
    free(c);
}

void
listener_conn_move(struct conn *oldc, struct conn *newc) {
    memcpy(newc, oldc, sizeof(struct conn));
    LIST_REMOVE(oldc, entry);
    newc->listener->conns_count--;
    free(oldc);
}

struct conn *
listener_conn_find(int fd) {
    struct listener *l;
    struct conn *c;

    LIST_FOREACH(l, &listeners, entry) {
        LIST_FOREACH(c, &l->conns, entry) {
            if (c->fd == fd)
                return c;
        }
    }
    return NULL;
}

struct conn *
listener_conn_find_orig(char orig) {
    struct listener *l;
    struct conn *c;

    LIST_FOREACH(l, &listeners, entry) {
        LIST_FOREACH(c, &l->conns, entry) {
            if (c->orig == orig)
                return c;
        }
    }
    return NULL;

}

/* XXX indicate that we are only looking in input pipe */
struct conn *
listener_conn_find_exec(int fd) {
    struct listener *l;
    struct conn *c;

    LIST_FOREACH(l, &listeners, entry) {
        LIST_FOREACH(c, &l->conns, entry) {
            if (c->exec.fd[0] == fd)
                return c;
        }
    }
    return NULL;
}

int
listener_conn_exec(struct conn *c, char *cmd, char **argv) {
    int pid;

    log_debug("listener_conn_exec %s", cmd);

    if (c->exec.cmd) {
        log_warn("%c is already executing a command", c->orig);
        return -1;
    }
    c->exec.cmd = cmd;
    c->exec.argv = argv;

    pid = execpipe(cmd, argv, c->exec.fd);
    if (pid < 0)
        return -1;
    c->exec.pid = pid;

    // XXX if async then writebuf ?
    // XXX if err then MSG_ERR

    return 0;
}

void
listener_conn_exec_kill(struct conn *c) {
    int i;

    if (c->exec.cmd) {
        free(c->exec.cmd);
        c->exec.cmd = NULL;
    }
    if (c->exec.argv) {
        for (i=0; c->exec.argv[i] != NULL; i++)
            free(c->exec.argv[i]);
        free(c->exec.argv);
        c->exec.argv = NULL;
    }
    kill(c->exec.pid, SIGTERM);
    close(c->exec.fd[0]);
    close(c->exec.fd[1]);
    if (c->exec.async_writebuf) {
        free(c->exec.async_writebuf);
        c->exec.async_writebuf = NULL;
    }
}


int
listener_conn_exec_bufferize(struct conn *c, int fd) {
    uint8_t buf[BUFMAX];

    log_warn("XXX conn exec bufferize not implemented");
    read(fd, buf, sizeof(buf));
    return -1;
}

static int
listener_start(struct listener *l) {
    switch (l->type) {
    case LISTENER_UNIX:
        if (unix_start(l) < 0)
            log_warn("listener socket failed to start");
        break;
    case LISTENER_INET:
        if (inet_start(l) < 0)
            log_warn("listener inet failed to start");
        break;
    }
    return 0;
}

static int
unix_start(struct listener *l) {
    int s, len;
    struct sockaddr_un local;
    char *path;

    path = xmalloc(sizeof(SOCK_NAME_MAXLEN));
    if (!path)
        return -1;
    snprintf(path, SOCK_NAME_MAXLEN, "%s", l->sock_unix.path);

    /* XXX monitor death of the socket, to restart it */

    if ((s = socket(AF_UNIX, SOCK_STREAM, 0)) == -1) {
        log_warn("unix_start: socket %s: %s", path, strerror(errno));
        free(path);
        return -1;
    }

    memset(&local, 0, sizeof(struct sockaddr_un));
    local.sun_family = AF_UNIX;
    strncpy(local.sun_path, path, sizeof(local.sun_path));
    unlink(local.sun_path);
    len = strlen(local.sun_path) + sizeof(local.sun_family);
    if (bind(s, (struct sockaddr *)&local, len) == -1) {
        log_warn("unix_start: bind %s: %s", path, strerror(errno));
        free(path);
        return -1;
    }

    if (listen(s, 5) == -1) {
        log_warn("unix_start: listen %s: %s", path, strerror(errno));
        free(path);
        return -1;
    }

    // XXX setnonblock(s);

    l->sock = s;
    l->sock_unix.path = path;

    return 0;
}

static int
inet_start(struct listener *l) {
    log_warn("inet_start: XXX not implemented yet");
    return -1;
}