/* non-blocking, returns socket descriptor or -1 if none available
 * NOTE:  This function assumes that the cursor of all buf's is zero,
 *        and it can call buf_reset(source, tag).
 */
int MPINU_rank_of_msg_avail_in_cache(int source, int tag) {
    if (source != MPI_ANY_SOURCE) {
      if (tag == MPI_ANY_TAG) {
	if (buf_avail(source, sizeof(struct msg_hdr)))
          return source;
      }
      else
	while (buf_avail(source, sizeof(struct msg_hdr))) {
	  struct msg_hdr *hdr = buf_peek(source, sizeof(struct msg_hdr));
	  if ( ntohl( hdr->tag ) == tag ) {
	    buf_reset(source, tag); /* clean up for next caller */
	    return source;
	  } else { /* Else skip header and body */
	    buf_skip( source, sizeof(struct msg_hdr) + ntohl(hdr->size) );
	  }
	}
      buf_reset(source, tag); /* clean up for next caller */
    }
    else /* else source == MPI_ANY_SOURCE */
      for (source = 0; source < MPINU_num_slaves; source++) {
	int rank = MPINU_rank_of_msg_avail_in_cache(source, tag);
	if (rank != -1)
	  return rank;
      }
    /* if no source had buf_avail, then no msg avail */
    return -1;
}

END_TEST

START_TEST(test_integer)
{
#define OVERSIZE ":19223372036854775807\r\n"
#define INVALID1 ":123lOl456\r\n"
#define INVALID2 ":\r\n"

    struct element el_c, el_p;
    int ret;

    struct int_pair {
        char *serialized;
        uint64_t num;
    } pairs[3] = {
        {":-1\r\n", -1},
        {":9223372036854775807\r\n", 9223372036854775807},
        {":128\r\n", 128}
    };


    test_reset();
    for (int i = 0; i < 3; i++) {
        size_t len = strlen(pairs[i].serialized);

        buf_reset(buf);
        el_c.type = ELEM_INT;
        el_c.num = pairs[i].num;
        ret = compose_element(&buf, &el_c);
        ck_assert(ret == len);
        ck_assert_int_eq(cc_bcmp(buf->rpos, pairs[i].serialized, len), 0);

        el_p.type = ELEM_UNKNOWN;
        ret = parse_element(&el_p, buf);
        ck_assert_int_eq(ret, PARSE_OK);
        ck_assert(buf->rpos == buf->wpos);
        ck_assert(el_p.type == ELEM_INT);
        ck_assert(el_p.num == pairs[i].num);
    }

    buf_reset(buf);
    buf_write(buf, OVERSIZE, sizeof(OVERSIZE) - 1);
    ret = parse_element(&el_p, buf);
    ck_assert_int_eq(ret, PARSE_EOVERSIZE);

    buf_reset(buf);
    buf_write(buf, INVALID1, sizeof(INVALID1) - 1);
    ret = parse_element(&el_p, buf);
    ck_assert_int_eq(ret, PARSE_EINVALID);

    buf_reset(buf);
    buf_write(buf, INVALID2, sizeof(INVALID2) - 1);
    ret = parse_element(&el_p, buf);
    ck_assert_int_eq(ret, PARSE_EINVALID);

#undef INVALID2
#undef INVALID1
#undef OVERSIZE
}

void gaelco_serial_device::device_stop()
{
	{
		std::lock_guard<std::mutex> guard(m_mutex);
		buf_reset(m_out_ptr);
		buf_reset(m_in_ptr);
	}
	osd_sharedmem_free(m_os_shmem);
}

void gaelco_serial_device::device_stop()
{
	shmem_lock(m_shmem);
	buf_reset(m_out_ptr);
	buf_reset(m_in_ptr);
	shmem_unlock(m_shmem);

	osd_sharedmem_free(m_os_shmem);
}

void _soReserve(struct so_t *so) {	//locked outside
	if (so->mFd)
		close(so->mFd);
	so->mFd = 0;
	so->mState = 0;
	so->mEnv = 0;
	buf_reset(so->mRBuf);
	buf_reset(so->mWBuf);
}

static DEVICE_STOP( gaelco_serial )
{
	gaelco_serial_state *state = get_token(device);

	shmem_lock(state->m_shmem);
	buf_reset(state->m_out_ptr);
	buf_reset(state->m_in_ptr);
	shmem_unlock(state->m_shmem);

	osd_sharedmem_free(state->m_os_shmem);
}

