/*
 * Packet interface
 * Copyright (C) 1999 Kunihiro Ishiguro
 *
 * This file is part of GNU Zebra.
 *
 * GNU Zebra is free software; you can redistribute it and/or modify it
 * under the terms of the GNU General Public License as published by the
 * Free Software Foundation; either version 2, or (at your option) any
 * later version.
 *
 * GNU Zebra is distributed in the hope that it will be useful, but
 * WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with GNU Zebra; see the file COPYING.  If not, write to the Free
 * Software Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA
 * 02111-1307, USA.  
 */

#include <zebra.h>
#include <stddef.h>

#include "stream.h"
#include "memory.h"
#include "network.h"
#include "prefix.h"
#include "log.h"

DEFINE_MTYPE_STATIC(LIB, STREAM,      "Stream")
DEFINE_MTYPE_STATIC(LIB, STREAM_DATA, "Stream data")
DEFINE_MTYPE_STATIC(LIB, STREAM_FIFO, "Stream FIFO")

/* Tests whether a position is valid */ 
#define GETP_VALID(S,G) \
  ((G) <= (S)->endp)
#define PUT_AT_VALID(S,G) GETP_VALID(S,G)
#define ENDP_VALID(S,E) \
  ((E) <= (S)->size)

/* asserting sanity checks. Following must be true before
 * stream functions are called:
 *
 * Following must always be true of stream elements
 * before and after calls to stream functions:
 *
 * getp <= endp <= size
 *
 * Note that after a stream function is called following may be true:
 * if (getp == endp) then stream is no longer readable
 * if (endp == size) then stream is no longer writeable
 *
 * It is valid to put to anywhere within the size of the stream, but only
 * using stream_put..._at() functions.
 */
#define STREAM_WARN_OFFSETS(S) \
  zlog_warn ("&(struct stream): %p, size: %lu, getp: %lu, endp: %lu\n", \
             (void *)(S), \
             (unsigned long) (S)->size, \
             (unsigned long) (S)->getp, \
             (unsigned long) (S)->endp)\

#define STREAM_VERIFY_SANE(S) \
  do { \
    if ( !(GETP_VALID(S, (S)->getp) && ENDP_VALID(S, (S)->endp)) ) \
      STREAM_WARN_OFFSETS(S); \
    assert ( GETP_VALID(S, (S)->getp) ); \
    assert ( ENDP_VALID(S, (S)->endp) ); \
  } while (0)

#define STREAM_BOUND_WARN(S, WHAT) \
  do { \
    zlog_warn ("%s: Attempt to %s out of bounds", __func__, (WHAT)); \
    STREAM_WARN_OFFSETS(S); \
    assert (0); \
  } while (0)

/* XXX: Deprecated macro: do not use */
#define CHECK_SIZE(S, Z) \
  do { \
    if (((S)->endp + (Z)) > (S)->size) \
      { \
        zlog_warn ("CHECK_SIZE: truncating requested size %lu\n", \
                   (unsigned long) (Z)); \
        STREAM_WARN_OFFSETS(S); \
        (Z) = (S)->size - (S)->endp; \
      } \
  } while (0);

/* Make stream buffer. */
struct stream *
stream_new (size_t size)
{
  struct stream *s;

  assert (size > 0);
  
  s = XCALLOC (MTYPE_STREAM, sizeof (struct stream));

  if (s == NULL)
    return s;
  
  if ( (s->data = XMALLOC (MTYPE_STREAM_DATA, size)) == NULL)
    {
      XFREE (MTYPE_STREAM, s);
      return NULL;
    }
  
  s->size = size;
  return s;
}

/* Free it now. */
void
stream_free (struct stream *s)
{
  if (!s)
    return;
  
  XFREE (MTYPE_STREAM_DATA, s->data);
  XFREE (MTYPE_STREAM, s);
}

struct stream *
stream_copy (struct stream *new, struct stream *src)
{
  STREAM_VERIFY_SANE (src);
  
  assert (new != NULL);
  assert (STREAM_SIZE(new) >= src->endp);

  new->endp = src->endp;
  new->getp = src->getp;
  
  memcpy (new->data, src->data, src->endp);
  
  return new;
}

struct stream *
stream_dup (struct stream *s)
{
  struct stream *new;

  STREAM_VERIFY_SANE (s);

  if ( (new = stream_new (s->endp)) == NULL)
    return NULL;

  return (stream_copy (new, s));
}

