ossp-pkg/sio/sio_buffer.c
/*
** 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
};
