domo-iot/lib/src/restd_server.c

/*!
 * restd_server.c
 *
 * Copyright (c) 2015-2019, NADAL Jean-Baptiste. All rights reserved.
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or (at your option) any later version.
 *
 * This library 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
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
 * MA 02110-1301 USA
 * 
 * @Author:  NADAL Jean-Baptiste
 * @Date: 23/12/2019
 *
 */

// This is an independent project of an individual developer. Dear PVS-Studio, please check it.
// PVS-Studio Static Code Analyzer for C, C++, C#, and Java: http://www.viva64.com

/*------------------------------- INCLUDES ----------------------------------*/

#include <stdbool.h>
#include <arpa/inet.h>

#include <sys/un.h>

#include <event2/buffer.h>
#include <event2/bufferevent.h>
#include <event2/event.h>
#include <event2/listener.h>

#ifdef __linux__
#include <sys/eventfd.h>
#else
#include <sys/event.h>
#endif

#include "macro.h"
#include "restd_server.h"

/*
 * Local variables.
 */
static bool initialized = false;

/*
 * Global variables.
 */
int _restd_log_level = RESTD_LOG_WARN;


/*
 * Local functions.
 */
static int notify_loopexit(restd_server_t *server);
static void notify_cb(struct bufferevent *buffer, void *userdata);
static void *server_loop(void *instance);
static void close_server(restd_server_t *server);
static void libevent_log_cb(int severity, const char *msg);
static int set_undefined_options(restd_server_t *server);
static void listener_cb(struct evconnlistener *listener,
                        evutil_socket_t evsocket, struct sockaddr *sockaddr,
                        int socklen, void *userdata);
static restd_conn_t *conn_new(restd_server_t *server, struct bufferevent *buffer);
static void conn_reset(restd_conn_t *conn);
static void conn_free(restd_conn_t *conn);
static void conn_read_cb(struct bufferevent *buffer, void *userdata);
static void conn_write_cb(struct bufferevent *buffer, void *userdata);
static void conn_event_cb(struct bufferevent *buffer, short what, void *userdata);
static void conn_cb(restd_conn_t *conn, int event);
static int call_hooks(short event, restd_conn_t *conn);
static void *set_userdata(restd_conn_t *conn, int index, const void *userdata, restd_userdata_free_cb free_cb);
static void *get_userdata(restd_conn_t *conn, int index);

/**
 * Server option names and default values.
 */
#define RESTD_SERVER_OPTIONS {                                                 \
    {"server.port", "8888"},                                                   \
                                                                               \
    {"server.root_path", ""},                                                  \
                                                                               \
    /* Addr format IPv4="1.2.3.4", IPv6="1:2:3:4:5:6", Unix="/path" */         \
    {"server.addr", "0.0.0.0"},                                                \
                                                                               \
    {"server.backlog", "128"},                                                 \
                                                                               \
    /* Set read timeout seconds. 0 means no timeout. */                        \
    {"server.timeout", "0"},                                                   \
                                                                               \
    /* Enable or disable request pipelining, this change AD_DONE's behavior */ \
    {"server.request_pipelining", "1"},                                        \
                                                                               \
    /* Run server in a separate thread */                                      \
    {"server.thread", "0"},                                                    \
                                                                               \
    /* Collect resources after stop */                                         \
    {"server.free_on_stop", "1"},                                              \
                                                                               \
    /* End of array marker. Do not remove */                                   \
    {"", "_END_"}};

/*------------------------------- FUNCTIONS ----------------------------------*/

/**
 * Set debug output level.
 *
 * @param debug_level debug output level. 0 to disable.
 *
 * @return previous debug level.
 *
 * @note
 *  debug_level:
 *    REST_LOG_DISABLE
 *    REST_LOG_ERROR
 *    REST_LOG_WARN (default)
 *    REST_LOG_INFO
 *    REST_LOG_DEBUG
 *    REST_LOG_DEBUG2
 */
