evt_tls_t *getSSL(evt_ctx_t *d_eng)
{
     evt_tls_t *con = malloc(sizeof(evt_tls_t));
     if ( !con ) {
	 return NULL;
     }
     memset( con, 0, sizeof *con);

     SSL *ssl  = SSL_new(d_eng->ctx);

     if ( !ssl ) {
	 return NULL;
     }
     con->ssl = ssl;

     //use default buf size for now.
     BIO_new_bio_pair(&(con->ssl_bio_), 0, &(con->app_bio_), 0);

     SSL_set_bio(con->ssl, con->ssl_bio_, con->ssl_bio_);

     QUEUE_INIT(&(con->q));
     QUEUE_INSERT_TAIL(&(d_eng->live_con), &(con->q));

     con->writer = d_eng->writer;

     return con;
}

/**
 * when new client connected.
 */
static void init_smts_client(smts_client_t *smts_client, uv_loop_t *loop)
{
	init_abstract_tcp_client((abstract_tcp_client_t*) smts_client, loop);
	QUEUE_INIT(&smts_client->queue);
	smts_client->session = NULL;
	smts_client->status = SMTS_CLIENT_ON_INIT;
}

void queue_push(Queue *queue, Event event)
{
  queue_item *item = malloc(sizeof(queue_item));
  item->item = event;
  QUEUE_INIT(&item->node);
  QUEUE_INSERT_TAIL(&queue->headtail, &item->node);
}

void uv__work_done(uv_async_t* handle) {
  struct uv__work* w;
  uv_loop_t* loop;
  QUEUE* q;
  QUEUE wq;
  int err;

  loop = container_of(handle, uv_loop_t, wq_async);
  QUEUE_INIT(&wq);

  uv_mutex_lock(&loop->wq_mutex);
  if (!QUEUE_EMPTY(&loop->wq)) {
    q = QUEUE_HEAD(&loop->wq);
    QUEUE_SPLIT(&loop->wq, q, &wq);
  }
  uv_mutex_unlock(&loop->wq_mutex);

  while (!QUEUE_EMPTY(&wq)) {
    q = QUEUE_HEAD(&wq);
    QUEUE_REMOVE(q);

    w = container_of(q, struct uv__work, wq);
    err = (w->work == uv__cancelled) ? -ECANCELED : 0;
    w->done(w, err);
  }
}

static void stack_push(QUEUE *queue, Token token)
{
  queue_item *item = malloc(sizeof(queue_item));
  item->item = token;
  QUEUE_INIT(&item->node);
  QUEUE_INSERT_HEAD(queue, &item->node);
}

yaml_emitter_initialize(yaml_emitter_t *emitter)
{
    assert(emitter);    /* Non-NULL emitter object expected. */

    memset(emitter, 0, sizeof(yaml_emitter_t));
    if (!BUFFER_INIT(emitter, emitter->buffer, OUTPUT_BUFFER_SIZE))
        goto error;
    if (!BUFFER_INIT(emitter, emitter->raw_buffer, OUTPUT_RAW_BUFFER_SIZE))
        goto error;
    if (!STACK_INIT(emitter, emitter->states, INITIAL_STACK_SIZE))
        goto error;
    if (!QUEUE_INIT(emitter, emitter->events, INITIAL_QUEUE_SIZE))
        goto error;
    if (!STACK_INIT(emitter, emitter->indents, INITIAL_STACK_SIZE))
        goto error;
    if (!STACK_INIT(emitter, emitter->tag_directives, INITIAL_STACK_SIZE))
        goto error;

    return 1;

error:

    BUFFER_DEL(emitter, emitter->buffer);
    BUFFER_DEL(emitter, emitter->raw_buffer);
    STACK_DEL(emitter, emitter->states);
    QUEUE_DEL(emitter, emitter->events);
    STACK_DEL(emitter, emitter->indents);
    STACK_DEL(emitter, emitter->tag_directives);

    return 0;
}

int uv__platform_loop_init(uv_loop_t* loop, int default_loop) {
  CFRunLoopSourceContext ctx;
  int r;

  if (uv__kqueue_init(loop))
    return -1;

  loop->cf_loop = NULL;
  if ((r = uv_mutex_init(&loop->cf_mutex)))
    return r;
  if ((r = uv_sem_init(&loop->cf_sem, 0)))
    return r;
  QUEUE_INIT(&loop->cf_signals);

  memset(&ctx, 0, sizeof(ctx));
  ctx.info = loop;
  ctx.perform = uv__cf_loop_cb;
  loop->cf_cb = CFRunLoopSourceCreate(NULL, 0, &ctx);

  if ((r = uv_thread_create(&loop->cf_thread, uv__cf_loop_runner, loop)))
    return r;

  /* Synchronize threads */
  uv_sem_wait(&loop->cf_sem);
  assert(ACCESS_ONCE(CFRunLoopRef, loop->cf_loop) != NULL);

  return 0;
}

