/**
 * Copyright (c) 2016 rxi
 *
 * This library is free software; you can redistribute it and/or modify it
 * under the terms of the MIT license. See LICENSE for details.
 */

#ifdef _WIN32
  #define _WIN32_WINNT 0x501
  #ifndef _CRT_SECURE_NO_WARNINGS
    #define _CRT_SECURE_NO_WARNINGS
  #endif
  #include <winsock2.h>
  #include <ws2tcpip.h>
  #include <windows.h>
#else
  #define _POSIX_C_SOURCE 200809L
  #ifdef __APPLE__
    #define _DARWIN_UNLIMITED_SELECT
  #endif
  #include <unistd.h>
  #include <netdb.h>
  #include <fcntl.h>
  #include <sys/types.h>
  #include <sys/socket.h>
  #include <sys/time.h>
  #include <netinet/in.h>
  #include <netinet/tcp.h>
  #include <arpa/inet.h>
#endif
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <stdarg.h>
#include <signal.h>
#include <errno.h>
#include <limits.h>

#include "dyad.h"

#define DYAD_VERSION "0.2.1"


#ifdef _WIN32
  #define close(a) closesocket(a)
  #define getsockopt(a,b,c,d,e) getsockopt((a),(b),(c),(char*)(d),(e))
  #define setsockopt(a,b,c,d,e) setsockopt((a),(b),(c),(char*)(d),(e))
  #define select(a,b,c,d,e) select((int)(a),(b),(c),(d),(e))
  #define bind(a,b,c) bind((a),(b),(int)(c))
  #define connect(a,b,c) connect((a),(b),(int)(c))

  #undef  errno
  #define errno WSAGetLastError()

  #undef  EWOULDBLOCK
  #define EWOULDBLOCK WSAEWOULDBLOCK

  const char *inet_ntop(int af, const void *src, char *dst, socklen_t size) {
    union { struct sockaddr sa; struct sockaddr_in sai;
            struct sockaddr_in6 sai6; } addr;
    int res;
    memset(&addr, 0, sizeof(addr));
    addr.sa.sa_family = af;
    if (af == AF_INET6) {
      memcpy(&addr.sai6.sin6_addr, src, sizeof(addr.sai6.sin6_addr));
    } else {
      memcpy(&addr.sai.sin_addr, src, sizeof(addr.sai.sin_addr));
    }
    res = WSAAddressToStringA(&addr.sa, sizeof(addr), 0, dst, (LPDWORD) &size);
    if (res != 0) return NULL;
    return dst;
  }
#endif

#ifndef INVALID_SOCKET
  #define INVALID_SOCKET -1
#endif


/*===========================================================================*/
/* Memory                                                                    */
/*===========================================================================*/

static void panic(const char *fmt, ...);

static void *dyad_realloc(void *ptr, int n) {
  ptr = realloc(ptr, n);
  if (!ptr && n != 0) {
    panic("out of memory");
  }
  return ptr;
}


static void dyad_free(void *ptr) {
  free(ptr);
}


/*===========================================================================*/
/* Vec (dynamic array)                                                       */
/*===========================================================================*/

static void vec_expand(char **data, int *length, int *capacity, int memsz) {
  if (*length + 1 > *capacity) {
    if (*capacity == 0) {
      *capacity = 1;
    } else {
      *capacity <<= 1;
    }
    *data = dyad_realloc(*data, *capacity * memsz);
  }
}

static void vec_splice(
  char **data, int *length, int *capacity, int memsz, int start, int count
) {
  (void) capacity;
  memmove(*data + start * memsz,
          *data + (start + count) * memsz,
          (*length - start - count) * memsz);
}


#define Vec(T)\
  struct { T *data; int length, capacity; }


#define vec_unpack(v)\
  (char**)&(v)->data, &(v)->length, &(v)->capacity, sizeof(*(v)->data)


#define vec_init(v)\
  memset((v), 0, sizeof(*(v)))


#define vec_deinit(v)\
  dyad_free((v)->data)


#define vec_clear(v)\
  ((v)->length = 0)


#define vec_push(v, val)\
  ( vec_expand(vec_unpack(v)),\
    (v)->data[(v)->length++] = (val) )


#define vec_splice(v, start, count)\
  ( vec_splice(vec_unpack(v), start, count),\
    (v)->length -= (count) )



