libevent.h

Go to the documentation of this file.

// Copyright 2016-2018 The NATS Authors
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
 
#ifndef LIBEVENT_H_
#define LIBEVENT_H_
 
#ifdef __cplusplus
extern "C" {
#endif
 
#include <event.h>
#include <event2/thread.h>
#include "../nats.h"
 
typedef struct
{
    natsConnection      *nc;
    struct event_base   *loop;
    struct event        *read;
    struct event        *write;
    struct event        *keepActive;
    natsSock            socketToClose;
 
} natsLibeventEvents;
 
// Forward declarations
natsStatus natsLibevent_Read(void *userData, bool add);
natsStatus natsLibevent_Detach(void *userData);
 
void
natsLibevent_Init(void)
{
#if _WIN32
    evthread_use_windows_threads();
#else
    evthread_use_pthreads();
#endif
}
 
static void
natsLibevent_ProcessEvent(evutil_socket_t fd, short event, void *arg)
{
    natsLibeventEvents *nle = (natsLibeventEvents*) arg;
 
    if (event & EV_READ)
        natsConnection_ProcessReadEvent(nle->nc);
 
    if (event & EV_WRITE)
        natsConnection_ProcessWriteEvent(nle->nc);
}
 
static void
keepAliveCb(evutil_socket_t fd, short flags, void * arg)
{
    // do nothing...
}
 
static void
natsLibeventEvents_free(natsLibeventEvents *nle, bool processDetachedEvents)
{
    if (nle->read != NULL)
        event_free(nle->read);
    if (nle->write != NULL)
        event_free(nle->write);
    if (nle->keepActive != NULL)
    {
        event_active(nle->keepActive, 0, 0);
        event_free(nle->keepActive);
    }
    if (processDetachedEvents)
        natsConnection_ProcessDetachedEvent(nle->nc);
    free(nle);
}
 
// This callback is invoked from the event loop thread and will free the
// `natsLibeventEvents` object.
static void
_freeCb(evutil_socket_t ignoredSocket, short ignoredEvent, void *arg)
{
    natsLibeventEvents *nle = (natsLibeventEvents*) arg;
    natsLibeventEvents_free(nle, true);
}
 
natsStatus
natsLibevent_Attach(void **userData, void *loop, natsConnection *nc, natsSock socket)
{
    struct event_base   *libeventLoop = (struct event_base*) loop;
    natsLibeventEvents  *nle          = (natsLibeventEvents*) (*userData);
    natsStatus          s             = NATS_OK;
    bool                created       = false;
 
    // This is the first attach (when reconnecting, nle will be non-NULL).
    if (nle == NULL)
    {
        nle = (natsLibeventEvents*) calloc(1, sizeof(natsLibeventEvents));
        if (nle == NULL)
            return NATS_NO_MEMORY;
 
        // Indicate that we have created the object here (in case we get a failure).
        created   = true;
        nle->nc   = nc;
        nle->loop = libeventLoop;
 
        nle->keepActive = event_new(nle->loop, -1, EV_PERSIST, keepAliveCb, NULL);
        if (nle->keepActive == NULL)
            s = NATS_NO_MEMORY;
 
        if (s == NATS_OK)
        {
            struct timeval timeout;
 
            timeout.tv_sec = 100000;
            timeout.tv_usec = 0;
 
            if (event_add(nle->keepActive, &timeout) != 0)
                s = NATS_ERR;
        }
    }
    else
    {
        if (nle->read != NULL)
        {
            event_free(nle->read);
            nle->read = NULL;
        }
        if (nle->write != NULL)
        {
            event_free(nle->write);
            nle->write = NULL;
        }
    }
 
    if (s == NATS_OK)
    {
        nle->read = event_new(nle->loop, socket, EV_READ|EV_PERSIST,
                              natsLibevent_ProcessEvent, (void*) nle);
        natsLibevent_Read((void*) nle, true);
 
        nle->write = event_new(nle->loop, socket, EV_WRITE|EV_PERSIST,
                               natsLibevent_ProcessEvent, (void*) nle);
    }
 
    if (s == NATS_OK)
        *userData = (void*) nle;
    else if (created)
        natsLibeventEvents_free(nle, false);
 
    return s;
}
 
static void
_closeCb(evutil_socket_t ignoredSocket, short ignoredEvent, void *arg)
{
    natsLibeventEvents  *nle = (natsLibeventEvents*) arg;
 
    // We have stopped polling for the "READ" event and are now in the
    // event loop thread and invoke this so that the NATS C client
    // library can proceed with the close of the socket/connection.
    natsConnection_ProcessCloseEvent(&(nle->socketToClose));
}
 
natsStatus
natsLibevent_Read(void *userData, bool add)
{
    natsLibeventEvents  *nle = (natsLibeventEvents*) userData;
    int                 res;
 
    if (add)
        res = event_add(nle->read, NULL);
    else
    {
        int socket = event_get_fd(nle->read);
        res = event_del_noblock(nle->read);
        if (res == 0)
        {
            // This will schedule a one-time event that guarantees that the
            // callback `_closeCb` will be invoked from the event loop thread.
            nle->socketToClose = (natsSock) socket;
            res = event_base_once(nle->loop, -1, EV_TIMEOUT, _closeCb, (void*) nle, NULL);
        }
    }
 
    return (res == 0 ? NATS_OK : NATS_ERR);
}
 
natsStatus
natsLibevent_Write(void *userData, bool add)
{
    natsLibeventEvents  *nle = (natsLibeventEvents*) userData;
    int                 res;
 
    if (add)
        res = event_add(nle->write, NULL);
    else
        res = event_del_noblock(nle->write);
 
    return (res == 0 ? NATS_OK : NATS_ERR);
}
 
natsStatus
natsLibevent_Detach(void *userData)
{
    natsLibeventEvents *nle = (natsLibeventEvents*) userData;
 
    // This will schedule a one-time event that guarantees that the
    // callback `_freeCb` will be invoked from the event loop thread.
    int res = event_base_once(nle->loop, -1, EV_TIMEOUT, _freeCb, (void*) nle, NULL);
 
    return (res == 0 ? NATS_OK : NATS_ERR);
}
 
 
#ifdef __cplusplus
}
#endif
 
#endif /* LIBEVENT_H_ */