enum restd_log_e restd_log_level(enum restd_log_e log_level)
{
    int prev = _restd_log_level;
    _restd_log_level = log_level;
    return prev;
}

/**
 * Create a server object.
 */
restd_server_t *restd_server_new(void)
{
    if (initialized)
    {
        initialized = true;
    }

    restd_server_t *server = NEW_OBJECT(restd_server_t);
    if (server == NULL)
    {
        return NULL;
    }

    // Initialize instance.
    server->options = qhashtbl(0, 0);
    server->stats = qhashtbl(100, QHASHTBL_THREADSAFE);
    server->hooks = qlist(0);
    if (server->options == NULL || server->stats == NULL || server->hooks == NULL)
    {
        restd_server_free(server);
        return NULL;
    }

    DEBUG("Created a server object.");
    return server;
}

/**
 * Release server object and all the resources.
 */
void restd_server_free(restd_server_t *server)
{
    if (server == NULL)
        return;

    int thread = restd_server_get_option_int(server, "server.thread");
    if (thread && server->thread)
    {
        notify_loopexit(server);
        sleep(1);
        close_server(server);
    }

    if (server->evbase)
    {
        event_base_free(server->evbase);
    }

    if (server->options)
    {
        server->options->free(server->options);
    }
    if (server->stats)
    {
        server->stats->free(server->stats);
    }
    if (server->hooks)
    {
        qlist_t *tbl = server->hooks;
        restd_hook_t *hook;
        while ((hook = tbl->popfirst(tbl, NULL)))
        {
            if (hook->method)
                free(hook->method);
            free(hook);
        }
        server->hooks->free(server->hooks);
    }
    free(server);
    DEBUG("Server terminated.");
}

/**
 * Start server.
 *
 * @return 0 if successful, otherwise -1.
 */
int restd_server_start(restd_server_t *server)
{
    DEBUG("Starting a server.");

    // Set default options that were not set by user..
    set_undefined_options(server);

    // Hookup libevent's log message.
    if (_restd_log_level >= RESTD_LOG_DEBUG)
    {
        event_set_log_callback(libevent_log_cb);
        if (_restd_log_level >= RESTD_LOG_DEBUG2)
        {
            event_enable_debug_mode();
        }
    }

    // Parse addr
    int port = restd_server_get_option_int(server, "server.port");
    char *addr = restd_server_get_option(server, "server.addr");
    struct sockaddr *sockaddr = NULL;
    size_t sockaddr_len = 0;
    if (addr[0] == '/')
    { // Unix socket.
        struct sockaddr_un unixaddr;
        bzero((void *)&unixaddr, sizeof(struct sockaddr_un));
        if (strlen(addr) >= sizeof(unixaddr.sun_path))
        {
            errno = EINVAL;
            DEBUG("Too long unix socket name. '%s'", addr);
            return -1;
        }
        unixaddr.sun_family = AF_UNIX;
        strcpy(unixaddr.sun_path, addr); // no need of strncpy()
        sockaddr = (struct sockaddr *)&unixaddr;
        sockaddr_len = sizeof(unixaddr);
    }
    else if (strstr(addr, ":"))
    { // IPv6
        struct sockaddr_in6 ipv6addr;
        bzero((void *)&ipv6addr, sizeof(struct sockaddr_in6));
        ipv6addr.sin6_family = AF_INET6;
        ipv6addr.sin6_port = htons(port);
        evutil_inet_pton(AF_INET6, addr, &ipv6addr.sin6_addr);
        sockaddr = (struct sockaddr *)&ipv6addr;
        sockaddr_len = sizeof(ipv6addr);
    }
    else
    { // IPv4
        struct sockaddr_in ipv4addr;
        bzero((void *)&ipv4addr, sizeof(struct sockaddr_in));
        ipv4addr.sin_family = AF_INET;
        ipv4addr.sin_port = htons(port);
        ipv4addr.sin_addr.s_addr =
            (IS_EMPTY_STR(addr)) ? INADDR_ANY : inet_addr(addr);
        sockaddr = (struct sockaddr *)&ipv4addr;
        sockaddr_len = sizeof(ipv4addr);
    }

    // Bind
    if (!server->evbase)
    {
        server->evbase = event_base_new();
        if (!server->evbase)
        {
            ERROR("Failed to create a new event base.");
            return -1;
        }
    }

    // Create a eventfd for notification channel.
#ifdef __linux__
    int notifyfd = eventfd(0, 0);
#else
    int notifyfd = kqueue();
#endif
    server->notify_buffer = bufferevent_socket_new(server->evbase, notifyfd, BEV_OPT_CLOSE_ON_FREE);
    bufferevent_setcb(server->notify_buffer, NULL, notify_cb, NULL, server);

    if (!server->listener)
    {
        server->listener = evconnlistener_new_bind(
            server->evbase, listener_cb, (void *)server,
            LEV_OPT_THREADSAFE | LEV_OPT_REUSEABLE | LEV_OPT_CLOSE_ON_FREE,
            restd_server_get_option_int(server, "server.backlog"),
            sockaddr, sockaddr_len);
        if (!server->listener)
        {
            ERROR("Failed to bind on %s:%d", addr, port);
            return -1;
        }
    }

    // Listen
    INFO("Listening on %s:%d", addr, port);

    int exitstatus = 0;
    if (restd_server_get_option_int(server, "server.thread"))
    {
        DEBUG("Launching server as a thread.")
        server->thread = NEW_OBJECT(pthread_t);
        pthread_create(server->thread, NULL, &server_loop, (void *)server);
        //pthread_detach(server->thread);
    }
    else
    {
        int *retval = server_loop(server);
        exitstatus = *retval;
        free(retval);

        close_server(server);
        if (restd_server_get_option_int(server, "server.free_on_stop"))
        {
            restd_server_free(server);
        }
    }

    return exitstatus;
}

