redict/deps/hiredict/adapters/libuv.h

/*
 * Copyright (c) 2009-2011, Salvatore Sanfilippo <antirez at gmail dot com>
 *
 * SPDX-FileCopyrightText: 2024 Hiredict Contributors
 * SPDX-FileCopyrightText: 2024 Salvatore Sanfilippo <antirez at gmail dot com>
 *
 * SPDX-License-Identifier: BSD-3-Clause
 * SPDX-License-Identifier: LGPL-3.0-or-later
 *
 */

#ifndef __HIREDICT_LIBUV_H__
#define __HIREDICT_LIBUV_H__
#include <stdlib.h>
#include <uv.h>
#include "../hiredict.h"
#include "../async.h"
#include <string.h>

typedef struct redictLibuvEvents {
    redictAsyncContext* context;
    uv_poll_t          handle;
    uv_timer_t         timer;
    int                events;
} redictLibuvEvents;


static void redictLibuvPoll(uv_poll_t* handle, int status, int events) {
    redictLibuvEvents* p = (redictLibuvEvents*)handle->data;
    int ev = (status ? p->events : events);

    if (p->context != NULL && (ev & UV_READABLE)) {
        redictAsyncHandleRead(p->context);
    }
    if (p->context != NULL && (ev & UV_WRITABLE)) {
        redictAsyncHandleWrite(p->context);
    }
}


static void redictLibuvAddRead(void *privdata) {
    redictLibuvEvents* p = (redictLibuvEvents*)privdata;

    if (p->events & UV_READABLE) {
        return;
    }

    p->events |= UV_READABLE;

    uv_poll_start(&p->handle, p->events, redictLibuvPoll);
}


static void redictLibuvDelRead(void *privdata) {
    redictLibuvEvents* p = (redictLibuvEvents*)privdata;

    p->events &= ~UV_READABLE;

    if (p->events) {
        uv_poll_start(&p->handle, p->events, redictLibuvPoll);
    } else {
        uv_poll_stop(&p->handle);
    }
}


static void redictLibuvAddWrite(void *privdata) {
    redictLibuvEvents* p = (redictLibuvEvents*)privdata;

    if (p->events & UV_WRITABLE) {
        return;
    }

    p->events |= UV_WRITABLE;

    uv_poll_start(&p->handle, p->events, redictLibuvPoll);
}


static void redictLibuvDelWrite(void *privdata) {
    redictLibuvEvents* p = (redictLibuvEvents*)privdata;

    p->events &= ~UV_WRITABLE;

    if (p->events) {
        uv_poll_start(&p->handle, p->events, redictLibuvPoll);
    } else {
        uv_poll_stop(&p->handle);
    }
}

static void on_timer_close(uv_handle_t *handle) {
    redictLibuvEvents* p = (redictLibuvEvents*)handle->data;
    p->timer.data = NULL;
    if (!p->handle.data) {
        // both timer and handle are closed
        hi_free(p);
    }
    // else, wait for `on_handle_close`
}

static void on_handle_close(uv_handle_t *handle) {
    redictLibuvEvents* p = (redictLibuvEvents*)handle->data;
    p->handle.data = NULL;
    if (!p->timer.data) {
        // timer never started, or timer already destroyed
        hi_free(p);
    }
    // else, wait for `on_timer_close`
}

// libuv removed `status` parameter since v0.11.23
// see: https://github.com/libuv/libuv/blob/v0.11.23/include/uv.h
#if (UV_VERSION_MAJOR == 0 && UV_VERSION_MINOR < 11) || \
    (UV_VERSION_MAJOR == 0 && UV_VERSION_MINOR == 11 && UV_VERSION_PATCH < 23)
static void redictLibuvTimeout(uv_timer_t *timer, int status) {
    (void)status; // unused
#else
static void redictLibuvTimeout(uv_timer_t *timer) {
#endif
    redictLibuvEvents *e = (redictLibuvEvents*)timer->data;
    redictAsyncHandleTimeout(e->context);
}

static void redictLibuvSetTimeout(void *privdata, struct timeval tv) {
    redictLibuvEvents* p = (redictLibuvEvents*)privdata;

    uint64_t millsec = tv.tv_sec * 1000 + tv.tv_usec / 1000.0;
    if (!p->timer.data) {
        // timer is uninitialized
        if (uv_timer_init(p->handle.loop, &p->timer) != 0) {
            return;
        }
        p->timer.data = p;
    }
    // updates the timeout if the timer has already started
    // or start the timer
    uv_timer_start(&p->timer, redictLibuvTimeout, millsec, 0);
}

static void redictLibuvCleanup(void *privdata) {
    redictLibuvEvents* p = (redictLibuvEvents*)privdata;

    p->context = NULL; // indicate that context might no longer exist
    if (p->timer.data) {
        uv_close((uv_handle_t*)&p->timer, on_timer_close);
    }
    uv_close((uv_handle_t*)&p->handle, on_handle_close);
}


static int redictLibuvAttach(redictAsyncContext* ac, uv_loop_t* loop) {
    redictContext *c = &(ac->c);

    if (ac->ev.data != NULL) {
        return REDICT_ERR;
    }

    ac->ev.addRead        = redictLibuvAddRead;
    ac->ev.delRead        = redictLibuvDelRead;
    ac->ev.addWrite       = redictLibuvAddWrite;
    ac->ev.delWrite       = redictLibuvDelWrite;
    ac->ev.cleanup        = redictLibuvCleanup;
    ac->ev.scheduleTimer  = redictLibuvSetTimeout;

    redictLibuvEvents* p = (redictLibuvEvents*)hi_malloc(sizeof(*p));
    if (p == NULL)
        return REDICT_ERR;

    memset(p, 0, sizeof(*p));

    if (uv_poll_init_socket(loop, &p->handle, c->fd) != 0) {
        hi_free(p);
        return REDICT_ERR;
    }

    ac->ev.data    = p;
    p->handle.data = p;
    p->context     = ac;

    return REDICT_OK;
}
#endif