/*===========================================================================*/
/* SelectSet                                                                 */
/*===========================================================================*/

/* A wrapper around the three fd_sets used for select(). The fd_sets' allocated
 * memory is automatically expanded to accommodate fds as they are added.
 *
 * On Windows fd_sets are implemented as arrays; the FD_xxx macros are not used
 * by the wrapper and instead the fd_set struct is manipulated directly. The
 * wrapper should perform better than the normal FD_xxx macros, given that we
 * don't bother with the linear search which FD_SET would perform to check for
 * duplicates.
 *
 * On non-Windows platforms the sets are assumed to be bit arrays. The FD_xxx
 * macros are not used in case their implementation attempts to do bounds
 * checking; instead we manipulate the fd_sets' bits directly.
 */

enum {
  SELECT_READ,
  SELECT_WRITE,
  SELECT_EXCEPT,
  SELECT_MAX
};

typedef struct {
  int capacity;
  dyad_Socket maxfd;
  fd_set *fds[SELECT_MAX];
} SelectSet;

#define DYAD_UNSIGNED_BIT (sizeof(unsigned) * CHAR_BIT)


static void select_deinit(SelectSet *s) {
  int i;
  for (i = 0; i < SELECT_MAX; i++) {
    dyad_free(s->fds[i]);
    s->fds[i] = NULL;
  }
  s->capacity = 0;
}


static void select_grow(SelectSet *s) {
  int i;
  int oldCapacity = s->capacity;
  s->capacity = s->capacity ? s->capacity << 1 : 1;
  for (i = 0; i < SELECT_MAX; i++) {
    s->fds[i] = dyad_realloc(s->fds[i], s->capacity * sizeof(fd_set));
    memset(s->fds[i] + oldCapacity, 0,
           (s->capacity - oldCapacity) * sizeof(fd_set));
  }
}


static void select_zero(SelectSet *s) {
  int i;
  if (s->capacity == 0) return;
  s->maxfd = 0;
  for (i = 0; i < SELECT_MAX; i++) {
#if _WIN32
    s->fds[i]->fd_count = 0;
#else
    memset(s->fds[i], 0, s->capacity * sizeof(fd_set));
#endif
  }
}


static void select_add(SelectSet *s, int set, dyad_Socket fd) {
#ifdef _WIN32
  fd_set *f;
  if (s->capacity == 0) select_grow(s);
  while ((unsigned) (s->capacity * FD_SETSIZE) < s->fds[set]->fd_count + 1) {
    select_grow(s);
  }
  f = s->fds[set];
  f->fd_array[f->fd_count++] = fd;
#else
  unsigned *p;
  while (s->capacity * FD_SETSIZE < fd) {
    select_grow(s);
  }
  p = (unsigned*) s->fds[set];
  p[fd / DYAD_UNSIGNED_BIT] |= 1 << (fd % DYAD_UNSIGNED_BIT);
  if (fd > s->maxfd) s->maxfd = fd;
#endif
}


static int select_has(SelectSet *s, int set, dyad_Socket fd) {
#ifdef _WIN32
  unsigned i;
  fd_set *f;
  if (s->capacity == 0) return 0;
  f = s->fds[set];
  for (i = 0; i < f->fd_count; i++) {
    if (f->fd_array[i] == fd) {
      return 1;
    }
  }
  return 0;
#else
  unsigned *p;
  if (s->maxfd < fd) return 0;
  p = (unsigned*) s->fds[set];
  return p[fd / DYAD_UNSIGNED_BIT] & (1 << (fd % DYAD_UNSIGNED_BIT));
#endif
}


/*===========================================================================*/
/* Core                                                                      */
/*===========================================================================*/

typedef struct {
  int event;
  dyad_Callback callback;
  void *udata;
} Listener;


struct dyad_Stream {
  int state, flags;
  dyad_Socket sockfd;
  char *address;
  int port;
  int bytesSent, bytesReceived;
  double lastActivity, timeout;
  Vec(Listener) listeners;
  Vec(char) lineBuffer;
  Vec(char) writeBuffer;
  dyad_Stream *next;
};

#define DYAD_FLAG_READY   (1 << 0)
#define DYAD_FLAG_WRITTEN (1 << 1)