struct stream *
stream_dupcat (struct stream *s1, struct stream *s2, size_t offset)
{
  struct stream *new;

  STREAM_VERIFY_SANE (s1);
  STREAM_VERIFY_SANE (s2);

  if ( (new = stream_new (s1->endp + s2->endp)) == NULL)
    return NULL;

  memcpy (new->data, s1->data, offset);
  memcpy (new->data + offset, s2->data, s2->endp);
  memcpy (new->data + offset + s2->endp, s1->data + offset,
	  (s1->endp - offset));
  new->endp = s1->endp + s2->endp;
  return new;
}

size_t
stream_resize (struct stream *s, size_t newsize)
{
  u_char *newdata;
  STREAM_VERIFY_SANE (s);
  
  newdata = XREALLOC (MTYPE_STREAM_DATA, s->data, newsize);
  
  if (newdata == NULL)
    return s->size;
  
  s->data = newdata;
  s->size = newsize;
  
  if (s->endp > s->size)
    s->endp = s->size;
  if (s->getp > s->endp)
    s->getp = s->endp;
  
  STREAM_VERIFY_SANE (s);
  
  return s->size;
}

size_t
stream_get_getp (struct stream *s)
{
  STREAM_VERIFY_SANE(s);
  return s->getp;
}

size_t
stream_get_endp (struct stream *s)
{
  STREAM_VERIFY_SANE(s);
  return s->endp;
}

size_t
stream_get_size (struct stream *s)
{
  STREAM_VERIFY_SANE(s);
  return s->size;
}

/* Stream structre' stream pointer related functions.  */
void
stream_set_getp (struct stream *s, size_t pos)
{
  STREAM_VERIFY_SANE(s);
  
  if (!GETP_VALID (s, pos))
    {
      STREAM_BOUND_WARN (s, "set getp");
      pos = s->endp;
    }

  s->getp = pos;
}

void
stream_set_endp (struct stream *s, size_t pos)
{
  STREAM_VERIFY_SANE(s);

  if (!ENDP_VALID(s, pos))
    {
      STREAM_BOUND_WARN (s, "set endp");
      return;
    }

  /*
   * Make sure the current read pointer is not beyond the new endp.
   */
  if (s->getp > pos)
    {
      STREAM_BOUND_WARN(s, "set endp");
      return;
    }

  s->endp = pos;
  STREAM_VERIFY_SANE(s);
}

/* Forward pointer. */
void
stream_forward_getp (struct stream *s, size_t size)
{
  STREAM_VERIFY_SANE(s);
  
  if (!GETP_VALID (s, s->getp + size))
    {
      STREAM_BOUND_WARN (s, "seek getp");
      return;
    }
  
  s->getp += size;
}

void
stream_forward_endp (struct stream *s, size_t size)
{
  STREAM_VERIFY_SANE(s);
  
  if (!ENDP_VALID (s, s->endp + size))
    {
      STREAM_BOUND_WARN (s, "seek endp");
      return;
    }
  
  s->endp += size;
}

/* Copy from stream to destination. */
void
stream_get (void *dst, struct stream *s, size_t size)
{
  STREAM_VERIFY_SANE(s);
  
  if (STREAM_READABLE(s) < size)
    {
      STREAM_BOUND_WARN (s, "get");
      return;
    }
  
  memcpy (dst, s->data + s->getp, size);
  s->getp += size;
}

/* Get next character from the stream. */
u_char
stream_getc (struct stream *s)
{
  u_char c;
  
  STREAM_VERIFY_SANE (s);

  if (STREAM_READABLE(s) < sizeof (u_char))
    {
      STREAM_BOUND_WARN (s, "get char");
      return 0;
    }
  c = s->data[s->getp++];
  
  return c;
}

/* Get next character from the stream. */
u_char
stream_getc_from (struct stream *s, size_t from)
{
  u_char c;

  STREAM_VERIFY_SANE(s);
  
  if (!GETP_VALID (s, from + sizeof (u_char)))
    {
      STREAM_BOUND_WARN (s, "get char");
      return 0;
    }
  
  c = s->data[from];
  
  return c;
}

/* Get next word from the stream. */
u_int16_t
stream_getw (struct stream *s)
{
  u_int16_t w;

  STREAM_VERIFY_SANE (s);

  if (STREAM_READABLE (s) < sizeof (u_int16_t))
    {
      STREAM_BOUND_WARN (s, "get ");
      return 0;
    }
  
  w = s->data[s->getp++] << 8;
  w |= s->data[s->getp++];
  
  return w;
}