static void uv__cf_loop_cb(void* arg) {
  uv_loop_t* loop;
  QUEUE* item;
  QUEUE split_head;
  uv__cf_loop_signal_t* s;

  loop = arg;

  uv_mutex_lock(&loop->cf_mutex);
  QUEUE_INIT(&split_head);
  if (!QUEUE_EMPTY(&loop->cf_signals)) {
    QUEUE* split_pos = QUEUE_HEAD(&loop->cf_signals);
    QUEUE_SPLIT(&loop->cf_signals, split_pos, &split_head);
  }
  uv_mutex_unlock(&loop->cf_mutex);

  while (!QUEUE_EMPTY(&split_head)) {
    item = QUEUE_HEAD(&split_head);

    s = QUEUE_DATA(item, uv__cf_loop_signal_t, member);

    /* This was a termination signal */
    if (s->cb == NULL)
      CFRunLoopStop(loop->cf_loop);
    else
      s->cb(s->arg);

    QUEUE_REMOVE(item);
    free(s);
  }
}

void uv__io_stop(uv_loop_t* loop, uv__io_t* w, unsigned int events) {
  assert(0 == (events & ~(UV__POLLIN | UV__POLLOUT)));
  assert(0 != events);

  if (w->fd == -1)
    return;

  assert(w->fd >= 0);

  /* Happens when uv__io_stop() is called on a handle that was never started. */
  if ((unsigned) w->fd >= loop->nwatchers)
    return;

  w->pevents &= ~events;

  if (w->pevents == 0) {
    QUEUE_REMOVE(&w->watcher_queue);
    QUEUE_INIT(&w->watcher_queue);

    if (loop->watchers[w->fd] != NULL) {
      assert(loop->watchers[w->fd] == w);
      assert(loop->nfds > 0);
      loop->watchers[w->fd] = NULL;
      loop->nfds--;
      w->events = 0;
    }
  }
  else if (QUEUE_EMPTY(&w->watcher_queue))
    QUEUE_INSERT_TAIL(&loop->watcher_queue, &w->watcher_queue);
}

void uv__io_start(uv_loop_t* loop, uv__io_t* w, unsigned int events) {
  assert(0 == (events & ~(UV__POLLIN | UV__POLLOUT)));
  assert(0 != events);
  assert(w->fd >= 0);
  assert(w->fd < INT_MAX);

  w->pevents |= events;
  maybe_resize(loop, w->fd + 1);

#if !defined(__sun)
  /* The event ports backend needs to rearm all file descriptors on each and
   * every tick of the event loop but the other backends allow us to
   * short-circuit here if the event mask is unchanged.
   */
  if (w->events == w->pevents) {
    if (w->events == 0 && !QUEUE_EMPTY(&w->watcher_queue)) {
      QUEUE_REMOVE(&w->watcher_queue);
      QUEUE_INIT(&w->watcher_queue);
    }
    return;
  }
#endif

  if (QUEUE_EMPTY(&w->watcher_queue))
    QUEUE_INSERT_TAIL(&loop->watcher_queue, &w->watcher_queue);

  if (loop->watchers[w->fd] == NULL) {
    loop->watchers[w->fd] = w;
    loop->nfds++;
  }
}

void uv_pipe_connect(uv_connect_t* req,
                    uv_pipe_t* handle,
                    const char* name,
                    uv_connect_cb cb) {
  struct sockaddr_un saddr;
  int saved_errno;
  int new_sock;
  int err;
  int r;

  saved_errno = errno;
  new_sock = (uv__stream_fd(handle) == -1);
  err = -1;

  if (new_sock)
    if ((handle->io_watcher.fd = uv__socket(AF_UNIX, SOCK_STREAM, 0)) == -1)
      goto out;

  memset(&saddr, 0, sizeof saddr);
  uv_strlcpy(saddr.sun_path, name, sizeof(saddr.sun_path));
  saddr.sun_family = AF_UNIX;

  do {
    r = connect(uv__stream_fd(handle),
                (struct sockaddr*)&saddr, sizeof saddr);
  }
  while (r == -1 && errno == EINTR);

  if (r == -1)
    if (errno != EINPROGRESS)
      goto out;

  if (new_sock)
    if (uv__stream_open((uv_stream_t*)handle,
                        uv__stream_fd(handle),
                        UV_STREAM_READABLE | UV_STREAM_WRITABLE))
      goto out;

  uv__io_start(handle->loop, &handle->io_watcher, UV__POLLIN | UV__POLLOUT);
  err = 0;

out:
  handle->delayed_error = err ? errno : 0; /* Passed to callback. */
  handle->connect_req = req;

  uv__req_init(handle->loop, req, UV_CONNECT);
  req->handle = (uv_stream_t*)handle;
  req->cb = cb;
  QUEUE_INIT(&req->queue);

  /* Force callback to run on next tick in case of error. */
  if (err != 0)
    uv__io_feed(handle->loop, &handle->io_watcher);

  /* Mimic the Windows pipe implementation, always
   * return 0 and let the callback handle errors.
   */
  errno = saved_errno;
}