static dyad_Stream *dyad_streams;
static int dyad_streamCount;
static char dyad_panicMsgBuffer[128];
static dyad_PanicCallback panicCallback;
static SelectSet dyad_selectSet;
static double dyad_updateTimeout = 1;
static double dyad_tickInterval = 1;
static double dyad_lastTick = 0;


static void panic(const char *fmt, ...) {
  va_list args;
  va_start(args, fmt);
  vsprintf(dyad_panicMsgBuffer, fmt, args);
  va_end(args);
  if (panicCallback) {
    panicCallback(dyad_panicMsgBuffer);
  } else {
    printf("dyad panic: %s\n", dyad_panicMsgBuffer);
  }
  exit(EXIT_FAILURE);
}


static dyad_Event createEvent(int type) {
  dyad_Event e;
  memset(&e, 0, sizeof(e));
  e.type = type;
  return e;
}


static void stream_destroy(dyad_Stream *stream);

static void destroyClosedStreams(void) {
  dyad_Stream *stream = dyad_streams;
  while (stream) {
    if (stream->state == DYAD_STATE_CLOSED) {
      dyad_Stream *next = stream->next;
      stream_destroy(stream);
      stream = next;
    } else {
      stream = stream->next;
    }
  }
}


static void stream_emitEvent(dyad_Stream *stream, dyad_Event *e);

static void updateTickTimer(void) {
  /* Update tick timer */
  if (dyad_lastTick == 0) {
    dyad_lastTick = dyad_getTime();
  }
  while (dyad_lastTick < dyad_getTime()) {
    /* Emit event on all streams */
    dyad_Stream *stream;
    dyad_Event e = createEvent(DYAD_EVENT_TICK);
    e.msg = "a tick has occured";
    stream = dyad_streams;
    while (stream) {
      stream_emitEvent(stream, &e);
      stream = stream->next;
    }
    dyad_lastTick += dyad_tickInterval;
  }
}


static void updateStreamTimeouts(void) {
  double currentTime = dyad_getTime();
  dyad_Stream *stream;
  dyad_Event e = createEvent(DYAD_EVENT_TIMEOUT);
  e.msg = "stream timed out";
  stream = dyad_streams;
  while (stream) {
    if (stream->timeout) {
      if (currentTime - stream->lastActivity > stream->timeout) {
        stream_emitEvent(stream, &e);
        dyad_close(stream);
      }
    }
    stream = stream->next;
  }
}



/*===========================================================================*/
/* Stream                                                                    */
/*===========================================================================*/

static void stream_destroy(dyad_Stream *stream) {
  dyad_Event e;
  dyad_Stream **next;
  /* Close socket */
  if (stream->sockfd != INVALID_SOCKET) {
    close(stream->sockfd);
  }
  /* Emit destroy event */
  e = createEvent(DYAD_EVENT_DESTROY);
  e.msg = "the stream has been destroyed";
  stream_emitEvent(stream, &e);
  /* Remove from list and decrement count */
  next = &dyad_streams;
  while (*next != stream) {
    next = &(*next)->next;
  }
  *next = stream->next;
  dyad_streamCount--;
  /* Destroy and free */
  vec_deinit(&stream->listeners);
  vec_deinit(&stream->lineBuffer);
  vec_deinit(&stream->writeBuffer);
  dyad_free(stream->address);
  dyad_free(stream);
}


static void stream_emitEvent(dyad_Stream *stream, dyad_Event *e) {
  int i;
  e->stream = stream;
  for (i = 0; i < stream->listeners.length; i++) {
    Listener *listener = &stream->listeners.data[i];
    if (listener->event == e->type) {
      e->udata = listener->udata;
      listener->callback(e);
    }
    /* Check to see if this listener was removed: If it was we decrement `i`
     * since the next listener will now be in this ones place */
    if (listener != &stream->listeners.data[i]) {
      i--;
    }
  }
}


static void stream_error(dyad_Stream *stream, const char *msg, int err) {
  char buf[256];
  dyad_Event e = createEvent(DYAD_EVENT_ERROR);
  if (err) {
    sprintf(buf, "%.160s (%.80s)", msg, strerror(err));
    e.msg = buf;
  } else {
    e.msg = msg;
  }
  stream_emitEvent(stream, &e);
  dyad_close(stream);
}