/*--------------------------------------------------------------------------*/

int restd_server_attach_event_loop(restd_server_t *server, struct event_base *ev_base)
{
    if (server == NULL)
        return -1;

    server->evbase = ev_base;
    return 0;
}

/**
 * Set server option.
 *
 * @see AD_SERVER_OPTIONS
 */
void restd_server_set_option(restd_server_t *server, const char *key, const char *value)
{
    server->options->putstr(server->options, key, value);
}

/**
 * Retrieve server option.
 */
char *restd_server_get_option(restd_server_t *server, const char *key)
{
    return server->options->getstr(server->options, key, false);
}

/**
 * Retrieve server option in integer format.
 */
int restd_server_get_option_int(restd_server_t *server, const char *key)
{
    char *value = restd_server_get_option(server, key);
    return (value) ? atoi(value) : 0;
}

/**
 * If there's no event, loopbreak or loopexit call won't work until one more
 * event arrived. So we use eventfd as a internal notification channel to let
 * server get out of the loop without waiting for an event.
 */
static int notify_loopexit(restd_server_t *server)
{
    uint64_t x = 0;
    return bufferevent_write(server->notify_buffer, &x, sizeof(uint64_t));
}

/*--------------------------------------------------------------------------*/

static void notify_cb(struct bufferevent *buffer, void *userdata)
{
    restd_server_t *server = (restd_server_t *)userdata;
    event_base_loopexit(server->evbase, NULL);
    DEBUG("Existing loop.");
}

/*--------------------------------------------------------------------------*/

static void *server_loop(void *instance)
{
    restd_server_t *server = (restd_server_t *)instance;

    int *retval = NEW_OBJECT(int);
    DEBUG("Loop start");
    event_base_loop(server->evbase, 0);
    DEBUG("Loop finished");
    *retval = (event_base_got_break(server->evbase)) ? -1 : 0;

    return retval;
}
/*--------------------------------------------------------------------------*/

