OSSP CVS Repository

ossp-pkg/sio/sio_buffer.c 1.10

---

/*
**  OSSP sio - Stream I/O
**  Copyright (c) 2002-2005 Cable & Wireless <http://www.cw.com/>
**  Copyright (c) 2002-2005 The OSSP Project <http://www.ossp.org/>
**  Copyright (c) 2002-2005 Ralf S. Engelschall <rse@engelschall.com>
**
**  This file is part of OSSP sio, a layered stream I/O library
**  which can be found at http://www.ossp.org/pkg/lib/sio/.
**
**  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.
**
**  sio_buffer.c: buffer stage
*/

#include <stddef.h>
#include <stdlib.h>
#include <string.h>

#include "al.h"
#include "sio.h"

typedef struct {
    size_t outputsize;
    size_t inputsize;
    al_t *input, *output;
    al_label_t data_label;
} private_t;

/*
 * create stage
 *
 * allocate private instance data
 */
static
sio_rc_t buffer_init(sio_t *sio, void **up)
{
    private_t *my;

    my = (private_t *)malloc(sizeof(private_t));
    if (my == NULL)
        return SIO_ERR_MEM;

    my->inputsize    = 0;
    my->outputsize   = 0;

    sio_label(sio, SIO_LN_DATA, &my->data_label);

    *up = my;

    return SIO_OK;
}

/*
 * configure stage
 *
 * pass two void pointers
 */
static
sio_rc_t buffer_configure(sio_t *sio, void *u, void *obj, void *val)
{
    private_t *my = (private_t *)u;
    const char *name = (const char *)obj;

    if (!strcmp(name, "inputsize")) {
        my->inputsize = *(size_t *)val;
    } else if (!strcmp(name, "outputsize")) {
        my->outputsize = *(size_t *)val;
    } else {
        return SIO_ERR_ARG;
    }

    return SIO_OK;
}

/*
 * destroy stage
 */
static
sio_rc_t buffer_cleanup(sio_t *sio, void *u)
{
    private_t *my = (private_t *)u;

    free(my);

    return SIO_OK;
}

static
sio_rc_t buffer_openr(sio_t *sio, al_t *al, void *u)
{
    private_t *my = (private_t *)u;
    al_rc_t arc;

    arc = al_create(&my->input);
    if (arc != AL_OK)
        return SIO_ERR_INT;

    return SIO_OK;
}

static
sio_rc_t buffer_closer(sio_t *sio, al_t *al, void *u)
{
    private_t *my = (private_t *)u;

    al_destroy(my->input);
    my->input = NULL;

    return SIO_OK;
}

static
sio_rc_t buffer_openw(sio_t *sio, al_t *al, void *u)
{
    private_t *my = (private_t *)u;
    al_rc_t arc;

    arc = al_create(&my->output);
    if (arc != AL_OK)
        return SIO_ERR_INT;

    return SIO_OK;
}

static
sio_rc_t buffer_closew(sio_t *sio, al_t *al, void *u)
{
    private_t *my = (private_t *)u;

    al_destroy(my->output);
    my->output = NULL;

    return SIO_OK;
}

/*
 * buffer logic
 *
 * gather data from producer
 * if buffer size reached or label changes then push data to consumer
 * (read -> downstream, write -> upstream)
 *
 * buffer size depends on label type
 *
 */
static
sio_rc_t buffer_inout(sio_t *sio, al_t *al, private_t *my,
                      al_t *buf, size_t size)
{
    size_t avail, data, needed;
    al_rc_t arc;
    al_label_t label;

    arc = al_splice(buf, al_bytes(buf), 0, al, NULL);
    if (arc != AL_OK)
        return SIO_ERR_INT;

    avail = al_bytes(buf);
    if (avail <= 0)
        return SIO_OK;

    arc = al_firstlabel(buf, 0, 1, AL_FORWARD, NULL, &label);
    if (arc != AL_OK)
        return SIO_ERR_INT;

    al_flatten(buf, 0, avail, AL_FORWARD_SPAN, label, NULL, &data);

    /* non-data is not buffered */
    if (label == my->data_label)
        needed = size;
    else
        needed = 1;

    /* flush if there is extra data (that must have a different label) */
    if (data < avail)
        needed = 1;

    if (data >= needed) {
        if (data >= size)
            data = size;
        al_splice(buf, 0, data, NULL, al);
        return SIO_OK;
    }

    return SIO_OK;
}

static
sio_rc_t buffer_input(sio_t *sio, al_t *al, void *u, sio_rc_t orc)
{
    private_t *my = (private_t *)u;

    return buffer_inout(sio, al, my, my->input, my->inputsize);
}

static
sio_rc_t buffer_output(sio_t *sio, al_t *al, void *u, sio_rc_t orc)
{
    private_t *my = (private_t *)u;

    return buffer_inout(sio, al, my, my->output, my->outputsize);
}

sio_module_t sio_module_buffer = {
    "buffer",
    buffer_init,
    buffer_configure,
    buffer_cleanup,
    buffer_openr,
    buffer_closer,
    buffer_openw,
    buffer_closew,
    buffer_input,
    buffer_output,
    NULL
};

---