static void stream_initAddress(dyad_Stream *stream) {
  union { struct sockaddr sa; struct sockaddr_storage sas;
          struct sockaddr_in sai; struct sockaddr_in6 sai6; } addr;
  socklen_t size;
  memset(&addr, 0, sizeof(addr));
  size = sizeof(addr);
  dyad_free(stream->address);
  stream->address = NULL;
  if (getpeername(stream->sockfd, &addr.sa, &size) == -1) {
    if (getsockname(stream->sockfd, &addr.sa, &size) == -1) {
      return;
    }
  }
  if (addr.sas.ss_family == AF_INET6) {
    stream->address = dyad_realloc(NULL, INET6_ADDRSTRLEN);
    inet_ntop(AF_INET6, &addr.sai6.sin6_addr, stream->address,
              INET6_ADDRSTRLEN);
    stream->port = ntohs(addr.sai6.sin6_port);
  } else {
    stream->address = dyad_realloc(NULL, INET_ADDRSTRLEN);
    inet_ntop(AF_INET, &addr.sai.sin_addr, stream->address, INET_ADDRSTRLEN);
    stream->port = ntohs(addr.sai.sin_port);
  }
}


static void stream_setSocketNonBlocking(dyad_Stream *stream, int opt) {
#ifdef _WIN32
  u_long mode = opt;
  ioctlsocket(stream->sockfd, FIONBIO, &mode);
#else
  int flags = fcntl(stream->sockfd, F_GETFL);
  fcntl(stream->sockfd, F_SETFL,
        opt ? (flags | O_NONBLOCK) : (flags & ~O_NONBLOCK));
#endif
}


static void stream_setSocket(dyad_Stream *stream, dyad_Socket sockfd) {
  stream->sockfd = sockfd;
  stream_setSocketNonBlocking(stream, 1);
  stream_initAddress(stream);
}


static int stream_initSocket(
  dyad_Stream *stream, int domain, int type, int protocol
) {
  stream->sockfd = socket(domain, type, protocol);
  if (stream->sockfd == INVALID_SOCKET) {
    stream_error(stream, "could not create socket", errno);
    return -1;
  }
  stream_setSocket(stream, stream->sockfd);
  return 0;
}


static int stream_hasListenerForEvent(dyad_Stream *stream, int event) {
  int i;
  for (i = 0; i < stream->listeners.length; i++) {
    Listener *listener = &stream->listeners.data[i];
    if (listener->event == event) {
      return 1;
    }
  }
  return 0;
}


static void stream_handleReceivedData(dyad_Stream *stream) {
  for (;;) {
    /* Receive data */
    dyad_Event e;
    char data[8192];
    int size = recv(stream->sockfd, data, sizeof(data) - 1, 0);
    if (size <= 0) {
      if (size == 0 || errno != EWOULDBLOCK) {
        /* Handle disconnect */
        dyad_close(stream);
        return;
      } else {
        /* No more data */
        return;
      }
    }
    data[size] = 0;
    /* Update status */
    stream->bytesReceived += size;
    stream->lastActivity = dyad_getTime();
    /* Emit data event */
    e = createEvent(DYAD_EVENT_DATA);
    e.msg = "received data";
    e.data = data;
    e.size = size;
    stream_emitEvent(stream, &e);
    /* Check stream state in case it was closed during one of the data event
     * handlers. */
    if (stream->state != DYAD_STATE_CONNECTED) {
      return;
    }

    /* Handle line event */
    if (stream_hasListenerForEvent(stream, DYAD_EVENT_LINE)) {
      int i, start;
      char *buf;
      for (i = 0; i < size; i++) {
        vec_push(&stream->lineBuffer, data[i]);
      }
      start = 0;
      buf = stream->lineBuffer.data;
      for (i = 0; i < stream->lineBuffer.length; i++) {
        if (buf[i] == '\n') {
          dyad_Event e;
          buf[i] = '\0';
          e = createEvent(DYAD_EVENT_LINE);
          e.msg = "received line";
          e.data = &buf[start];
          e.size = i - start;
          /* Check and strip carriage return */
          if (e.size > 0 && e.data[e.size - 1] == '\r') {
            e.data[--e.size] = '\0';
          }
          stream_emitEvent(stream, &e);
          start = i + 1;
          /* Check stream state in case it was closed during one of the line
           * event handlers. */
          if (stream->state != DYAD_STATE_CONNECTED) {
            return;
          }
        }
      }
      if (start == stream->lineBuffer.length) {
        vec_clear(&stream->lineBuffer);
      } else {
        vec_splice(&stream->lineBuffer, 0, start);
      }
    }
  }
}


