Streams

#include <stream.h>

typedef long long mu_off_t;
typedef void *mu_transport_t;

int mu_file_stream_create(mu_stream_t *stream, const char *filename, int flags);
int mu_filter_prog_stream_create(mu_stream_t *stream, const char *progname, mu_stream_t input);
int mu_mapfile_stream_create(mu_stream_t *stream, const char *filename, int flags);
int mu_memory_stream_create(mu_stream_t *stream, const char *filename, int flags);
int mu_prog_stream_create(mu_stream_t *stream, const char *progname, int flags);
int mu_socket_stream_create(mu_stream_t *stream, const char *filename, int flags);
int mu_stdio_stream_create(mu_stream_t *stream, FILE* file, int flags);
int mu_tcp_stream_create(mu_stream_t *stream, const char *host, int port, int flags);
int mu_tcp_stream_create_with_source_host(mu_stream_t *stream, const char *host, int port,const char* source_host, int flags);
int mu_tcp_stream_create_with_source_ip(mu_stream_t *stream, const char *host, int port,unsigned long source_ip, int flags);
int mu_temp_file_stream_create(mu_stream_t *stream, const char *dir);

int mu_stream_close(mu_stream_t stream);
void mu_stream_destroy(mu_stream_t *stream, void *owner);
int mu_stream_flush(mu_stream_t stream);
int mu_stream_getline(mu_stream_t stream, char **buf, size_t *buffsize, mu_off_t offset, size_t *pnread);
int mu_stream_get_flags(mu_stream_t stream, int *flags);
void* mu_stream_get_owner(mu_stream_t stream);
int mu_stream_get_property(mu_stream_t stream, mu_property_t *prop);
int mu_stream_get_state(mu_stream_t stream);
int mu_stream_get_transport(mu_stream_t stream, mu_transport_t *pt);
int mu_stream_get_transport2(mu_stream_t stream, mu_transport_t *pt1, mu_transport_t *pt2);
int mu_stream_is_seekable(mu_stream_t stream);
int mu_stream_open(mu_stream_t stream);
int mu_stream_printf(mu_stream_t stream, mu_off_t *poff, const char *fmt, ...);
int mu_stream_read(mu_stream_t stream, char *buf, size_t count, mu_off_t offset, size_t *pnread);
int mu_stream_readline(mu_stream_t stream, char *buf, size_t count, mu_off_t offset, size_t *pnread);
int mu_stream_seek(mu_stream_t stream, mu_off_t off, int whence);
int mu_stream_sequential_getline(mu_stream_t stream, char **pbuf, size_t *pbufsize, size_t *nbytes);
int mu_stream_sequential_printf(mu_stream_t stream, const char *fmt, ...);
int mu_stream_sequential_read(mu_stream_t stream, char *buff,size_t size, size_t *nbytes);
int mu_stream_sequential_readline(mu_stream_t stream, char *buff,size_t size, size_t *nbytes);
int mu_stream_sequential_vprintf(mu_stream_t stream, const char *fmt, va_list ap);
int mu_stream_sequential_write(mu_stream_t stream, char *buff,size_t size);
int mu_stream_setbufsiz(mu_stream_t stream, size_t size);
int mu_stream_set_flags(mu_stream_t stream, int flags);
int mu_stream_set_property(mu_stream_t stream, mu_property_t property, void *owner);
int mu_stream_shutdown(mu_stream_t stream, int how);
int mu_stream_size(mu_stream_t stream, mu_off_t *psize);
int mu_stream_strerror(mu_stream_t stream, const char **p);
int mu_stream_truncate(mu_stream_t stream, mu_off_t len);
int mu_stream_vprintf(mu_stream_t stream, mu_off_t *poff, const char *fmt, va_list ap);
int mu_stream_wait(mu_stream_t stream, int *pflags, struct timeval *tvp);
int mu_stream_write(mu_stream_t stream, const char *buf, size_t count, mu_off_t offset, size_t *pnwrite);