static void ref_push(QUEUE *queue, void *ref, char v_type)
{
  queue_ref_item *item = malloc(sizeof(queue_ref_item));
  item->item.v_type = v_type;
  item->item.ref = ref;
  QUEUE_INIT(&item->node);
  QUEUE_INSERT_HEAD(queue, &item->node);
}

int uv__tcp_connect(uv_connect_t* req,
                    uv_tcp_t* handle,
                    const struct sockaddr* addr,
                    unsigned int addrlen,
                    uv_connect_cb cb) {
  int err;
  int r;

  assert(handle->type == UV_TCP);

  if (handle->connect_req != NULL)
    return -EALREADY;  /* FIXME(bnoordhuis) -EINVAL or maybe -EBUSY. */

  err = maybe_new_socket(handle,
                         addr->sa_family,
                         UV_STREAM_READABLE | UV_STREAM_WRITABLE);
  if (err)
    return err;

  handle->delayed_error = 0;

  do {
    errno = 0;
    r = connect(uv__stream_fd(handle), addr, addrlen);
  } while (r == -1 && errno == EINTR);

  /* We not only check the return value, but also check the errno != 0.
   * Because in rare cases connect() will return -1 but the errno
   * is 0 (for example, on Android 4.3, OnePlus phone A0001_12_150227)
   * and actually the tcp three-way handshake is completed.
   */
  if (r == -1 && errno != 0) {
    if (errno == EINPROGRESS)
      ; /* not an error */
    else if (errno == ECONNREFUSED)
    /* If we get a ECONNREFUSED wait until the next tick to report the
     * error. Solaris wants to report immediately--other unixes want to
     * wait.
     */
      handle->delayed_error = -errno;
    else
      return -errno;
  }

  uv__req_init(handle->loop, req, UV_CONNECT);
  req->cb = cb;
  req->handle = (uv_stream_t*) handle;
  QUEUE_INIT(&req->queue);
  handle->connect_req = req;

  uv__io_start(handle->loop, &handle->io_watcher, POLLOUT);

  if (handle->delayed_error)
    uv__io_feed(handle->loop, &handle->io_watcher);

  return 0;
}

int uv_chan_init(uv_chan_t *chan) {
    int r = uv_mutex_init(&chan->mutex);
    if (r == -1)
        return r;

    QUEUE_INIT(&chan->q);

    return uv_cond_init(&chan->cond);
}

void setup() {
    for(int i = 0; i < SIGNAL_COUNT; i++) {
        SIGNALS[i].writable = true;
        SIGNALS[i].received = false;
        SIGNALS[i].sendSame = true;
        SIGNALS[i].sendFrequency = 1;
        SIGNALS[i].sendClock = 0;
    }
    QUEUE_INIT(CanMessage, &bus.sendQueue);
}

/*
	initialize a process given its stack
	returns the pointer to the top of the stack (needed by activate)
*/
unsigned int *init_process(struct Process *proc, unsigned int size, int (*task)(void)) {
	/* this is necessary to run something in user mode */
	pipe_init(&(proc->msgs));
	QUEUE_INIT(proc->writers);
	proc->blocked = 0;
	proc->stack[size-16] = (unsigned int)task; /* (pc) program counter */
	proc->stack[size-15] = 0x10; /* (SPSR) saved state */
	proc->stack[size-14] = &yield_task;
	return proc->stack + size - 16;
}

uint32_t evt_queue_init(radio_evt_handler_t evt_handler)
{
    m_evt_handler = evt_handler;

    QUEUE_INIT(m_evt_queue, RADIO_EVT_QUEUE_SIZE, sizeof(radio_evt_type_t));

    NVIC_SetPriority(SWI0_IRQn, 2);
    NVIC_EnableIRQ(SWI0_IRQn);

    return SUCCESS;
}