static void stream_acceptPendingConnections(dyad_Stream *stream) {
  for (;;) {
    dyad_Stream *remote;
    dyad_Event e;
    int err = 0;
    dyad_Socket sockfd = accept(stream->sockfd, NULL, NULL);
    if (sockfd == INVALID_SOCKET) {
      err = errno;
      if (err == EWOULDBLOCK) {
        /* No more waiting sockets */
        return;
      }
    }
    /* Create client stream */
    remote = dyad_newStream();
    remote->state = DYAD_STATE_CONNECTED;
    /* Set stream's socket */
    stream_setSocket(remote, sockfd);
    /* Emit accept event */
    e = createEvent(DYAD_EVENT_ACCEPT);
    e.msg = "accepted connection";
    e.remote = remote;
    stream_emitEvent(stream, &e);
    /* Handle invalid socket -- the stream is still made and the ACCEPT event
     * is still emitted, but its shut immediately with an error */
    if (remote->sockfd == INVALID_SOCKET) {
      stream_error(remote, "failed to create socket on accept", err);
      return;
    }
  }
}


static int stream_flushWriteBuffer(dyad_Stream *stream) {
  stream->flags &= ~DYAD_FLAG_WRITTEN;
  if (stream->writeBuffer.length > 0) {
    /* Send data */
    int size = send(stream->sockfd, stream->writeBuffer.data,
                    stream->writeBuffer.length, 0);
    if (size <= 0) {
      if (errno == EWOULDBLOCK) {
        /* No more data can be written */
        return 0;
      } else {
        /* Handle disconnect */
        dyad_close(stream);
        return 0;
      }
    }
    if (size == stream->writeBuffer.length) {
      vec_clear(&stream->writeBuffer);
    } else {
      vec_splice(&stream->writeBuffer, 0, size);
    }
    /* Update status */
    stream->bytesSent += size;
    stream->lastActivity = dyad_getTime();
  }

  if (stream->writeBuffer.length == 0) {
    dyad_Event e;
    /* If this is a 'closing' stream we can properly close it now */
    if (stream->state == DYAD_STATE_CLOSING) {
      dyad_close(stream);
      return 0;
    }
    /* Set ready flag and emit 'ready for data' event */
    stream->flags |= DYAD_FLAG_READY;
    e = createEvent(DYAD_EVENT_READY);
    e.msg = "stream is ready for more data";
    stream_emitEvent(stream, &e);
  }
  /* Return 1 to indicate that more data can immediately be written to the
   * stream's socket */
  return 1;
}



/*===========================================================================*/
/* API                                                                       */
/*===========================================================================*/

/*---------------------------------------------------------------------------*/
/* Core                                                                      */
/*---------------------------------------------------------------------------*/