/* Get next word from the stream. */
u_int16_t
stream_getw_from (struct stream *s, size_t from)
{
  u_int16_t w;

  STREAM_VERIFY_SANE(s);
  
  if (!GETP_VALID (s, from + sizeof (u_int16_t)))
    {
      STREAM_BOUND_WARN (s, "get ");
      return 0;
    }
  
  w = s->data[from++] << 8;
  w |= s->data[from];
  
  return w;
}

/* Get next 3-byte from the stream. */
u_int32_t
stream_get3_from (struct stream *s, size_t from)
{
  u_int32_t l;

  STREAM_VERIFY_SANE(s);
  
  if (!GETP_VALID (s, from + 3))
    {
      STREAM_BOUND_WARN (s, "get 3byte");
      return 0;
    }
  
  l  = s->data[from++] << 16;
  l |= s->data[from++] << 8;
  l |= s->data[from];
  
  return l;
}

u_int32_t
stream_get3 (struct stream *s)
{
  u_int32_t l;

  STREAM_VERIFY_SANE(s);
  
  if (STREAM_READABLE (s) < 3)
    {
      STREAM_BOUND_WARN (s, "get 3byte");
      return 0;
    }
  
  l  = s->data[s->getp++] << 16;
  l |= s->data[s->getp++] << 8;
  l |= s->data[s->getp++];
  
  return l;
}

/* Get next long word from the stream. */
u_int32_t
stream_getl_from (struct stream *s, size_t from)
{
  u_int32_t l;

  STREAM_VERIFY_SANE(s);
  
  if (!GETP_VALID (s, from + sizeof (u_int32_t)))
    {
      STREAM_BOUND_WARN (s, "get long");
      return 0;
    }
  
  l  = s->data[from++] << 24;
  l |= s->data[from++] << 16;
  l |= s->data[from++] << 8;
  l |= s->data[from];
  
  return l;
}

/* Copy from stream at specific location to destination. */
void
stream_get_from (void *dst, struct stream *s, size_t from, size_t size)
{
  STREAM_VERIFY_SANE(s);

  if (!GETP_VALID (s, from + size))
    {
      STREAM_BOUND_WARN (s, "get from");
      return;
    }

  memcpy (dst, s->data + from, size);
}

u_int32_t
stream_getl (struct stream *s)
{
  u_int32_t l;

  STREAM_VERIFY_SANE(s);
  
  if (STREAM_READABLE (s) < sizeof (u_int32_t))
    {
      STREAM_BOUND_WARN (s, "get long");
      return 0;
    }
  
  l  = s->data[s->getp++] << 24;
  l |= s->data[s->getp++] << 16;
  l |= s->data[s->getp++] << 8;
  l |= s->data[s->getp++];
  
  return l;
}

/* Get next quad word from the stream. */
uint64_t
stream_getq_from (struct stream *s, size_t from)
{
  uint64_t q;

  STREAM_VERIFY_SANE(s);
  
  if (!GETP_VALID (s, from + sizeof (uint64_t)))
    {
      STREAM_BOUND_WARN (s, "get quad");
      return 0;
    }
  
  q  = ((uint64_t) s->data[from++]) << 56;
  q |= ((uint64_t) s->data[from++]) << 48;
  q |= ((uint64_t) s->data[from++]) << 40;
  q |= ((uint64_t) s->data[from++]) << 32;  
  q |= ((uint64_t) s->data[from++]) << 24;
  q |= ((uint64_t) s->data[from++]) << 16;
  q |= ((uint64_t) s->data[from++]) << 8;
  q |= ((uint64_t) s->data[from++]);
  
  return q;
}

uint64_t
stream_getq (struct stream *s)
{
  uint64_t q;

  STREAM_VERIFY_SANE(s);
  
  if (STREAM_READABLE (s) < sizeof (uint64_t))
    {
      STREAM_BOUND_WARN (s, "get quad");
      return 0;
    }
  
  q  = ((uint64_t) s->data[s->getp++]) << 56;
  q |= ((uint64_t) s->data[s->getp++]) << 48;
  q |= ((uint64_t) s->data[s->getp++]) << 40;
  q |= ((uint64_t) s->data[s->getp++]) << 32;  
  q |= ((uint64_t) s->data[s->getp++]) << 24;
  q |= ((uint64_t) s->data[s->getp++]) << 16;
  q |= ((uint64_t) s->data[s->getp++]) << 8;
  q |= ((uint64_t) s->data[s->getp++]);
  
  return q;
}