static DEVICE_RESET( gaelco_serial )
{
	gaelco_serial_state *state = get_token(device);

	state->m_status = GAELCOSER_STATUS_READY	|GAELCOSER_STATUS_IRQ_ENABLE ;

	state->m_last_in_msg_cnt = -1;
	state->m_slack_cnt = LINK_SLACK_B;

	shmem_lock(state->m_shmem);
	buf_reset(state->m_out_ptr);
	buf_reset(state->m_in_ptr);
	shmem_unlock(state->m_shmem);
}

static int my_ping_resp_cb(int fd, void *arg)
{
    int res = 0, done;
    my_conn_t *my;
    buf_t *buf;

    my = (my_conn_t *)arg;
    buf = &(my->buf);

    if( (res = my_real_read(fd, buf, &done)) < 0 ){
        log_err(g_log, "my_real_read error\n");
        goto end;
    }

    if(done){
        if( (res = del_handler(fd)) < 0 ){
            log(g_log, "del_handler fd[%d] error\n", fd);
            goto end;
        }

        buf_reset(buf);

        my_conn_put(my, 0 );//ping不更新mysql的使用时间，否则不好玩了，永远释放不了
    }

    return res;

end:
    my_conn_close(my);

    return res;
}

static void encode(const char *ps, int len, struct buf *buf)
{
    const unsigned char *p = (const unsigned char *)ps;

    buf_reset(buf);
    /* allocate enough space plus a little slop to cover
     * escaping a few characters */
    buf_ensure(buf, len+10);

    for ( ; len > 0 ; len--, p++) {
        switch (*p) {
        case '\0':
        case '\t':
        case '\r':
        case '\n':
            buf_putc(buf, ESCAPE);
            buf_putc(buf, 0x80|(*p));
            break;
        case ESCAPE:
            buf_putc(buf, ESCAPE);
            buf_putc(buf, ESCAPE);
            break;
        default:
            buf_putc(buf, *p);
            break;
        }
    }

    /* ensure the buf is NUL-terminated; we pass the buf's data to
     * bsearch_mem(), which expects a C string, several times */
    buf_cstring(buf);
}

static void print_horizontal_tabbar(struct window *win)
{
	int i;

	buf_reset(win->x, win->w, 0);
	buf_move_cursor(win->x, win->y);

	calculate_tabbar(win);
	for (i = win->first_tab_idx; i < win->views.count; i++) {
		struct view *v = win->views.ptrs[i];

		if (obuf.x + v->tt_truncated_width > win->w)
			break;
		print_horizontal_tab_title(v, i);
	}
	set_builtin_color(BC_TABBAR);
	if (i != win->views.count) {
		while (obuf.x < obuf.width - 1)
			buf_put_char(' ');
		if (obuf.x == obuf.width - 1)
			buf_put_char('>');
	} else {
		buf_clear_eol();
	}
}

int hfp_info_start(int fd, unsigned int mtu, struct hfp_info *info)
{
	info->fd = fd;
	info->mtu = mtu;

	/* Initialize to MTU, it may change when actually read the socket. */
	info->packet_size = mtu;
	buf_reset(info->playback_buf);
	buf_reset(info->capture_buf);

	audio_thread_add_callback(info->fd, hfp_info_callback, info);

	info->started = 1;

	return 0;
}

static void decode(const char *ps, int len, struct buf *buf)
{
    const unsigned char *p = (const unsigned char *)ps;

    buf_reset(buf);
    /* allocate enough space; we don't need slop because
     * decoding can only shrink the result */
    buf_ensure(buf, len);

    for ( ; len > 0 ; len--, p++) {
        if (*p == ESCAPE) {
            if (len < 2) {
                /* invalid encoding, silently ignore */
                continue;
            }
            len--;
            p++;
            if (*p == ESCAPE)
                buf_putc(buf, ESCAPE);
            else
                buf_putc(buf, (*p) & ~0x80);
        }
        else
            buf_putc(buf, *p);
    }
    /* Note: buf is not NUL-terminated.  It happens that neither
     * skiplist nor berkeley backends guarantee any such thing,
     * and so code that depends on it is quite broken anyway */
}

static int zlib_decompress(struct transaction_t *txn,
                           const char *buf, unsigned len)
{
    struct ws_context *ctx = (struct ws_context *) txn->ws_ctx;
    z_stream *zstrm = ctx->pmce.deflate.zstrm;

    zstrm->next_in = (Bytef *) buf;
    zstrm->avail_in = len;

    buf_reset(&txn->zbuf);

    do {
        int zr;

        buf_ensure(&txn->zbuf, 4096);

        zstrm->next_out = (Bytef *) txn->zbuf.s + txn->zbuf.len;
        zstrm->avail_out = txn->zbuf.alloc - txn->zbuf.len;

        zr = inflate(zstrm, Z_SYNC_FLUSH);
        if (!(zr == Z_OK || zr == Z_STREAM_END || zr == Z_BUF_ERROR)) {
            /* something went wrong */
            syslog(LOG_ERR, "zlib deflate error: %d %s", zr, zstrm->msg);
            return -1;
        }

        txn->zbuf.len = txn->zbuf.alloc - zstrm->avail_out;

    } while (!zstrm->avail_out);

    return 0;
}