int mu_stream_create(mu_stream_t *stream,int flags,void *owner);
int mu_stream_set_close(mu_stream_t stream, int(*_close)(mu_stream_t), void *owner);
int mu_stream_set_destroy(mu_stream_t stream, void(*_destroy)(mu_stream_t), void *owner);
int mu_stream_set_flush(mu_stream_t stream, int(*_flush)(mu_stream_t) ,void *owner);
int mu_stream_set_get_transport2(mu_stream_t stream,int(*get_transport2)(mu_stream_t, mu_transport_t*, mu_transport_t*), void *owner);
int mu_stream_set_open(mu_stream_t stream, int(*_open)(mu_stream_t), void *owner);
int mu_stream_set_read(mu_stream_t stream, int(*_read)(mu_stream_t,char*,size_t,mu_off_t,size_t*),void *owner);
int mu_stream_set_readline(mu_stream_t stream, int(*_readline)(mu_stream_t,char*,size_t,mu_off_t,size_t*),void *owner);
int mu_stream_set_shutdown(mu_stream_t stream, int(*_shutdown)(mu_stream_t,int),void *owner);
int mu_stream_set_size(mu_stream_t stream, int(*_size)(mu_stream_t,mu_off_t*) ,void *owner);
int mu_stream_set_strerror(mu_stream_t stream, int(*_fp)(mu_stream_t,const char**),void *owner);
int mu_stream_set_truncate(mu_stream_t stream,int(*_truncate)(mu_stream_t,mu_off_t),void *owner);
int mu_stream_set_wait(mu_stream_t stream, int(*_wait)(mu_stream_t,int*, struct timeval*),void *owner);
int mu_stream_set_write(mu_stream_t stream, int(*_write)(mu_stream_t, const char* size_t, mu_off_t, size_t*),void *owner);

typedef struct _mu_stream  *mu_stream_t

This is an opaque data structure. None of the fields should be accessed by a user program.

The flags that can be passed to the stream creation functions are:

MU_STREAM_READ: Stream is open read only
MU_STREAM_WRITE: Stream is open write only
MU_STREAM_RDWR: Stream is open for reading and writing
MU_STREAM_APPEND:
MU_STREAM_CREAT: The underlying resource will be created if it does not exist when the stream is opened.
MU_STREAM_NONBLOCK:
MU_STREAM_NO_CHECK: The owner of the stream is not checked when the stream is destroyed
MU_STREAM_SEEKABLE:
MU_STREAM_NO_CLOSE: The underlying resource is not closed when the stream is closed or destroyed
MU_STREAM_ALLOW_LINKS:
MU_STREAM_NONLOCK:
MU_STREAM_QACCESS: Allows quick access to mailbox messages
MU_STREAM_IRGRP:
MU_STREAM_IWGRP:
MU_STREAM_IROTH:
MU_STREAM_IWOTH:
MU_STREAM_IMASK:

int mu_file_stream_create(mu_stream_t *stream, const char *filename, int flags);

Creates a stream based on the file filename. The function returns 0 if successful or an error code.

int mu_filter_prog_stream_create(mu_stream_t *stream, const char *progname, mu_stream_t input);

Creates a stream where input from the stream input is filtered through progname. The output of progname provides the raw data for the main stream. The function returns 0 if successful or an error code.

int mu_mapfile_stream_create(mu_stream_t *stream, const char *filename, int flags);

Creates a stream based on the file filename, which is mapped into memory when it is opened, providing that the underlying system supports memory mapped files. The function returns 0 if successful or an error code.

int mu_memory_stream_create(mu_stream_t *stream, const char *filename, int flags);

Creates a stream based on a block of memory. If filename is not NULL then the block of memory is initialised to its contents. Changes to the memory are not written back to the file. The function returns 0 if successful or an error code.

int mu_prog_stream_create(mu_stream_t *stream, const char *progname, int flags);

Creates a stream based on the output of the program progname. The function returns 0 if successful or an error code.