/* Get next long word from the stream. */
u_int32_t
stream_get_ipv4 (struct stream *s)
{
  u_int32_t l;

  STREAM_VERIFY_SANE(s);
  
  if (STREAM_READABLE (s) < sizeof(u_int32_t))
    {
      STREAM_BOUND_WARN (s, "get ipv4");
      return 0;
    }
  
  memcpy (&l, s->data + s->getp, sizeof(u_int32_t));
  s->getp += sizeof(u_int32_t);

  return l;
}

float
stream_getf (struct stream *s)
{
  union {
    float r;
    uint32_t d;
  } u;
  u.d = stream_getl (s);
  return u.r;
}

double
stream_getd (struct stream *s)
{
  union {
    double r;
    uint64_t d;
  } u;
  u.d = stream_getq (s);
  return u.r;
}

/* Copy to source to stream.
 *
 * XXX: This uses CHECK_SIZE and hence has funny semantics -> Size will wrap
 * around. This should be fixed once the stream updates are working.
 *
 * stream_write() is saner
 */
void
stream_put (struct stream *s, const void *src, size_t size)
{

  /* XXX: CHECK_SIZE has strange semantics. It should be deprecated */
  CHECK_SIZE(s, size);
  
  STREAM_VERIFY_SANE(s);
  
  if (STREAM_WRITEABLE (s) < size)
    {
      STREAM_BOUND_WARN (s, "put");
      return;
    }
  
  if (src)
    memcpy (s->data + s->endp, src, size);
  else
    memset (s->data + s->endp, 0, size);

  s->endp += size;
}

/* Put character to the stream. */
int
stream_putc (struct stream *s, u_char c)
{
  STREAM_VERIFY_SANE(s);
  
  if (STREAM_WRITEABLE (s) < sizeof(u_char))
    {
      STREAM_BOUND_WARN (s, "put");
      return 0;
    }
  
  s->data[s->endp++] = c;
  return sizeof (u_char);
}

/* Put word to the stream. */
int
stream_putw (struct stream *s, u_int16_t w)
{
  STREAM_VERIFY_SANE (s);

  if (STREAM_WRITEABLE (s) < sizeof (u_int16_t))
    {
      STREAM_BOUND_WARN (s, "put");
      return 0;
    }
  
  s->data[s->endp++] = (u_char)(w >>  8);
  s->data[s->endp++] = (u_char) w;

  return 2;
}

/* Put long word to the stream. */
int
stream_put3 (struct stream *s, u_int32_t l)
{
  STREAM_VERIFY_SANE (s);

  if (STREAM_WRITEABLE (s) < 3)
    {
      STREAM_BOUND_WARN (s, "put");
      return 0;
    }
  
  s->data[s->endp++] = (u_char)(l >> 16);
  s->data[s->endp++] = (u_char)(l >>  8);
  s->data[s->endp++] = (u_char)l;

  return 3;
}

/* Put long word to the stream. */
int
stream_putl (struct stream *s, u_int32_t l)
{
  STREAM_VERIFY_SANE (s);

  if (STREAM_WRITEABLE (s) < sizeof (u_int32_t))
    {
      STREAM_BOUND_WARN (s, "put");
      return 0;
    }
  
  s->data[s->endp++] = (u_char)(l >> 24);
  s->data[s->endp++] = (u_char)(l >> 16);
  s->data[s->endp++] = (u_char)(l >>  8);
  s->data[s->endp++] = (u_char)l;

  return 4;
}

/* Put quad word to the stream. */
int
stream_putq (struct stream *s, uint64_t q)
{
  STREAM_VERIFY_SANE (s);

  if (STREAM_WRITEABLE (s) < sizeof (uint64_t))
    {
      STREAM_BOUND_WARN (s, "put quad");
      return 0;
    }
  
  s->data[s->endp++] = (u_char)(q >> 56);
  s->data[s->endp++] = (u_char)(q >> 48);
  s->data[s->endp++] = (u_char)(q >> 40);
  s->data[s->endp++] = (u_char)(q >> 32);
  s->data[s->endp++] = (u_char)(q >> 24);
  s->data[s->endp++] = (u_char)(q >> 16);
  s->data[s->endp++] = (u_char)(q >>  8);
  s->data[s->endp++] = (u_char)q;

  return 8;
}