static void close_server(restd_server_t *server)
{
    DEBUG("Closing server.");

    if (server->notify_buffer)
    {
        bufferevent_free(server->notify_buffer);
        server->notify_buffer = NULL;
    }

    if (server->listener)
    {
        evconnlistener_free(server->listener);
        server->listener = NULL;
    }

    if (server->thread)
    {
        void *retval = NULL;
        DEBUG("Waiting server's last loop to finish.");
        pthread_join(*(server->thread), &retval);
        free(retval);
        free(server->thread);
        server->thread = NULL;
    }
    INFO("Server closed.");
}

/*--------------------------------------------------------------------------*/

static void libevent_log_cb(int severity, const char *msg)
{
    switch (severity)
    {
    case _EVENT_LOG_MSG:
    {
        INFO("%s", msg);
        break;
    }
    case _EVENT_LOG_WARN:
    {
        WARN("%s", msg);
        break;
    }
    case _EVENT_LOG_ERR:
    {
        ERROR("%s", msg);
        break;
    }
    default:
    {
        DEBUG("%s", msg);
        break;
    }
    }
}

/*--------------------------------------------------------------------------*/

// Set default options that were not set by user..
static int set_undefined_options(restd_server_t *server)
{
    int newentries = 0;
    char *default_options[][2] = RESTD_SERVER_OPTIONS;
    for (int i = 0; !IS_EMPTY_STR(default_options[i][0]); i++)
    {
        if (!restd_server_get_option(server, default_options[i][0]))
        {
            restd_server_set_option(server, default_options[i][0], default_options[i][1]);
            newentries++;
        }
        DEBUG("%s=%s", default_options[i][0], restd_server_get_option(server, default_options[i][0]));
    }
    return newentries;
}

/*--------------------------------------------------------------------------*/

static void listener_cb(struct evconnlistener *listener, evutil_socket_t socket,
                        struct sockaddr *sockaddr, int socklen, void *userdata)
{
    DEBUG("New connection.");
    restd_server_t *server = (restd_server_t *)userdata;

    // Create a new buffer.
    struct bufferevent *buffer = NULL;

    buffer = bufferevent_socket_new(server->evbase, socket, BEV_OPT_CLOSE_ON_FREE);

    if (buffer == NULL)
        goto error;

    // Set read timeout.
    int timeout = restd_server_get_option_int(server, "server.timeout");
    if (timeout > 0)
    {
        struct timeval tm;
        bzero((void *)&tm, sizeof(struct timeval));
        tm.tv_sec = timeout;
        bufferevent_set_timeouts(buffer, &tm, NULL);
    }

    // Create a connection.
    void *conn = conn_new(server, buffer);
    if (!conn)
        goto error;

    return;

error:
    if (buffer)
        bufferevent_free(buffer);
    ERROR("Failed to create a connection handler.");
    event_base_loopbreak(server->evbase);
    server->errcode = ENOMEM;
}

/*--------------------------------------------------------------------------*/

static restd_conn_t *conn_new(restd_server_t *server, struct bufferevent *buffer)
{
    if (server == NULL || buffer == NULL)
    {
        return NULL;
    }

    // Create a new connection container.
    restd_conn_t *conn = NEW_OBJECT(restd_conn_t);
    if (conn == NULL)
        return NULL;

    // Initialize with default values.
    conn->server = server;
    conn->buffer = buffer;
    conn->in = bufferevent_get_input(buffer);
    conn->out = bufferevent_get_output(buffer);
    conn_reset(conn);

    // Bind callback
    bufferevent_setcb(buffer, conn_read_cb, conn_write_cb, conn_event_cb, (void *)conn);
    bufferevent_setwatermark(buffer, EV_WRITE, 0, 0);
    bufferevent_enable(buffer, EV_WRITE);
    bufferevent_enable(buffer, EV_READ);

    // Run callbacks with AD_EVENT_INIT event.
    conn->status = call_hooks(RESTD_EVENT_INIT | RESTD_EVENT_WRITE, conn);

    return conn;
}