int mu_stream_set_close(mu_stream_t stream, int(*_close)(mu_stream_t), void *owner);
int mu_stream_set_destroy(mu_stream_t stream, void(*_destroy)(mu_stream_t), void *owner);
int mu_stream_set_flush(mu_stream_t stream, int(*_flush)(mu_stream_t) ,void *owner);
int mu_stream_set_get_transport2(mu_stream_t stream,int(*get_transport2)(mu_stream_t, mu_transport_t*, mu_transport_t*), void *owner);
int mu_stream_set_open(mu_stream_t stream, int(*_open)(mu_stream_t), void *owner);
int mu_stream_set_read(mu_stream_t stream, int(*_read)(mu_stream_t,char*,size_t,mu_off_t,size_t*),void *owner);
int mu_stream_set_readline(mu_stream_t stream, int(*_readline)(mu_stream_t,char*,size_t,mu_off_t,size_t*),void *owner);
int mu_stream_set_shutdown(mu_stream_t stream, int(*_shutdown)(mu_stream_t,int),void *owner);
int mu_stream_set_size(mu_stream_t stream, int(*_size)(mu_stream_t,mu_off_t*) ,void *owner);
int mu_stream_set_strerror(mu_stream_t stream, int(*_fp)(mu_stream_t,const char**),void *owner);
int mu_stream_set_truncate(mu_stream_t stream,int(*_truncate)(mu_stream_t,mu_off_t),void *owner);
int mu_stream_set_wait(mu_stream_t stream, int(*_wait)(mu_stream_t,int*, struct timeval*),void *owner);
int mu_stream_set_write(mu_stream_t stream, int(*_write)(mu_stream_t, const char* size_t, mu_off_t, size_t*),void *owner);

These routines are used to define the interface routines for each data source. The owner argument should match the owner argument passed to the mu_stream_create function. The functions returns 0 if successful or an error code.

int mu_socket_stream_create(mu_stream_t *stream, const char *filename, int flags);

Creates a stream based on the Unix socket specified by filename. The function returns 0 if successful or an error code.

int mu_stdio_stream_create(mu_stream_t *stream, FILE* file, int flags);

Creates a stream based on the previously opened file file. The function returns 0 if successful or an error code.

int mu_stream_close(mu_stream_t stream);

Closes the stream. The resources allocated to the stream are not released until mu_stream_destroy is called. The function returns 0 if successful or an error code.

int mu_stream_create(mu_stream_t *stream,int flags,void *owner);

The stream routines provide a generic I/O interface. This routine allocates a memory structure used to manage the interface and returns a pointer to it in the stream argument. The mu_xxx_stream_create functions all call this routine to allocate the resources, passing a pointer to a structure used to manage the actual I/O mechanism as the owner argument. The function returns 0 if successful or an error code.

void mu_stream_destroy(mu_stream_t *stream, void *owner);

Closes and releases the resources allocated to the stream. The owner argument must equal the owner argument passed to the mu_stream_create function unless the stream was created with the MU_STREAM_NO_CHECK flag set.

int mu_stream_flush(mu_stream_t stream);

This functions is called to flush the output buffers of the stream's underlying data source. The function return 0 if successful or an error code.

int mu_stream_getline(mu_stream_t stream, char **buf, size_t *buffsize, mu_off_t offset, size_t *pnread);

Reads characters from the underlying data until a newline or EOF is detected and stores them in buf. The size of buf is stored in buffsize. If buf is too small then its size is increased (in 128 byte chunks) as much as is needed to hold the data. If the underlying data source supports the concept of an offset (e.g. file based streams) then the data is read from offset bytes from the beginning of the data. The actual number of characters read is returned in pnread. The function returns 0 if successful or an error code.

This routine relies on the mu_stream_readline function being implemented for the data source.

int mu_stream_get_flags(mu_stream_t stream, int *flags);

Returns in flags the current stream flags. The function returns 0 if successful or an error code.

void* mu_stream_get_owner(mu_stream_t stream);

This function returns the owner argument which was passed to the mu_stream_create function. This is usually an opaque memory structure which is used to manage the actual I/O operations of the stream.

int mu_stream_get_property(mu_stream_t stream, mu_property_t *prop);

Creates and returns in prop a pointer to an mu_property_t structure which is a collection of keys and values. The function returns 0 if successful or an error code.