void dyad_update(void) {
  dyad_Stream *stream;
  struct timeval tv;

  destroyClosedStreams();
  updateTickTimer();
  updateStreamTimeouts();

  /* Create fd sets for select() */
  select_zero(&dyad_selectSet);

  stream = dyad_streams;
  while (stream) {
    switch (stream->state) {
      case DYAD_STATE_CONNECTED:
        select_add(&dyad_selectSet, SELECT_READ, stream->sockfd);
        if (!(stream->flags & DYAD_FLAG_READY) ||
            stream->writeBuffer.length != 0
        ) {
          select_add(&dyad_selectSet, SELECT_WRITE, stream->sockfd);
        }
        break;
      case DYAD_STATE_CLOSING:
        select_add(&dyad_selectSet, SELECT_WRITE, stream->sockfd);
        break;
      case DYAD_STATE_CONNECTING:
        select_add(&dyad_selectSet, SELECT_WRITE, stream->sockfd);
        select_add(&dyad_selectSet, SELECT_EXCEPT, stream->sockfd);
        break;
      case DYAD_STATE_LISTENING:
        select_add(&dyad_selectSet, SELECT_READ, stream->sockfd);
        break;
    }
    stream = stream->next;
  }

  /* Init timeout value and do select */
  #ifdef _MSC_VER
    #pragma warning(push)
    /* Disable double to long implicit conversion warning,
     * because the type of timeval's fields don't agree across platforms */
    #pragma warning(disable: 4244)
  #endif
  tv.tv_sec = dyad_updateTimeout;
  tv.tv_usec = (dyad_updateTimeout - tv.tv_sec) * 1e6;
  #ifdef _MSC_VER
    #pragma warning(pop)
  #endif

  select(dyad_selectSet.maxfd + 1,
         dyad_selectSet.fds[SELECT_READ],
         dyad_selectSet.fds[SELECT_WRITE],
         dyad_selectSet.fds[SELECT_EXCEPT],
         &tv);

  /* Handle streams */
  stream = dyad_streams;
  while (stream) {
    switch (stream->state) {

      case DYAD_STATE_CONNECTED:
        if (select_has(&dyad_selectSet, SELECT_READ, stream->sockfd)) {
          stream_handleReceivedData(stream);
          if (stream->state == DYAD_STATE_CLOSED) {
            break;
          }
        }
        /* Fall through */

      case DYAD_STATE_CLOSING:
        if (select_has(&dyad_selectSet, SELECT_WRITE, stream->sockfd)) {
          stream_flushWriteBuffer(stream);
        }
        break;

      case DYAD_STATE_CONNECTING:
        if (select_has(&dyad_selectSet, SELECT_WRITE, stream->sockfd)) {
          /* Check socket for error */
          int optval = 0;
          socklen_t optlen = sizeof(optval);
          dyad_Event e;
          getsockopt(stream->sockfd, SOL_SOCKET, SO_ERROR, &optval, &optlen);
          if (optval != 0) goto connectFailed;
          /* Handle succeselful connection */
          stream->state = DYAD_STATE_CONNECTED;
          stream->lastActivity = dyad_getTime();
          stream_initAddress(stream);
          /* Emit connect event */
          e = createEvent(DYAD_EVENT_CONNECT);
          e.msg = "connected to server";
          stream_emitEvent(stream, &e);
        } else if (
          select_has(&dyad_selectSet, SELECT_EXCEPT, stream->sockfd)
        ) {
          /* Handle failed connection */
connectFailed:
          stream_error(stream, "could not connect to server", 0);
        }
        break;

      case DYAD_STATE_LISTENING:
        if (select_has(&dyad_selectSet, SELECT_READ, stream->sockfd)) {
          stream_acceptPendingConnections(stream);
        }
        break;
    }

    /* If data was just now written to the stream we should immediately try to
     * send it */
    if (
      stream->flags & DYAD_FLAG_WRITTEN &&
      stream->state != DYAD_STATE_CLOSED
    ) {
      stream_flushWriteBuffer(stream);
    }

    stream = stream->next;
  }
}


void dyad_init(void) {
#ifdef _WIN32
  WSADATA dat;
  int err = WSAStartup(MAKEWORD(2, 2), &dat);
  if (err != 0) {
    panic("WSAStartup failed (%d)", err);
  }
#else
  /* Stops the SIGPIPE signal being raised when writing to a closed socket */
  signal(SIGPIPE, SIG_IGN);
#endif
}


void dyad_shutdown(void) {
  /* Close and destroy all the streams */
  while (dyad_streams) {
    dyad_close(dyad_streams);
    stream_destroy(dyad_streams);
  }
  /* Clear up everything */
  select_deinit(&dyad_selectSet);
#ifdef _WIN32
  WSACleanup();
#endif
}


const char *dyad_getVersion(void) {
  return DYAD_VERSION;
}


double dyad_getTime(void) {
#ifdef _WIN32
  FILETIME ft;
  GetSystemTimeAsFileTime(&ft);
  return (ft.dwHighDateTime * 4294967296.0 / 1e7) + ft.dwLowDateTime / 1e7;
#else
  struct timeval tv;
  gettimeofday(&tv, NULL);
  return tv.tv_sec + tv.tv_usec / 1e6;
#endif
}


int dyad_getStreamCount(void) {
  return dyad_streamCount;
}


void dyad_setTickInterval(double seconds) {
  dyad_tickInterval = seconds;
}


void dyad_setUpdateTimeout(double seconds) {
  dyad_updateTimeout = seconds;
}


dyad_PanicCallback dyad_atPanic(dyad_PanicCallback func) {
  dyad_PanicCallback old = panicCallback;
  panicCallback = func;
  return old;
}