static void
test_reset(void)
{
    request_reset(req);
    response_reset(rsp);
    buf_reset(buf);
}

static void
flacdec_read_vorbis_comment(BitstreamReader *comment,
                            unsigned channel_count,
                            int *channel_mask)
{
    struct bs_buffer *line = buf_new();
    unsigned line_len;
    unsigned total_lines;
    const char mask_prefix[] =
        "WAVEFORMATEXTENSIBLE_CHANNEL_MASK=";

    if (!setjmp(*br_try(comment))) {
        /*skip over vendor string*/
        line_len = comment->read(comment, 32);
        comment->skip_bytes(comment, line_len);

        /*walk through all entries in the comment*/
        for (total_lines = comment->read(comment, 32);
             total_lines > 0;
             total_lines--) {
            const char *s;

            /*populate entry one character at a time
              (this avoids allocating a big chunk of space
               if the length field is something way too large)*/
            buf_reset(line);

            for (line_len = comment->read(comment, 32);
                 line_len > 0;
                 line_len--) {
                buf_putc(
                    toupper((int)comment->read(comment, 8)),
                    line);
            }
            buf_putc(0, line);  /*NULL terminator*/

            s = (const char *)buf_window_start(line);

            /*if line starts with mask prefix*/
            if (strstr(s, mask_prefix) == s) {
                /*convert rest of line to base-16 integer*/
                unsigned mask = (unsigned)strtoul(
                    s + strlen(mask_prefix), NULL, 16);
                /*and populate mask field if its number of channel bits
                  matches the stream's channel count*/
                if (channel_bits(mask) == channel_count) {
                    *channel_mask = mask;
                }
            }
        }
        br_etry(comment);
    } else {
        /*read error in VORBIS_COMMENT
          (probably invalid length field somewhere)*/
        br_etry(comment);
    }

    buf_close(line);
}

void end_xcal(struct buf *buf)
{
    /* End xCal stream */
    buf_reset(buf);
    buf_printf_markup(buf, 2, "</components>");
    buf_printf_markup(buf, 1, "</vcalendar>");
    buf_printf_markup(buf, 0, "</icalendar>");
}

int cli_hs_stage3_cb(int fd, void *arg)
{
    int done, res = 0;
    cli_conn_t *cli;
    conn_t *c;
    buf_t *buf;

    cli = (cli_conn_t *)arg;
    buf = &(cli->buf);
    c = cli->conn;


    if( (res = my_real_write(fd, buf, &done)) < 0 ){
        log_err(g_log, "conn:%u my_real_write error\n", c->connid);
        goto end;
    }

    if(done){
        if( (res = del_handler(fd)) < 0 ){
            log(g_log, "conn:%u del_handler error\n", c->connid);
            goto end;
        }

        res = add_handler(fd, EPOLLIN, cli_query_cb, arg);
        if(res < 0){
            log(g_log, "conn:%u add_handler error\n", c->connid);
            goto end;
        }

        buf_reset(buf);
        buf_reset(&(c->buf));

        conn_state_set_idle(c);
    }

    return res;

end:
    //fix me
    conn_close(c);

    return res;
}

int hfp_info_add_iodev(struct hfp_info *info, struct cras_iodev *dev)
{
	if (dev->direction == CRAS_STREAM_OUTPUT) {
		if (info->odev)
			goto invalid;
		info->odev = dev;

		buf_reset(info->playback_buf);
	} else if (dev->direction == CRAS_STREAM_INPUT) {
		if (info->idev)
			goto invalid;
		info->idev = dev;

		buf_reset(info->capture_buf);
	}

	return 0;

invalid:
	return -EINVAL;
}

void vs_start(void)
{
  DEBUGOUT("VS: start\n");

  vs_playing = 1;

  vs_pause();
  vs_setvolume(vs_vol);

  buf_reset();

  return;
}

static int conn_release(conn_t *c)
{
    if(c == NULL){
        log(g_log, "conn_release error\n");
        return -1;
    }

    c->my = NULL;
    c->cli = NULL;
    buf_reset(&(c->buf));

    return genpool_release_page(conn_pool, c);
}