/*--------------------------------------------------------------------------*/

static void conn_reset(restd_conn_t *conn)
{
    conn->status = RESTD_OK;

    for (int i = 0; i < RESTD_NUM_USERDATA; i++)
    {
        if (conn->userdata[i])
        {
            if (conn->userdata_free_cb[i] != NULL)
            {
                conn->userdata_free_cb[i](conn, conn->userdata[i]);
            }
            else
            {
                WARN("Found unreleased userdata.");
            }
            conn->userdata[i] = NULL;
        }
    }

    if (conn->method)
    {
        free(conn->method);
        conn->method = NULL;
    }
}

/*--------------------------------------------------------------------------*/

static void conn_free(restd_conn_t *conn)
{
    if (conn)
    {
        if (conn->status != RESTD_CLOSE)
        {
            call_hooks(RESTD_EVENT_CLOSE | RESTD_EVENT_SHUTDOWN, conn);
        }
        conn_reset(conn);
        if (conn->buffer)
        {

            bufferevent_free(conn->buffer);
        }
        free(conn);
    }
}

/*--------------------------------------------------------------------------*/

#define DRAIN_EVBUFFER(b) evbuffer_drain(b, evbuffer_get_length(b))
static void conn_read_cb(struct bufferevent *buffer, void *userdata)
{
    DEBUG("read_cb");
    restd_conn_t *conn = userdata;
    conn_cb(conn, RESTD_EVENT_READ);
}

/*--------------------------------------------------------------------------*/

static void conn_write_cb(struct bufferevent *buffer, void *userdata)
{
    DEBUG("write_cb");
    restd_conn_t *conn = userdata;
    conn_cb(conn, RESTD_EVENT_WRITE);
}

/*--------------------------------------------------------------------------*/

static void conn_event_cb(struct bufferevent *buffer, short what, void *userdata)
{
    DEBUG("event_cb 0x%x", what);
    restd_conn_t *conn = userdata;

    if (what & BEV_EVENT_EOF || what & BEV_EVENT_ERROR || what & BEV_EVENT_TIMEOUT)
    {
        conn->status = RESTD_CLOSE;
        conn_cb(conn, RESTD_EVENT_CLOSE | ((what & BEV_EVENT_TIMEOUT) ? RESTD_EVENT_TIMEOUT : 0));
    }
}

/*--------------------------------------------------------------------------*/

static void conn_cb(restd_conn_t *conn, int event)
{
    DEBUG("conn_cb: status:0x%x, event:0x%x", conn->status, event)
    if (conn->status == RESTD_OK || conn->status == RESTD_TAKEOVER)
    {
        int status = call_hooks(event, conn);
        // Update status only when it's higher then before.
        if (!(conn->status == RESTD_CLOSE || (conn->status == RESTD_DONE && conn->status >= status)))
        {
            conn->status = status;
        }
    }

    if (conn->status == RESTD_DONE)
    {
        if (restd_server_get_option_int(conn->server, "server.request_pipelining"))
        {
            call_hooks(RESTD_EVENT_CLOSE, conn);
            conn_reset(conn);
            call_hooks(RESTD_EVENT_INIT, conn);
        }
        else
        {
            // Do nothing but drain input buffer.
            if (event == RESTD_EVENT_READ)
            {
                DEBUG("Draining in-buffer. %d", conn->status);
                DRAIN_EVBUFFER(conn->in);
            }
        }
        return;
    }
    else if (conn->status == RESTD_CLOSE)
    {
        if (evbuffer_get_length(conn->out) <= 0)
        {
            int newevent = (event & RESTD_EVENT_CLOSE) ? event : RESTD_EVENT_CLOSE;
            call_hooks(newevent, conn);
            conn_free(conn);
            DEBUG("Connection closed.");
            return;
        }
    }
}