static void uv_loop_init(uv_loop_t* loop) {
  /* Create an I/O completion port */
  loop->iocp = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 1);
  if (loop->iocp == NULL) {
    uv_fatal_error(GetLastError(), "CreateIoCompletionPort");
  }

  /* To prevent uninitialized memory access, loop->time must be intialized */
  /* to zero before calling uv_update_time for the first time. */
  loop->time = 0;
  uv_update_time(loop);

  QUEUE_INIT(&loop->handle_queue);
  QUEUE_INIT(&loop->active_reqs);
  loop->active_handles = 0;

  loop->pending_reqs_tail = NULL;

  loop->endgame_handles = NULL;

  RB_INIT(&loop->timers);

  loop->check_handles = NULL;
  loop->prepare_handles = NULL;
  loop->idle_handles = NULL;

  loop->next_prepare_handle = NULL;
  loop->next_check_handle = NULL;
  loop->next_idle_handle = NULL;

  memset(&loop->poll_peer_sockets, 0, sizeof loop->poll_peer_sockets);

  loop->active_tcp_streams = 0;
  loop->active_udp_streams = 0;

  loop->timer_counter = 0;
  loop->stop_flag = 0;

  loop->last_err = uv_ok_;
}

void tls__write_done_cb(uv_write_t* w, int status)
{
    tr_uv_wi_t* wi = NULL;
    int i;
    QUEUE* q;
    GET_TLS(w);

    tt->is_writing = 0;

    if (status) {
        pc_lib_log(PC_LOG_ERROR, "tcp__write_done_cb - uv_write callback error: %s", uv_strerror(status));
    }

    status = status ? PC_RC_ERROR : PC_RC_OK;

    pc_mutex_lock(&tt->wq_mutex);
    while(!QUEUE_EMPTY(&tt->writing_queue)) {
        q = QUEUE_HEAD(&tt->writing_queue); 
        QUEUE_REMOVE(q);
        QUEUE_INIT(q);

        wi = (tr_uv_wi_t* )QUEUE_DATA(q, tr_uv_wi_t, queue);

        if (!status && TR_UV_WI_IS_RESP(wi->type)) {
            pc_lib_log(PC_LOG_DEBUG, "tls__write_to_tcp - move wi from  writing queue to resp pending queue,"
                " seq_num: %u, req_id: %u", wi->seq_num, wi->req_id);
            QUEUE_INSERT_TAIL(&tt->resp_pending_queue, q);
            continue;
        };

        pc_lib_free(wi->buf.base);
        wi->buf.base = NULL;
        wi->buf.len = 0;

        if (TR_UV_WI_IS_NOTIFY(wi->type)) {
            pc_trans_sent(tt->client, wi->seq_num, status);
        }

        if (TR_UV_WI_IS_RESP(wi->type)) {
            pc_trans_resp(tt->client, wi->req_id, status, NULL);
        }
        // if internal, do nothing here.

        if (PC_IS_PRE_ALLOC(wi->type)) {
            PC_PRE_ALLOC_SET_IDLE(wi->type);
        } else {
            pc_lib_free(wi);
        }
    }
    pc_mutex_unlock(&tt->wq_mutex);
    tls__write_to_tcp(tls);
}

int uv_fs_event_start(uv_fs_event_t* handle,
                      uv_fs_event_cb cb,
                      const char* path,
                      unsigned int flags) {
  struct watcher_list* w;
  int events;
  int err;
  int wd;

  if (uv__is_active(handle))
    return -EINVAL;

  err = init_inotify(handle->loop);
  if (err)
    return err;

  events = UV__IN_ATTRIB
         | UV__IN_CREATE
         | UV__IN_MODIFY
         | UV__IN_DELETE
         | UV__IN_DELETE_SELF
         | UV__IN_MOVE_SELF
         | UV__IN_MOVED_FROM
         | UV__IN_MOVED_TO;

  wd = uv__inotify_add_watch(handle->loop->inotify_fd, path, events);
  if (wd == -1)
    return -errno;

  w = find_watcher(handle->loop, wd);
  if (w)
    goto no_insert;

  w = uv__malloc(sizeof(*w) + strlen(path) + 1);
  if (w == NULL)
    return -ENOMEM;

  w->wd = wd;
  w->path = strcpy((char*)(w + 1), path);
  QUEUE_INIT(&w->watchers);
  w->iterating = 0;
  RB_INSERT(watcher_root, CAST(&handle->loop->inotify_watchers), w);

no_insert:
  uv__handle_start(handle);
  QUEUE_INSERT_TAIL(&w->watchers, &handle->watchers);
  handle->path = w->path;
  handle->cb = cb;
  handle->wd = wd;

  return 0;
}