int mu_stream_get_state(mu_stream_t stream);

Returns the state of the stream, which will be one of:
MU_STREAM_STATE_OPEN stream has been open'ed
MU_STREAM_STATE_READ last operation was a read
MU_STREAM_STATE_WRITE last operation was a write
MU_STREAM_STATE_CLOSE stream has been close'd

int mu_stream_get_transport(mu_stream_t stream, mu_transport_t *pt);
int mu_stream_get_transport2(mu_stream_t stream, mu_transport_t *pt1, mu_transport_t *pt2);

These functions return data structures relating to the underlying data source. mu_stream_get_transport returns the same as the pt1 argument of mu_stream_get_transport2. The functions return 0 if successful or an error code.

int mu_stream_is_seekable(mu_stream_t stream);

Returns TRUE if the MU_STREAM_SEEKABLE flag is set.

int mu_stream_open(mu_stream_t stream);

Opens the stream. The function returns 0 if successful or an error code.

int mu_stream_printf(mu_stream_t stream, mu_off_t *poff, const char *fmt, ...);

Writes a formatted string to the stream, fmt holds a “printf” style string with the following arguments being the parameters for the string. If the underlying data source supports the concept of an offset (e.g. file based streams) then the data is written starting at *poff bytes from the beginning of the data. On return *poff has been increased by the number of characters written. The function returns 0 if successful or an error code.

int mu_stream_read(mu_stream_t stream, char *buf, size_t count, mu_off_t offset, size_t *pnread);

Reads up to count-1 characters and stores them in buff. The actual number of characters read is returned in pnread. If the underlying data source supports the concept of an offset (e.g. file based streams) then the data is read from offset bytes from the beginning of the data. The function returns 0 if successful or an error code.

int mu_stream_readline(mu_stream_t stream, char *buf, size_t count, mu_off_t offset, size_t *pnread);

Reads up to count-1 characters and stores them in buff. Characters are read until a newline or EOF is detected or the buffer is filled. If the underlying data source supports the concept of an offset (e.g. file based streams) then the data is read from offset bytes from the beginning of the data. The actual number of characters read is returned in pnread. The function returns 0 if successful or an error code.

int mu_stream_seek(mu_stream_t stream, mu_off_t off, int whence);

Positions the stream at off relative to an origin that is defined by whence. Whence can be SEEK_SET (beginning of stream), SEEK_CUR (current stream position), SEEK_END (end ofthe file). The function returns 0 is successful or an error code.

This function requires an implentation of mu_stream_size in order to function correctly.

int mu_stream_sequential_getline(mu_stream_t stream, char **pbuf, size_t *pbufsize, size_t *nbytes);

Reads characters from the point where the last operation finished until a newline or EOF is detected and stores them in pbuf. The size of pbuf is stored in pbufsize. If pbuf is too small then its size is increased (in 128 byte chunks) as much as is needed to hold the data. The actual number of characters read is returned in nbytes. The function returns 0 if successful or an error code.

This routine relies on the mu_stream_getline function being implemented for the data source.

int mu_stream_sequential_printf(mu_stream_t stream, const char *fmt, ...);

Writes a formatted string to the stream, fmt holds a “printf” style string with the following arguments being the parameters for the string. The data is written at the point in the stream where the last operation finished. The function returns 0 if successful or an error code.

int mu_stream_sequential_read(mu_stream_t stream, char *buff,size_t size, size_t *nbytes);

Attempts to read size-1 bytes of data from the stream from the point where the last operation finished. Characters are read until a newline, EOF or the end of the buffer is reached. The actual number of bytes read is stored in nbytes. The function returns 0 if successful or an error code.

This function requires an implementation of mu_stream_readline.

int mu_stream_sequential_readline(mu_stream_t stream, char *buff,size_t size, size_t *nbytes);

Attempts to read size-1 bytes of data from the stream from the point where the last operation finished. The actual number of bytes read is stored in nbytes. The function returns 0 if successful or an error code.

int mu_stream_sequential_vprintf(mu_stream_t stream, const char *fmt, va_list ap);