int
stream_putf (struct stream *s, float f)
{
  union {
    float i;
    uint32_t o;
  } u;
  u.i = f;
  return stream_putl (s, u.o);
}

int
stream_putd (struct stream *s, double d)
{
  union {
    double i;
    uint64_t o;
  } u;
  u.i = d;
  return stream_putq (s, u.o);
}

int
stream_putc_at (struct stream *s, size_t putp, u_char c)
{
  STREAM_VERIFY_SANE(s);
  
  if (!PUT_AT_VALID (s, putp + sizeof (u_char)))
    {
      STREAM_BOUND_WARN (s, "put");
      return 0;
    }
  
  s->data[putp] = c;
  
  return 1;
}

int
stream_putw_at (struct stream *s, size_t putp, u_int16_t w)
{
  STREAM_VERIFY_SANE(s);
  
  if (!PUT_AT_VALID (s, putp + sizeof (u_int16_t)))
    {
      STREAM_BOUND_WARN (s, "put");
      return 0;
    }
  
  s->data[putp] = (u_char)(w >>  8);
  s->data[putp + 1] = (u_char) w;
  
  return 2;
}

int
stream_put3_at (struct stream *s, size_t putp, u_int32_t l)
{
  STREAM_VERIFY_SANE(s);
  
  if (!PUT_AT_VALID (s, putp + 3))
    {
      STREAM_BOUND_WARN (s, "put");
      return 0;
    }
  s->data[putp] = (u_char)(l >> 16);
  s->data[putp + 1] = (u_char)(l >>  8);
  s->data[putp + 2] = (u_char)l;
  
  return 3;
}

int
stream_putl_at (struct stream *s, size_t putp, u_int32_t l)
{
  STREAM_VERIFY_SANE(s);
  
  if (!PUT_AT_VALID (s, putp + sizeof (u_int32_t)))
    {
      STREAM_BOUND_WARN (s, "put");
      return 0;
    }
  s->data[putp] = (u_char)(l >> 24);
  s->data[putp + 1] = (u_char)(l >> 16);
  s->data[putp + 2] = (u_char)(l >>  8);
  s->data[putp + 3] = (u_char)l;
  
  return 4;
}

int
stream_putq_at (struct stream *s, size_t putp, uint64_t q)
{
  STREAM_VERIFY_SANE(s);
  
  if (!PUT_AT_VALID (s, putp + sizeof (uint64_t)))
    {
      STREAM_BOUND_WARN (s, "put");
      return 0;
    }
  s->data[putp] =     (u_char)(q >> 56);
  s->data[putp + 1] = (u_char)(q >> 48);
  s->data[putp + 2] = (u_char)(q >> 40);
  s->data[putp + 3] = (u_char)(q >> 32);
  s->data[putp + 4] = (u_char)(q >> 24);
  s->data[putp + 5] = (u_char)(q >> 16);
  s->data[putp + 6] = (u_char)(q >>  8);
  s->data[putp + 7] = (u_char)q;
  
  return 8;
}

/* Put long word to the stream. */
int
stream_put_ipv4 (struct stream *s, u_int32_t l)
{
  STREAM_VERIFY_SANE(s);
  
  if (STREAM_WRITEABLE (s) < sizeof (u_int32_t))
    {
      STREAM_BOUND_WARN (s, "put");
      return 0;
    }
  memcpy (s->data + s->endp, &l, sizeof (u_int32_t));
  s->endp += sizeof (u_int32_t);

  return sizeof (u_int32_t);
}

/* Put long word to the stream. */
int
stream_put_in_addr (struct stream *s, struct in_addr *addr)
{
  STREAM_VERIFY_SANE(s);
  
  if (STREAM_WRITEABLE (s) < sizeof (u_int32_t))
    {
      STREAM_BOUND_WARN (s, "put");
      return 0;
    }

  memcpy (s->data + s->endp, addr, sizeof (u_int32_t));
  s->endp += sizeof (u_int32_t);

  return sizeof (u_int32_t);
}

/* Put in_addr at location in the stream. */
int
stream_put_in_addr_at (struct stream *s, size_t putp, struct in_addr *addr)
{
  STREAM_VERIFY_SANE(s);

  if (!PUT_AT_VALID (s, putp + 4))
    {
      STREAM_BOUND_WARN (s, "put");
      return 0;
    }

  memcpy (&s->data[putp], addr, 4);
  return 4;
}

