lmtp2nntp-1.4.1/lib_l2/l2_ch_buffer.c

/*
**  OSSP l2 - Flexible Logging
**  Copyright (c) 2001-2005 Cable & Wireless <http://www.cw.com/>
**  Copyright (c) 2001-2005 The OSSP Project <http://www.ossp.org/>
**  Copyright (c) 2001-2005 Ralf S. Engelschall <rse@engelschall.com>
**
**  This file is part of OSSP l2, a flexible logging library which
**  can be found at http://www.ossp.org/pkg/lib/l2/.
**
**  Permission to use, copy, modify, and distribute this software for
**  any purpose with or without fee is hereby granted, provided that
**  the above copyright notice and this permission notice appear in all
**  copies.
**
**  THIS SOFTWARE IS PROVIDED ``AS IS'' AND ANY EXPRESSED OR IMPLIED
**  WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
**  MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
**  IN NO EVENT SHALL THE AUTHORS AND COPYRIGHT HOLDERS AND THEIR
**  CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
**  SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
**  LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
**  USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
**  ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
**  OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT
**  OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
**  SUCH DAMAGE.
**
**  l2_ch_buffer.c: buffering channel implementation
*/

#include "l2.h"
#include "l2_p.h"     /* for TRACE macro              */

#include <string.h>
#include <unistd.h>   /* for alarm(3)                 */
#include <signal.h>   /* for sigaction(2) and SIGALRM */

#ifdef HAVE_SYS_TIME_H
#include <sys/time.h> /* for setitimer(2)             */
#endif

#define L2_BUFFER_TIMER ITIMER_REAL /* calls to [s|g]etitimer() and alarm() */

/* declare private channel configuration */
typedef struct {
    char       *buf;
    int         bufpos;
    int         bufsize;
    long        bufinterval;
    struct      sigaction sigalrm;
#if defined(HAVE_SETITIMER) && defined(HAVE_SYS_TIME_H)
    struct      itimerval valprev;
#endif
    int         levelflush;
    l2_level_t  level;
} l2_ch_buffer_t;

/* Sets the VIRTUAL timer to preconfigured value in cfg */
static int set_alarm(l2_ch_buffer_t *cfg)
{
#if defined(HAVE_SETITIMER) && defined(HAVE_SYS_TIME_H)
    struct itimerval valtest, valnew;

    /* initialize auto vars before using them */
    memset(&valnew, 0, sizeof(valnew));

    valnew.it_interval.tv_sec = cfg->bufinterval;
    valnew.it_interval.tv_usec = 0;
    valnew.it_value.tv_sec = cfg->bufinterval;
    valnew.it_value.tv_usec = 0;
    if  ((getitimer(L2_BUFFER_TIMER, &valtest) == 0) &&
        ((valtest.it_value.tv_sec | valtest.it_value.tv_usec |
        valtest.it_interval.tv_sec | valtest.it_interval.tv_usec) == 0))
        return setitimer(L2_BUFFER_TIMER, &valnew, &cfg->valprev);
    else {
        cfg->bufinterval = -1L; /* mark this timer as broken */
        assert(FALSE);
        return 1; /* to make the compiler happy */
    }
#else
    unsigned int uiAlarmed = 0;

    assert(uiAlarmed = alarm(cfg->bufinterval));
    if (uiAlarmed) {            /* check if SIGALRM is occupied          */
        alarm(uiAlarmed);       /* ...if so, then hack in the old value  */
        cfg->bufinterval = -1L; /* ...mark this timer as broken          */
    }
    return 0;
#endif
}

/* Resets the VIRTUAL timer to preconfigured value in cfg */
static int reset_alarm(l2_ch_buffer_t *cfg)
{
#if defined(HAVE_SETITIMER) && defined(HAVE_SYS_TIME_H)
    struct itimerval valnew;

    /* initialize auto vars before using them */
    memset(&valnew, 0, sizeof(valnew));

    valnew.it_interval.tv_sec = cfg->bufinterval;
    valnew.it_interval.tv_usec = 0;
    valnew.it_value.tv_sec = cfg->bufinterval;
    valnew.it_value.tv_usec = 0;
    return setitimer(L2_BUFFER_TIMER, &valnew, 0);
#else
    alarm(cfg->bufinterval);
    return 0;
#endif
}

