stream.h

Go to the documentation of this file.

#ifndef _CEPTR_STREAM_H
#define _CEPTR_STREAM_H

#include <stdio.h>
#include <pthread.h>
#include <stdbool.h>

enum StreamTypes {UnixStream,SocketStream};
enum {StreamHasData=0x0001,StreamCloseOnFree=0x0002,StreamReader=0x0004,StreamWaiting=0x0008,StreamAlive=0x8000,StreamCloseAfterOneWrite=0x0010,StreamDying=0x0100,StreamLoadByLine=0x0200};

typedef struct Stream Stream;

typedef void (*hasDataCallbackFn)(Stream *);

enum ScanStates {StreamScanInitial,StreamScanSuccess,StreamScanPartial,StreamScanComplete};
struct Stream {
    int type;
    int flags;
    int err;
    pthread_t pthread;
    pthread_mutex_t mutex;
    pthread_cond_t cv;
    char *buf;
    size_t buf_size;
    size_t bytes_used;
    int scan_state;
    size_t unit_start;
    size_t unit_size;
    size_t partial;
    hasDataCallbackFn callback;
    void *callback_arg1;
    int callback_arg2;
    char *delim;
    int delim_len;
    // hold data for the different types of streams that are implemented
    // for now just unix_streams
    union {
        FILE *unix_stream;
        int socket_stream;
    } data;
};

typedef struct SocketListener SocketListener;
typedef void (*lisenterConnectionCallbackFn)(Stream *,void *);

struct SocketListener {
    int port;
    int sockfd;
    pthread_t pthread;
    lisenterConnectionCallbackFn callback;
    void *callback_arg;
    bool alive;
    char *delim;
};

char *DELIM_LF;
char *DELIM_CRLF;

#define DEFAULT_READER_BUFFER_SIZE 1000
#define _st_new_unix_stream(s,r) __st_new_unix_stream(s,r?DEFAULT_READER_BUFFER_SIZE:0)
Stream *__st_new_unix_stream(FILE *stream,size_t reader_buffer_size);
Stream *__st_alloc_stream();
size_t __st_unix_stream_load(Stream *st);
#define __st_init_scan(s) s->scan_state = StreamScanInitial
#define __st_buf_full(s) (s->bytes_used == s->buf_size)

void __st_scan(Stream *st);

SocketListener *_st_new_socket_listener(int port,lisenterConnectionCallbackFn fn,void *callback_arg,char * delim);
void _st_close_listener(SocketListener *l);

void _st_start_read(Stream *st);
void _st_data_consumed(Stream *st);
#define _st_is_alive(st) ((st->flags & StreamAlive) || (st->flags & StreamReader && (st->scan_state != StreamScanComplete)))
void _st_kill(Stream *st);
void _st_free(Stream *);
#define _st_data(st) (&(st)->buf[st->unit_start])
#define _st_data_size(st) (st)->unit_size

int _st_write(Stream *stream,char *buf,size_t len);
int _st_writeln(Stream *stream,char *buf);

#endif