/* Put in6_addr at location in the stream. */
int
stream_put_in6_addr_at (struct stream *s, size_t putp, struct in6_addr *addr)
{
  STREAM_VERIFY_SANE(s);

  if (!PUT_AT_VALID (s, putp + 16))
    {
      STREAM_BOUND_WARN (s, "put");
      return 0;
    }

  memcpy (&s->data[putp], addr, 16);
  return 16;
}

/* Put prefix by nlri type format. */
int
stream_put_prefix_addpath (struct stream *s, struct prefix *p,
                           int addpath_encode, u_int32_t addpath_tx_id)
{
  size_t psize;
  size_t psize_with_addpath;
  
  STREAM_VERIFY_SANE(s);
  
  psize = PSIZE (p->prefixlen);

  if (addpath_encode)
    psize_with_addpath = psize + 4;
  else
    psize_with_addpath = psize;
  
  if (STREAM_WRITEABLE (s) < (psize_with_addpath + sizeof (u_char)))
    {
      STREAM_BOUND_WARN (s, "put");
      return 0;
    }
  
  if (addpath_encode)
    {
        s->data[s->endp++] = (u_char)(addpath_tx_id >> 24);
        s->data[s->endp++] = (u_char)(addpath_tx_id >> 16);
        s->data[s->endp++] = (u_char)(addpath_tx_id >>  8);
        s->data[s->endp++] = (u_char)addpath_tx_id;
    }

  s->data[s->endp++] = p->prefixlen;
  memcpy (s->data + s->endp, &p->u.prefix, psize);
  s->endp += psize;
  
  return psize;
}

int
stream_put_prefix (struct stream *s, struct prefix *p)
{
  return stream_put_prefix_addpath (s, p, 0, 0);
}

/* Put NLRI with label */
int
stream_put_labeled_prefix (struct stream *s, struct prefix *p, u_char *label)
{
  size_t psize;

  STREAM_VERIFY_SANE(s);

  psize = PSIZE (p->prefixlen);

  if (STREAM_WRITEABLE (s) < (psize + 3))
    {
      STREAM_BOUND_WARN (s, "put");
      return 0;
    }

  stream_putc (s, (p->prefixlen + 24));
  stream_putc(s, label[0]);
  stream_putc(s, label[1]);
  stream_putc(s, label[2]);
  memcpy (s->data + s->endp, &p->u.prefix, psize);
  s->endp += psize;

  return (psize + 3);
}

/* Read size from fd. */
int
stream_read (struct stream *s, int fd, size_t size)
{
  int nbytes;

  STREAM_VERIFY_SANE(s);
  
  if (STREAM_WRITEABLE (s) < size)
    {
      STREAM_BOUND_WARN (s, "put");
      return 0;
    }
  
  nbytes = readn (fd, s->data + s->endp, size);

  if (nbytes > 0)
    s->endp += nbytes;
  
  return nbytes;
}

ssize_t
stream_read_try(struct stream *s, int fd, size_t size)
{
  ssize_t nbytes;

  STREAM_VERIFY_SANE(s);
  
  if (STREAM_WRITEABLE(s) < size)
    {
      STREAM_BOUND_WARN (s, "put");
      /* Fatal (not transient) error, since retrying will not help
         (stream is too small to contain the desired data). */
      return -1;
    }

  if ((nbytes = read(fd, s->data + s->endp, size)) >= 0)
    {
      s->endp += nbytes;
      return nbytes;
    }
  /* Error: was it transient (return -2) or fatal (return -1)? */
  if (ERRNO_IO_RETRY(errno))
    return -2;
  zlog_warn("%s: read failed on fd %d: %s", __func__, fd, safe_strerror(errno));
  return -1;
}

/* Read up to size bytes into the stream from the fd, using recvmsgfrom
 * whose arguments match the remaining arguments to this function
 */
ssize_t 
stream_recvfrom (struct stream *s, int fd, size_t size, int flags,
                 struct sockaddr *from, socklen_t *fromlen)                     
{
  ssize_t nbytes;

  STREAM_VERIFY_SANE(s);
  
  if (STREAM_WRITEABLE(s) < size)
    {
      STREAM_BOUND_WARN (s, "put");
      /* Fatal (not transient) error, since retrying will not help
         (stream is too small to contain the desired data). */
      return -1;
    }