Writes a formatted string to the stream, fmt holds a “printf” style string and ap holds the parameters for the string in va_list format. The data is written at the point in the stream where the last operation finished. The function returns 0 if successful or an error code.

The function returns 0 if successful or an error code.

int mu_stream_sequential_write(mu_stream_t stream, char *buff,size_t size);

Writes size characters from buff to the stream. The data is written at the point where the last operation finished. The function returns 0 if successful or an error code.

int mu_stream_setbufsiz(mu_stream_t stream, size_t size);

Passing a non-zero value of size enables a read buffer within the stream. The function returns 0 if successful or an error code.

int mu_stream_set_flags(mu_stream_t stream, int flags);

This sets the flags associated with the stream to the value passed in flags. The function returns 0 if successful or an error code.

int mu_stream_set_property(mu_stream_t stream, mu_property_t property, void *owner);

Replaces any property structure associated with the stream with the one passed in property. To be successful the owner argument must match the owner argument passed when the stream was created. The function returns 0 if successful or an error code.

int mu_stream_shutdown(mu_stream_t stream, int how);

Shuts down the reading or writing of data for a stream. how should be MU_STREAM_READ to shut down reading, or MU_STREAM_WRITE to shutdown writing. The function returns 0 if successful or an error code.

int mu_stream_size(mu_stream_t stream, mu_off_t *psize);

This function returns in psize the size of the underlying stream. The function returns 0 if successful or an error code.

int mu_stream_strerror(mu_stream_t stream, const char **p);

Returns a formatted error message in p. The function returns 0 if successful or an error code.

int mu_stream_truncate(mu_stream_t stream, mu_off_t len);

Truncates the underlying data source at len bytes from the beginning. The function returns 0 if successful or an error code.

int mu_stream_vprintf(mu_stream_t stream, mu_off_t *poff, const char *fmt, va_list ap);

Writes a formatted string to the stream, fmt holds a “printf” style string and ap holds the parameters for the string in va_list format. If the underlying data source supports the concept of an offset (e.g. file based streams) then the data is written starting at *poff bytes from the beginning of the data. On return *poff has been increased by the number of characters written. The function returns 0 if successful or an error code.

int mu_stream_wait(mu_stream_t stream, int *pflags, struct timeval *tvp);

Monitors the file descriptors associated with the stream until they become ready for reading, writing or an exception occurs. The events monitored are controlled by pflags which can be one or more of MU_STREAM_READY_RD, MU_STREAM_READY_WR and MU_STREAM_READY_EX. There is a protective timer which is defined by tvp. If NULL is passed instead of a pointer to a timeval then the program will wait indefinitely. When an event occurs, pflags will be modified to indicate which event(s) have occurred. The function returns 0 if successful or an error code.

int mu_stream_write(mu_stream_t stream, const char *buf, size_t count, mu_off_t offset, size_t *pnwrite);

Writes count characters from buf to the stream. If the underlying data source supports the concept of an offset (e.g. file based streams) then the data is written starting at offset bytes from the beginning of the data. The actual number of characters written is returned in pnwrite. The function returns 0 if successful or an error code.

int mu_tcp_stream_create(mu_stream_t *stream, const char *host, int port, int flags);

Establishes a stream based on a TCP connection to port of host. The function returns 0 is successful or an error code.

int mu_tcp_stream_create_with_source_host(mu_stream_t *stream, const char *host, int port,const char* source_host, int flags);

Establishes a stream based on a TCP connection to port of host. The socket on the local machine is bound to the IP address associated with the source_host argument. The function returns 0 is successful or an error code.

int mu_tcp_stream_create_with_source_ip(mu_stream_t *stream, const char *host, int port,unsigned long source_ip, int flags);

Establishes a stream based on a TCP connection to port of host. The socket on the local machine is bound to the IP address in the source_ip argument. The function returns 0 is successful or an error code.

int mu_temp_file_stream_create(mu_stream_t *stream, const char *dir);

This creates a stream based on a temporary file which is opened in the directory dir. If dir is NULL then the directory is taken from the TMPDIR environment variable or /tmp. The function returns 0 if successful or an error code.