/*---------------------------------------------------------------------------*/
/* Stream                                                                    */
/*---------------------------------------------------------------------------*/

dyad_Stream *dyad_newStream(void) {
  dyad_Stream *stream = dyad_realloc(NULL, sizeof(*stream));
  memset(stream, 0, sizeof(*stream));
  stream->state = DYAD_STATE_CLOSED;
  stream->sockfd = INVALID_SOCKET;
  stream->lastActivity = dyad_getTime();
  /* Add to list and increment count */
  stream->next = dyad_streams;
  dyad_streams = stream;
  dyad_streamCount++;
  return stream;
}


void dyad_addListener(
  dyad_Stream *stream, int event, dyad_Callback callback, void *udata
) {
  Listener listener;
  listener.event = event;
  listener.callback = callback;
  listener.udata = udata;
  vec_push(&stream->listeners, listener);
}


void dyad_removeListener(
  dyad_Stream *stream, int event, dyad_Callback callback, void *udata
) {
  int i = stream->listeners.length;
  while (i--) {
    Listener *x = &stream->listeners.data[i];
    if (x->event == event && x->callback == callback && x->udata == udata) {
      vec_splice(&stream->listeners, i, 1);
    }
  }
}


void dyad_removeAllListeners(dyad_Stream *stream, int event) {
  if (event == DYAD_EVENT_NULL) {
    vec_clear(&stream->listeners);
  } else {
    int i = stream->listeners.length;
    while (i--) {
      if (stream->listeners.data[i].event == event) {
        vec_splice(&stream->listeners, i, 1);
      }
    }
  }
}


void dyad_close(dyad_Stream *stream) {
  dyad_Event e;
  if (stream->state == DYAD_STATE_CLOSED) return;
  stream->state = DYAD_STATE_CLOSED;
  /* Close socket */
  if (stream->sockfd != INVALID_SOCKET) {
    close(stream->sockfd);
    stream->sockfd = INVALID_SOCKET;
  }
  /* Emit event */
  e = createEvent(DYAD_EVENT_CLOSE);
  e.msg = "stream closed";
  stream_emitEvent(stream, &e);
  /* Clear buffers */
  vec_clear(&stream->lineBuffer);
  vec_clear(&stream->writeBuffer);
}


void dyad_end(dyad_Stream *stream) {
  if (stream->state == DYAD_STATE_CLOSED) return;
  if (stream->writeBuffer.length > 0) {
    stream->state = DYAD_STATE_CLOSING;
  } else {
    dyad_close(stream);
  }
}


int dyad_listenEx(
  dyad_Stream *stream, const char *host, int port, int backlog
) {
  struct addrinfo hints, *ai = NULL;
  int err, optval;
  char buf[64];
  dyad_Event e;

  /* Get addrinfo */
  memset(&hints, 0, sizeof(hints));
  hints.ai_family = AF_UNSPEC;
  hints.ai_socktype = SOCK_STREAM;
  hints.ai_flags = AI_PASSIVE;
  sprintf(buf, "%d", port);
  err = getaddrinfo(host, buf, &hints, &ai);
  if (err) {
    stream_error(stream, "could not get addrinfo", errno);
    goto fail;
  }
  /* Init socket */
  err = stream_initSocket(stream, ai->ai_family, ai->ai_socktype,
                          ai->ai_protocol);
  if (err) goto fail;
  /* Set SO_REUSEADDR so that the socket can be immediately bound without
   * having to wait for any closed socket on the same port to timeout */
  optval = 1;
  setsockopt(stream->sockfd, SOL_SOCKET, SO_REUSEADDR,
             &optval, sizeof(optval));
  /* Bind and listen */
  err = bind(stream->sockfd, ai->ai_addr, ai->ai_addrlen);
  if (err) {
    stream_error(stream, "could not bind socket", errno);
    goto fail;
  }
  err = listen(stream->sockfd, backlog);
  if (err) {
    stream_error(stream, "socket failed on listen", errno);
    goto fail;
  }
  stream->state = DYAD_STATE_LISTENING;
  stream->port = port;
  stream_initAddress(stream);
  /* Emit listening event */
  e = createEvent(DYAD_EVENT_LISTEN);
  e.msg = "socket is listening";
  stream_emitEvent(stream, &e);
  freeaddrinfo(ai);
  return 0;
  fail:
  if (ai) freeaddrinfo(ai);
  return -1;
}