/*--------------------------------------------------------------------------*/

void restd_server_register_request_handler(restd_server_t *server, restd_callback cb)
{
    server->request_handler = cb;
}

/*--------------------------------------------------------------------------*/

void restd_server_register_error_handler(restd_server_t *server, restd_callback cb)
{
    server->error_handler = cb;    
}

/*--------------------------------------------------------------------------*/

/**
 * Register user hook.
 */
void restd_server_register_hook(restd_server_t *server, restd_callback cb, void *userdata)
{
    restd_server_register_hook_on_method(server, NULL, cb, userdata);
}

/*--------------------------------------------------------------------------*/

/**
 * Register user hook on method name.
 */
void restd_server_register_hook_on_method(restd_server_t *server, const char *method, restd_callback cb, void *userdata)
{
    restd_server_register_hook_on_path(server, method, NULL, cb, userdata);
}

/*--------------------------------------------------------------------------*/

void restd_server_register_hook_on_path(restd_server_t *server, const char *method, const char *path, restd_callback cb, void *userdata)
{
    restd_hook_t hook;
    bzero((void *)&hook, sizeof(restd_hook_t));
    hook.method = (method) ? strdup(method) : NULL;
    hook.path = (path) ? strdup(path) : NULL;
    hook.cb = cb;
    hook.userdata = userdata;

    server->hooks->addlast(server->hooks, (void *)&hook, sizeof(restd_hook_t));
}

/*--------------------------------------------------------------------------*/

static int call_hooks(short event, restd_conn_t *conn)
{
    DEBUG("call_hooks: event 0x%x", event);
    qlist_t *hooks = conn->server->hooks;

    qlist_obj_t obj;
    bzero((void *)&obj, sizeof(qlist_obj_t));
    while (hooks->getnext(hooks, &obj, false) == true)
    {
        restd_hook_t *hook = (restd_hook_t *)obj.data;
        if (hook->cb)
        {
            if (hook->method && conn->method && strcmp(hook->method, conn->method))
            {
                continue;
            }
            int status = hook->cb(event, conn, hook->userdata);
            if (status != RESTD_OK)
            {
                return status;
            }
        }
    }
    return RESTD_OK;
}

/*--------------------------------------------------------------------------*/

static void *set_userdata(restd_conn_t *conn, int index, const void *userdata, restd_userdata_free_cb free_cb)
{
    void *prev = conn->userdata;
    conn->userdata[index] = (void *)userdata;
    conn->userdata_free_cb[index] = free_cb;
    return prev;
}

/*--------------------------------------------------------------------------*/

static void *get_userdata(restd_conn_t *conn, int index)
{
    return conn->userdata[index];
}

/*--------------------------------------------------------------------------*/

/**
 * Set extra userdata into the connection.
 *
 * @return previous userdata;
 *
 * @note
 *   Extra userdata is for default protocol handler such as ad_http_handler to
 *   provide higher abstraction. End users should always use only ad_conn_set_userdata()
 *   to avoid any conflict with default handlers.
 */
void *restd_conn_set_extra(restd_conn_t *conn, const void *extra, restd_userdata_free_cb free_cb)
{
    return set_userdata(conn, 1, extra, free_cb);
}

/**
 * Get extra userdata attached in this connection.
 */
void *restd_conn_get_extra(restd_conn_t *conn)
{
    return get_userdata(conn, 1);
}

/**
 * Set method name on this connection.
 *
 * Once the method name is set, hooks registered by ad_server_register_hook_on_method()
 * will be called if method name matches with the registered name.
 *
 * @see ad_server_register_hook_on_method()
 */
void restd_conn_set_method(restd_conn_t *conn, char *method)
{
    char *prev = conn->method;
    conn->method = (method != NULL) ? strdup(method) : NULL;
    if (prev)
    {
        free(prev);
    }
}