  if ((nbytes = recvfrom (fd, s->data + s->endp, size, 
                          flags, from, fromlen)) >= 0)
    {
      s->endp += nbytes;
      return nbytes;
    }
  /* Error: was it transient (return -2) or fatal (return -1)? */
  if (ERRNO_IO_RETRY(errno))
    return -2;
  zlog_warn("%s: read failed on fd %d: %s", __func__, fd, safe_strerror(errno));
  return -1;
}

/* Read up to smaller of size or SIZE_REMAIN() bytes to the stream, starting
 * from endp.
 * First iovec will be used to receive the data.
 * Stream need not be empty.
 */
ssize_t
stream_recvmsg (struct stream *s, int fd, struct msghdr *msgh, int flags, 
                size_t size)
{
  int nbytes;
  struct iovec *iov;
  
  STREAM_VERIFY_SANE(s);
  assert (msgh->msg_iovlen > 0);  
  
  if (STREAM_WRITEABLE (s) < size)
    {
      STREAM_BOUND_WARN (s, "put");
      /* This is a logic error in the calling code: the stream is too small
         to hold the desired data! */
      return -1;
    }
  
  iov = &(msgh->msg_iov[0]);
  iov->iov_base = (s->data + s->endp);
  iov->iov_len = size;
  
  nbytes = recvmsg (fd, msgh, flags);
  
  if (nbytes > 0)
    s->endp += nbytes;
  
  return nbytes;
}
  
/* Write data to buffer. */
size_t
stream_write (struct stream *s, const void *ptr, size_t size)
{

  CHECK_SIZE(s, size);

  STREAM_VERIFY_SANE(s);
  
  if (STREAM_WRITEABLE (s) < size)
    {
      STREAM_BOUND_WARN (s, "put");
      return 0;
    }
  
  memcpy (s->data + s->endp, ptr, size);
  s->endp += size;

  return size;
}

/* Return current read pointer. 
 * DEPRECATED!
 * Use stream_get_pnt_to if you must, but decoding streams properly
 * is preferred
 */
u_char *
stream_pnt (struct stream *s)
{
  STREAM_VERIFY_SANE(s);
  return s->data + s->getp;
}

/* Check does this stream empty? */
int
stream_empty (struct stream *s)
{
  STREAM_VERIFY_SANE(s);

  return (s->endp == 0);
}

/* Reset stream. */
void
stream_reset (struct stream *s)
{
  STREAM_VERIFY_SANE (s);

  s->getp = s->endp = 0;
}

/* Write stream contens to the file discriptor. */
int
stream_flush (struct stream *s, int fd)
{
  int nbytes;
  
  STREAM_VERIFY_SANE(s);
  
  nbytes = write (fd, s->data + s->getp, s->endp - s->getp);
  
  return nbytes;
}

/* Stream first in first out queue. */

struct stream_fifo *
stream_fifo_new (void)
{
  struct stream_fifo *new;
 
  new = XCALLOC (MTYPE_STREAM_FIFO, sizeof (struct stream_fifo));
  return new;
}

/* Add new stream to fifo. */
void
stream_fifo_push (struct stream_fifo *fifo, struct stream *s)
{
  if (fifo->tail)
    fifo->tail->next = s;
  else
    fifo->head = s;
     
  fifo->tail = s;

  fifo->count++;
}

/* Delete first stream from fifo. */
struct stream *
stream_fifo_pop (struct stream_fifo *fifo)
{
  struct stream *s;
  
  s = fifo->head; 

  if (s)
    { 
      fifo->head = s->next;

      if (fifo->head == NULL)
	fifo->tail = NULL;

      fifo->count--;
    }

  return s; 
}

/* Return first fifo entry. */
struct stream *
stream_fifo_head (struct stream_fifo *fifo)
{
  return fifo->head;
}

void
stream_fifo_clean (struct stream_fifo *fifo)
{
  struct stream *s;
  struct stream *next;

  for (s = fifo->head; s; s = next)
    {
      next = s->next;
      stream_free (s);
    }
  fifo->head = fifo->tail = NULL;
  fifo->count = 0;
}

void
stream_fifo_free (struct stream_fifo *fifo)
{
  stream_fifo_clean (fifo);
  XFREE (MTYPE_STREAM_FIFO, fifo);
}