int dyad_listen(dyad_Stream *stream, int port) {
  return dyad_listenEx(stream, NULL, port, 511);
}


int dyad_connect(dyad_Stream *stream, const char *host, int port) {
  struct addrinfo hints, *ai = NULL;
  int err;
  char buf[64];

  /* Resolve host */
  memset(&hints, 0, sizeof(hints));
  hints.ai_family = AF_UNSPEC;
  hints.ai_socktype = SOCK_STREAM;
  sprintf(buf, "%d", port);
  err = getaddrinfo(host, buf, &hints, &ai);
  if (err) {
    stream_error(stream, "could not resolve host", 0);
    goto fail;
  }
  /* Start connecting */
  err = stream_initSocket(stream, ai->ai_family, ai->ai_socktype,
                          ai->ai_protocol);
  if (err) goto fail;
  connect(stream->sockfd, ai->ai_addr, ai->ai_addrlen);
  stream->state = DYAD_STATE_CONNECTING;
  freeaddrinfo(ai);
  return 0;
fail:
  if (ai) freeaddrinfo(ai);
  return -1;
}


void dyad_write(dyad_Stream *stream, const void *data, int size) {
  const char *p = data;
  while (size--) {
    vec_push(&stream->writeBuffer, *p++);
  }
  stream->flags |= DYAD_FLAG_WRITTEN;
}


void dyad_vwritef(dyad_Stream *stream, const char *fmt, va_list args) {
  char buf[512];
  char *str;
  char f[] = "%_";
  FILE *fp;
  int c;
  while (*fmt) {
    if (*fmt == '%') {
      fmt++;
      switch (*fmt) {
        case 'r':
          fp = va_arg(args, FILE*);
          if (fp == NULL) {
            str = "(null)";
            goto writeStr;
          }
          while ((c = fgetc(fp)) != EOF) {
            vec_push(&stream->writeBuffer, c);
          }
          break;
        case 'c':
          vec_push(&stream->writeBuffer, va_arg(args, int));
          break;
        case 's':
          str = va_arg(args, char*);
          if (str == NULL) str = "(null)";
          writeStr:
          while (*str) {
            vec_push(&stream->writeBuffer, *str++);
          }
          break;
        case 'b':
          str = va_arg(args, char*);
          c = va_arg(args, int);
          while (c--) {
            vec_push(&stream->writeBuffer, *str++);
          }
          break;
        default:
          f[1] = *fmt;
          switch (*fmt) {
            case 'f':
            case 'g': sprintf(buf, f, va_arg(args, double));    break;
            case 'd':
            case 'i': sprintf(buf, f, va_arg(args, int));       break;
            case 'x':
            case 'X': sprintf(buf, f, va_arg(args, unsigned));  break;
            case 'p': sprintf(buf, f, va_arg(args, void*));     break;
            default : buf[0] = *fmt; buf[1] = '\0';
          }
          str = buf;
          goto writeStr;
      }
    } else {
      vec_push(&stream->writeBuffer, *fmt);
    }
    fmt++;
  }
  stream->flags |= DYAD_FLAG_WRITTEN;
}


void dyad_writef(dyad_Stream *stream, const char *fmt, ...) {
  va_list args;
  va_start(args, fmt);
  dyad_vwritef(stream, fmt, args);
  va_end(args);
}


void dyad_setTimeout(dyad_Stream *stream, double seconds) {
  stream->timeout = seconds;
}


void dyad_setNoDelay(dyad_Stream *stream, int opt) {
  opt = !!opt;
  setsockopt(stream->sockfd, IPPROTO_TCP, TCP_NODELAY, &opt, sizeof(opt));
}


int dyad_getState(dyad_Stream *stream) {
  return stream->state;
}


const char *dyad_getAddress(dyad_Stream *stream) {
  return stream->address ? stream->address : "";
}


int dyad_getPort(dyad_Stream *stream) {
  return stream->port;
}


int dyad_getBytesSent(dyad_Stream *stream) {
  return stream->bytesSent;
}


int dyad_getBytesReceived(dyad_Stream *stream) {
  return stream->bytesReceived;
}


dyad_Socket dyad_getSocket(dyad_Stream *stream) {
  return stream->sockfd;
}