static void catchsignal(int sig, ...)
{
    va_list ap;
    static  l2_channel_t   *ch  = NULL;
    static  l2_ch_buffer_t *cfg = NULL;

    if (sig == 0) {
        va_start(ap, sig);
        ch  = va_arg(ap, l2_channel_t   *); /* init the handler just like */
        cfg = va_arg(ap, l2_ch_buffer_t *); /* Thomas Lotterer does       */
        va_end(ap);
    }
    else if (sig == SIGALRM) {
        TRACE("SIGALRM caught");
        l2_channel_flush(ch);
        reset_alarm(cfg); /* alarm(3) doesn't auto-reset like setitime(2) */
    }
}

/* create channel */
static l2_result_t hook_create(l2_context_t *ctx, l2_channel_t *ch)
{
    l2_ch_buffer_t *cfg;

    /* allocate private channel configuration */
    if ((cfg = (l2_ch_buffer_t *)malloc(sizeof(l2_ch_buffer_t))) == NULL)
        return L2_ERR_MEM;

    /* initialize configuration with reasonable defaults */
    cfg->buf         = NULL;
    cfg->bufpos      = 0;
    cfg->bufsize     = 4096;
    cfg->bufinterval = 0;
    cfg->levelflush  = 0;
    cfg->level       = L2_LEVEL_NONE;
    memset(&cfg->sigalrm, 0, sizeof(cfg->sigalrm));
#if defined(HAVE_SETITIMER) && defined(HAVE_SYS_TIME_H)
    memset(&cfg->valprev, 0, sizeof(cfg->valprev));
#endif

    /* link private channel configuration into channel context */
    ctx->vp = cfg;

    return L2_OK;
}

/* configure channel */
static l2_result_t hook_configure(l2_context_t *ctx, l2_channel_t *ch, const char *fmt, va_list *ap)
{
    l2_ch_buffer_t *cfg = (l2_ch_buffer_t *)ctx->vp;
    l2_param_t pa[4];
    l2_result_t rv;
    l2_env_t *env;

    /* feed and call generic parameter parsing engine */
    L2_PARAM_SET(pa[0], size,       INT, &cfg->bufsize);
    L2_PARAM_SET(pa[1], interval,   INT, &cfg->bufinterval);
    L2_PARAM_SET(pa[2], levelflush, INT, &cfg->levelflush);
    L2_PARAM_END(pa[3]);
    l2_channel_env(ch, &env);
    rv = l2_util_setparams(env, pa, fmt, ap);
    if (cfg->bufinterval == -1L) /* -1 is reserved by L2 */
        return L2_ERR_ARG;       /* set_alarm() uses it  */

    if (cfg->bufsize < 0)
        return L2_ERR_ARG;

    return rv;
}

/* open channel */
static l2_result_t hook_open(l2_context_t *ctx, l2_channel_t *ch)
{
    l2_ch_buffer_t *cfg = (l2_ch_buffer_t *)ctx->vp;
    struct sigaction locact;

    if ((cfg->bufinterval != 0) && (cfg->bufinterval != -1L)) {
        /* initialize auto vars before using them */
        memset(&locact, 0, sizeof(locact));

        locact.sa_handler = (void(*)(int))catchsignal;
        sigemptyset(&locact.sa_mask);
        locact.sa_flags = 0;

        catchsignal(0, ch, (l2_ch_buffer_t *)ctx->vp);
        /* save old signal context before replacing with our own */
        if (sigaction(SIGALRM, &locact, &cfg->sigalrm) < 0)
            return L2_ERR_SYS;

        if (set_alarm(cfg))      /* this is our own L2 set_alarm */
            return L2_ERR_SYS;
    }

    /* open channel buffer */
    if (cfg->bufsize > 0) {
        if ((cfg->buf = malloc(cfg->bufsize)) == NULL)
            return L2_ERR_MEM;
        cfg->bufpos = 0;
    }

    return L2_OK_PASS;
}

