/*
    * ustream - library for stream buffer management
    *
    * Copyright (C) 2012 Felix Fietkau <nbd@openwrt.org>
    *
    * Permission to use, copy, modify, and/or 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.
    *
   * THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
   * WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
   * MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
   * ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
   * WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
   * ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
   * OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
   */
  
  #ifndef __USTREAM_H
  #define __USTREAM_H
  
  #include <stdarg.h>
  #include "uloop.h"
  
  struct ustream;
  struct ustream_buf;
  
  enum read_blocked_reason {
          READ_BLOCKED_USER = (1 << 0),
          READ_BLOCKED_FULL = (1 << 1),
  };
  
  struct ustream_buf_list {
          struct ustream_buf *head;
          struct ustream_buf *data_tail;
          struct ustream_buf *tail;
  
          int (*alloc)(struct ustream *s, struct ustream_buf_list *l);
  
          int data_bytes;
  
          int min_buffers;
          int max_buffers;
          int buffer_len;
  
          int buffers;
  };
  
  struct ustream {
          struct ustream_buf_list r, w;
          struct uloop_timeout state_change;
          struct ustream *next;
  
          /*
           * notify_read: (optional)
           * called by the ustream core to notify that new data is available
           * for reading.
           * must not free the ustream from this callback
           */
          void (*notify_read)(struct ustream *s, int bytes_new);
  
          /*
           * notify_write: (optional)
           * called by the ustream core to notify that some buffered data has
           * been written to the stream.
           * must not free the ustream from this callback
           */
          void (*notify_write)(struct ustream *s, int bytes);
  
          /*
           * notify_state: (optional)
           * called by the ustream implementation to notify that the read
           * side of the stream is closed (eof is set) or there was a write
           * error (write_error is set).
           * will be called again after the write buffer has been emptied when
           * the read side has hit EOF.
           */
          void (*notify_state)(struct ustream *s);
  
          /*
           * write:
           * must be defined by ustream implementation, accepts new write data.
           * 'more' is used to indicate that a subsequent call will provide more
           * data (useful for aggregating writes)
           * returns the number of bytes accepted, or -1 if no more writes can
           * be accepted (link error)
           */
          int (*write)(struct ustream *s, const char *buf, int len, bool more);
  
          /*
           * free: (optional)
           * defined by ustream implementation, tears down the ustream and frees data
           */
          void (*free)(struct ustream *s);
  
          /*
           * set_read_blocked: (optional)
           * defined by ustream implementation, called when the read_blocked flag
           * changes
          */
         void (*set_read_blocked)(struct ustream *s);
 
         /*
          * poll: (optional)
          * defined by the upstream implementation, called to request polling for
          * available data.
          * returns true if data was fetched.
          */
         bool (*poll)(struct ustream *s);
 
         /*
          * ustream user should set this if the input stream is expected
          * to contain string data. the core will keep all data 0-terminated.
          */
         bool string_data;
         bool write_error;
         bool eof;
         uint8_t pending_cb;
 
         enum read_blocked_reason read_blocked;
 };
 
 struct ustream_fd {
         struct ustream stream;
         struct uloop_fd fd;
 };
 
 struct ustream_buf {
         struct ustream_buf *next;
 
         char *data;
         char *tail;
         char *end;
 
         char head[];
 };
 
 /* ustream_fd_init: create a file descriptor ustream (uses uloop) */
 void ustream_fd_init(struct ustream_fd *s, int fd);
 
 /* ustream_free: free all buffers and data associated with a ustream */
 void ustream_free(struct ustream *s);
 
 /* ustream_consume: remove data from the head of the read buffer */
 void ustream_consume(struct ustream *s, int len);
 
 /*
  * ustream_read: read and consume data in read buffer into caller-specified
  * area.  Return length of data read.
  */
 int ustream_read(struct ustream *s, char *buf, int buflen);
 /* ustream_write: add data to the write buffer */
 int ustream_write(struct ustream *s, const char *buf, int len, bool more);
 int ustream_printf(struct ustream *s, const char *format, ...)
         __attribute__ ((format (printf, 2, 3)));
 int ustream_vprintf(struct ustream *s, const char *format, va_list arg)
         __attribute__ ((format (printf, 2, 0)));
 
 /* ustream_get_read_buf: get a pointer to the next read buffer data */
 char *ustream_get_read_buf(struct ustream *s, int *buflen);
 
 /*
  * ustream_set_read_blocked: set read blocked state
  *
  * if set, the ustream will no longer fetch pending data.
  */
 void ustream_set_read_blocked(struct ustream *s, bool set);
 
 static inline bool ustream_read_blocked(struct ustream *s)
 {
         return !!(s->read_blocked & READ_BLOCKED_USER);
 }
 
 static inline int ustream_pending_data(struct ustream *s, bool write)
 {
         struct ustream_buf_list *b = write ? &s->w : &s->r;
         return b->data_bytes;
 }
 
 static inline bool ustream_read_buf_full(struct ustream *s)
 {
         struct ustream_buf *buf = s->r.data_tail;
         return buf && buf->data == buf->head && buf->tail == buf->end &&
                s->r.buffers == s->r.max_buffers;
 }
 
 /*** --- functions only used by ustream implementations --- ***/
 
 /* ustream_init_defaults: fill default callbacks and options */
 void ustream_init_defaults(struct ustream *s);
 
 /*
  * ustream_reserve: allocate rx buffer space
  *
  * len: hint for how much space is needed (not guaranteed to be met)
  * maxlen: pointer to where the actual buffer size is going to be stored
  */
 char *ustream_reserve(struct ustream *s, int len, int *maxlen);
 
 /* ustream_fill_read: mark rx buffer space as filled */
 void ustream_fill_read(struct ustream *s, int len);
 
 /*
  * ustream_write_pending: attempt to write more data from write buffers
  * returns true if all write buffers have been emptied.
  */
 bool ustream_write_pending(struct ustream *s);
 
 static inline void ustream_state_change(struct ustream *s)
 {
         uloop_timeout_set(&s->state_change, 0);
 }
 
 static inline bool ustream_poll(struct ustream *s)
 {
         if (!s->poll)
                 return false;
 
         return s->poll(s);
 }
 
 #endif
 

• Git
• Wiki