/* write to channel */
static l2_result_t hook_write(l2_context_t *ctx, l2_channel_t *ch,
                              l2_level_t level, const char *buf, size_t buf_size)
{
    l2_ch_buffer_t *cfg = (l2_ch_buffer_t *)ctx->vp;
    l2_channel_t *downstream;
    l2_result_t rv;

    if (buf_size > (cfg->bufsize - cfg->bufpos)) {
        /* flush buffer if necessary */
        if (cfg->bufpos > 0) {
            downstream = NULL;
            while ((rv = l2_channel_downstream(ch, &downstream)) == L2_OK)
                if ((rv = l2_channel_write(downstream, cfg->level, cfg->buf, cfg->bufpos)) != L2_OK)
                    return rv;
            cfg->bufpos = 0;
            cfg->level  = L2_LEVEL_NONE;
        }
        /* pass through immediately to downstream if still too large */
        if (buf_size > cfg->bufsize) {
            downstream = NULL;
            while ((rv = l2_channel_downstream(ch, &downstream)) == L2_OK)
                if ((rv = l2_channel_write(downstream, level, buf, buf_size)) != L2_OK)
                    return rv;
            return L2_OK;
        }
    }

    /* flush if incoming message level differs from those already in buffer */
    if (   (cfg->levelflush)        /* if different levels force a flush    */
        && (cfg->bufpos > 0)        /* and there is something in the buffer */
        && (cfg->level != L2_LEVEL_NONE) /* and a remembered level is known */
        && (level != cfg->level)         /* and the levels really differ    */)
    {
        downstream = NULL;
        while (l2_channel_downstream(ch, &downstream) == L2_OK)
            if ((rv = l2_channel_write(downstream, cfg->level, cfg->buf, cfg->bufpos)) != L2_OK)
                return rv;
        cfg->bufpos = 0;
        cfg->level  = L2_LEVEL_NONE;
    }

    /* finally write incoming message to channel buffer */
    memcpy(cfg->buf+cfg->bufpos, buf, buf_size);
    cfg->bufpos += buf_size;
    cfg->level = level;

    return L2_OK;
}

/* flush channel */
static l2_result_t hook_flush(l2_context_t *ctx, l2_channel_t *ch)
{
    l2_ch_buffer_t *cfg = (l2_ch_buffer_t *)ctx->vp;
    l2_channel_t *downstream;
    l2_result_t rv;

    /* write the buffer contents downstream */
TRACE("l2_ch_buffer hook_flush called\n");
    if (cfg->bufpos > 0) {
        downstream = NULL;
        while (l2_channel_downstream(ch, &downstream) == L2_OK)
            if ((rv = l2_channel_write(downstream, cfg->level, cfg->buf, cfg->bufpos)) != L2_OK)
                return rv;
        cfg->bufpos = 0;
        cfg->level  = L2_LEVEL_NONE; /* reset this->context->level */
    }

    /* reset the flush alarm timer to synchronize the buffer */
    if ((cfg->bufinterval != 0) && (cfg->bufinterval != -1L))
        if (reset_alarm(cfg))
            return L2_ERR_SYS;

    return L2_OK_PASS;
}

/* close channel */
static l2_result_t hook_close(l2_context_t *ctx, l2_channel_t *ch)
{
    l2_ch_buffer_t *cfg = (l2_ch_buffer_t *)ctx->vp;
    l2_channel_t *downstream;
    l2_result_t rv;

    if ((cfg->bufinterval != 0) && (cfg->bufinterval != -1L)) {
#if defined(HAVE_SETITIMER) && defined(HAVE_SYS_TIME_H)
        if (setitimer(L2_BUFFER_TIMER, &cfg->valprev, 0)) /* restore timer */
            return L2_ERR_SYS;
#else
        alarm(0);
#endif
        /* restore previous signal context if previously saved & replaced  */
        if (&cfg->sigalrm.sa_handler)
            if (sigaction(SIGALRM, &cfg->sigalrm, 0) < 0)
                rv = L2_ERR_SYS;
    }

    /* write pending data before closing down */
    if (cfg->bufpos > 0) {
        downstream = NULL;
        while (l2_channel_downstream(ch, &downstream) == L2_OK)
            if ((rv = l2_channel_write(downstream, cfg->level, cfg->buf, cfg->bufpos)) != L2_OK)
                return rv;
        cfg->bufpos = 0;
        cfg->level  = L2_LEVEL_NONE; /* reset this->context->level */
    }

    /* close channel buffer */
    if (cfg->buf != NULL) {
        free(cfg->buf);
        cfg->buf = NULL;
    }

    return L2_OK_PASS;
}

/* destroy channel */
static l2_result_t hook_destroy(l2_context_t *ctx, l2_channel_t *ch)
{
    l2_ch_buffer_t *cfg = (l2_ch_buffer_t *)ctx->vp;

    /* destroy channel configuration */
    if (cfg->buf != NULL)
        free(cfg->buf);
    free(cfg);

    return L2_OK_PASS;
}

/* exported channel handler structure */
l2_handler_t l2_handler_buffer = {
    "buffer",
    L2_CHANNEL_FILTER,
    hook_create,
    hook_configure,
    hook_open,
    hook_write,
    hook_flush,
    hook_close,
    hook_destroy
};