/**************************************** 
            API: create stream 
*****************************************/
static pj_status_t stream_stopBDIMAD(pjmedia_aud_stream *s)
{
    struct bd_stream *strm = (struct bd_stream*)s;
    pj_status_t status = PJ_SUCCESS;
    int i, err = 0;

    PJ_ASSERT_RETURN(strm != NULL, PJ_EINVAL);

    strm->go = 0;    
    
    for (i=0; !strm->rec_thread_exited && i<100; ++i)
	pj_thread_sleep(10);
    for (i=0; !strm->play_thread_exited && i<100; ++i)
	pj_thread_sleep(10);

    pj_thread_sleep(1);

    PJ_LOG(5,(THIS_FILE, "Stopping stream.."));

    strm->play_thread_initialized = 0;
    strm->rec_thread_initialized = 0;

    PJ_LOG(5,(THIS_FILE, "Done, status=%d", err));

    return status;
}

static pj_status_t stream_destroyBDIMAD(pjmedia_aud_stream *s)
{
    struct bd_stream *strm = (struct bd_stream*)s;
    int i = 0;

    PJ_ASSERT_RETURN(strm != NULL, PJ_EINVAL);

    stream_stopBDIMAD(s);

    // DeInit BDIMAD
    bdIMADpj_FreeAEC(&strm->bdIMADpjInstance); 
	PJ_LOG(4, (THIS_FILE, "Free AEC"));

    bdIMADpj_FreeStructures(&strm->bdIMADpjSettingsPtr, 
			    &strm->bdIMADpjWarningPtr);
    PJ_LOG(4, (THIS_FILE, "Free AEC Structure"));

    strm->bdIMADpjInstance = NULL;
    strm->bdIMADpjSettingsPtr = NULL;
    strm->bdIMADpjWarningPtr = NULL;    

    strm->quit_flag = 1;
    for (i=0; !strm->rec_thread_exited && i<100; ++i) {
	pj_thread_sleep(1);
    }
    for (i=0; !strm->play_thread_exited && i<100; ++i) {
	pj_thread_sleep(1);
    }

    PJ_LOG(5,(THIS_FILE, "Destroying stream.."));

    pj_pool_release(strm->pool);
    return PJ_SUCCESS;
}

/* API: stop stream. */
static pj_status_t strm_stop(pjmedia_aud_stream *s)
{
    struct pa_aud_stream *stream = (struct pa_aud_stream*)s;
    int i, err = 0;

    stream->quit_flag = 1;
    for (i=0; !stream->rec_thread_exited && i<100; ++i)
	pj_thread_sleep(10);
    for (i=0; !stream->play_thread_exited && i<100; ++i)
	pj_thread_sleep(10);

    pj_thread_sleep(1);

    PJ_LOG(5,(THIS_FILE, "Stopping stream.."));

    if (stream->play_strm)
	err = Pa_StopStream(stream->play_strm);

    if (stream->rec_strm && stream->rec_strm != stream->play_strm)
	err = Pa_StopStream(stream->rec_strm);

    stream->play_thread_initialized = 0;
    stream->rec_thread_initialized = 0;

    PJ_LOG(5,(THIS_FILE, "Done, status=%d", err));

    return err ? PJMEDIA_AUDIODEV_ERRNO_FROM_PORTAUDIO(err) : PJ_SUCCESS;
}

/* API: destroy stream. */
static pj_status_t strm_destroy(pjmedia_aud_stream *s)
{
    struct pa_aud_stream *stream = (struct pa_aud_stream*)s;
    int i, err = 0;

    stream->quit_flag = 1;
    for (i=0; !stream->rec_thread_exited && i<100; ++i) {
	pj_thread_sleep(1);
    }
    for (i=0; !stream->play_thread_exited && i<100; ++i) {
	pj_thread_sleep(1);
    }

    PJ_LOG(5,(THIS_FILE, "Closing %.*s: %lu underflow, %lu overflow",
			 (int)stream->name.slen,
			 stream->name.ptr,
			 stream->underflow, stream->overflow));

    if (stream->play_strm)
	err = Pa_CloseStream(stream->play_strm);

    if (stream->rec_strm && stream->rec_strm != stream->play_strm)
	err = Pa_CloseStream(stream->rec_strm);

    pj_pool_release(stream->pool);

    return err ? PJMEDIA_AUDIODEV_ERRNO_FROM_PORTAUDIO(err) : PJ_SUCCESS;
}

static int stun_destroy_test_session(struct stun_test_session *test_sess)
{

    unsigned i;
    pj_stun_sock_cb stun_cb;
    pj_status_t status;
    pj_stun_sock *stun_sock[MAX_SOCK_CLIENTS];

    pj_bzero(&stun_cb, sizeof(stun_cb));
    stun_cb.on_status = &stun_sock_on_status;

    pj_event_reset(test_sess->server_event);

    /* Create all clients first */
    for (i=0; i<MAX_SOCK_CLIENTS; ++i) {
	char name[10];
	sprintf(name, "stun%02d", i);
	status = pj_stun_sock_create(&test_sess->stun_cfg, name, pj_AF_INET(),
	                             &stun_cb, NULL, test_sess,
	                             &stun_sock[i]);
	if (status != PJ_SUCCESS) {
	    PJ_PERROR(1,(THIS_FILE, status, "Error creating stun socket"));
	    return -10;
	}
    }

    /* Start resolution */
    for (i=0; i<MAX_SOCK_CLIENTS; ++i) {
	pj_str_t server_ip = pj_str("127.0.0.1");
	status = pj_stun_sock_start(stun_sock[i], &server_ip,
	                            (pj_uint16_t)test_sess->server_port, NULL);
	if (status != PJ_SUCCESS) {
	    PJ_PERROR(1,(THIS_FILE, status, "Error starting stun socket"));
	    return -20;
	}
    }

    /* settle down */
    pj_thread_sleep(test_sess->param.client_sleep_after_start);

    /* Resume server threads */
    pj_event_set(test_sess->server_event);

    pj_thread_sleep(test_sess->param.client_sleep_before_destroy);

    /* Destroy clients */
    for (i=0; i<MAX_SOCK_CLIENTS; ++i) {
	status = pj_stun_sock_destroy(stun_sock[i]);
	if (status != PJ_SUCCESS) {
	    PJ_PERROR(1,(THIS_FILE, status, "Error destroying stun socket"));
	}
    }

    /* Give some time to ioqueue to free sockets */
    pj_thread_sleep(PJ_IOQUEUE_KEY_FREE_DELAY);

    return 0;
}

static void benchmark(void)
{
    FILETIME creation_time, exit_time;
    struct Times start, end;
    DWORD ts, te;
    LARGE_INTEGER elapsed;
    BOOL rc;
    int i;
    double pct;

    puts("Test started!"); fflush(stdout);

    ts = GetTickCount();
    rc = GetProcessTimes(GetCurrentProcess(), &creation_time, &exit_time,
			 &start.kernel_time, &start.user_time);
    for (i=DURATION; i>0; --i) {
	printf("\r%d ", i); fflush(stdout);
	pj_thread_sleep(1000);
    }
    rc = GetProcessTimes(GetCurrentProcess(), &creation_time, &exit_time,
			 &end.kernel_time, &end.user_time);
    te = GetTickCount();

    process(&start);
    process(&end);

    elapsed.QuadPart = end.u_total.QuadPart - start.u_total.QuadPart;

    pct = elapsed.QuadPart * 100.0 / ((te-ts)*10000.0);

    printf("CPU usage=%6.4f%%\n", pct); fflush(stdout);
}

static int simple_sleep_test(void)
{
    enum { COUNT = 10 };
    int i;
    pj_status_t rc;
    
    PJ_LOG(3,(THIS_FILE, "..will write messages every 1 second:"));
    
    for (i=0; i<COUNT; ++i) {
	pj_time_val tv;
	pj_parsed_time pt;

	rc = pj_thread_sleep(1000);
	if (rc != PJ_SUCCESS) {
	    app_perror("...error: pj_thread_sleep()", rc);
	    return -10;
	}

	rc = pj_gettimeofday(&tv);
	if (rc != PJ_SUCCESS) {
	    app_perror("...error: pj_gettimeofday()", rc);
	    return -11;
	}

	pj_time_decode(&tv, &pt);

	PJ_LOG(3,(THIS_FILE, 
		  "...%04d-%02d-%02d %02d:%02d:%02d.%03d",
		  pt.year, pt.mon, pt.day,
		  pt.hour, pt.min, pt.sec, pt.msec));

    }

    return 0;
}

/* Utility: display error message and exit application (usually
 * because of fatal error.
 */
static void err_exit(const char *title, pj_status_t status)
{
    if (status != PJ_SUCCESS) {
	icedemo_perror(title, status);
    }
    PJ_LOG(3,(THIS_FILE, "Shutting down.."));

    if (icedemo.icest)
	pj_ice_strans_destroy(icedemo.icest);
    
    pj_thread_sleep(500);

    icedemo.thread_quit_flag = PJ_TRUE;
    if (icedemo.thread) {
	pj_thread_join(icedemo.thread);
	pj_thread_destroy(icedemo.thread);
    }

    if (icedemo.ice_cfg.stun_cfg.ioqueue)
	pj_ioqueue_destroy(icedemo.ice_cfg.stun_cfg.ioqueue);

    if (icedemo.ice_cfg.stun_cfg.timer_heap)
	pj_timer_heap_destroy(icedemo.ice_cfg.stun_cfg.timer_heap);

    pj_caching_pool_destroy(&icedemo.cp);

    pj_shutdown();

    if (icedemo.log_fhnd) {
	fclose(icedemo.log_fhnd);
	icedemo.log_fhnd = NULL;
    }

    exit(status != PJ_SUCCESS);
}

static int timer_thread_run(void* p)
{
    pj_time_val tick = {0, 10};

    timer_running = 1;

    while (timer_running)
    {
        if (pj_timer_heap_count(timer) == 0)
        {
            pj_sem_wait(timer_sem);
        }
        else
        {
            pj_thread_sleep(PJ_TIME_VAL_MSEC(tick));
            pj_timer_heap_poll(timer, NULL);
        }
    }
    pj_timer_heap_destroy(timer);
    timer = NULL;
    pj_sem_destroy(timer_sem);
    timer_sem = NULL;
    pj_pool_release(timer_pool);
    timer_pool = NULL;
    timer_initialized = 0;
    return 0;
}

static void on_response(pj_http_req *hreq, const pj_http_resp *resp)
{
    pj_size_t i;

    PJ_UNUSED_ARG(hreq);
    PJ_UNUSED_ARG(resp);
    PJ_UNUSED_ARG(i);

#ifdef VERBOSE
    printf("%.*s, %d, %.*s\n", STR_PREC(resp->version),
           resp->status_code, STR_PREC(resp->reason));
    for (i = 0; i < resp->headers.count; i++) {
        printf("%.*s : %.*s\n", 
               STR_PREC(resp->headers.header[i].name),
               STR_PREC(resp->headers.header[i].value));
    }
#endif

    if (test_cancel) {
	/* Need to delay closing the client socket here, otherwise the
	 * server will get SIGPIPE when sending response.
	 */
	pj_thread_sleep(100);
        pj_http_req_cancel(hreq, PJ_TRUE);
        test_cancel = PJ_FALSE;
    }
}

static int timestamp_accuracy()
{
    pj_timestamp freq, t1, t2;
    pj_time_val tv1, tv2, tvtmp;
    pj_uint32_t msec, tics;
    pj_int64_t diff;

    PJ_LOG(3,(THIS_FILE, "...testing frequency accuracy (pls wait)"));

    pj_get_timestamp_freq(&freq);

    /* Get the start time */
    pj_gettimeofday(&tvtmp);
    do {
	pj_gettimeofday(&tv1);
    } while (PJ_TIME_VAL_EQ(tvtmp, tv1));
    pj_get_timestamp(&t1);

    /* Sleep for 5 seconds */
    pj_thread_sleep(10000);

    /* Get end time */
    pj_gettimeofday(&tvtmp);
    do {
	pj_gettimeofday(&tv2);
    } while (PJ_TIME_VAL_EQ(tvtmp, tv2));
    pj_get_timestamp(&t2);

    /* Get the elapsed time */
    PJ_TIME_VAL_SUB(tv2, tv1);
    msec = PJ_TIME_VAL_MSEC(tv2);

    /* Check that the frequency match the elapsed time */
    tics = (unsigned)(t2.u64 - t1.u64);
    diff = tics - (msec * freq.u64 / 1000);
    if (diff < 0)
	diff = -diff;

    /* Only allow 1 msec mismatch */
    if (diff > (pj_int64_t)(freq.u64 / 1000)) {
	PJ_LOG(3,(THIS_FILE, "....error: timestamp drifted by %d usec after "
			     "%d msec", 
			     (pj_uint32_t)(diff * 1000000 / freq.u64), 
			     msec));
	return -2000;

    /* Otherwise just print warning if timestamp drifted by >1 usec */
    } else if (diff > (pj_int64_t)(freq.u64 / 1000000)) {
	PJ_LOG(3,(THIS_FILE, "....warning: timestamp drifted by %d usec after "
			     "%d msec", 
			     (pj_uint32_t)(diff * 1000000 / freq.u64), 
			     msec));
    } else {
	PJ_LOG(3,(THIS_FILE, "....good. Timestamp is accurate down to"
			     " nearest usec."));
    }

    return 0;
}

int echo_srv_common_loop(pj_atomic_t *bytes_counter)
{
    pj_highprec_t last_received, avg_bw, highest_bw;
    pj_time_val last_print;
    unsigned count;
    const char *ioqueue_name;

    last_received = 0;
    pj_gettimeofday(&last_print);
    avg_bw = highest_bw = 0;
    count = 0;

    ioqueue_name = pj_ioqueue_name();

    for (;;) {
        pj_highprec_t received, cur_received, bw;
        unsigned msec;
        pj_time_val now, duration;

        pj_thread_sleep(1000);

        received = cur_received = pj_atomic_get(bytes_counter);
        cur_received = cur_received - last_received;

        pj_gettimeofday(&now);
        duration = now;
        PJ_TIME_VAL_SUB(duration, last_print);
        msec = PJ_TIME_VAL_MSEC(duration);
        
        bw = cur_received;
        pj_highprec_mul(bw, 1000);
        pj_highprec_div(bw, msec);

        last_print = now;
        last_received = received;

        avg_bw = avg_bw + bw;
        count++;

        PJ_LOG(3,("", "%s UDP (%d threads): %u KB/s (avg=%u KB/s) %s", 
		  ioqueue_name,
                  ECHO_SERVER_MAX_THREADS, 
                  (unsigned)(bw / 1000),
                  (unsigned)(avg_bw / count / 1000),
                  (count==20 ? "<ses avg>" : "")));

        if (count==20) {
            if (avg_bw/count > highest_bw)
                highest_bw = avg_bw/count;

            count = 0;
            avg_bw = 0;

            PJ_LOG(3,("", "Highest average bandwidth=%u KB/s",
                          (unsigned)(highest_bw/1000)));
        }
    }
    PJ_UNREACHED(return 0;)
}

/*
 * Clock thread
 */
static int clock_thread(void *arg)
{
    pj_timestamp now;
    pjmedia_clock *clock = (pjmedia_clock*) arg;

    /* Set thread priority to maximum unless not wanted. */
    if ((clock->options & PJMEDIA_CLOCK_NO_HIGHEST_PRIO) == 0) {
	int max = pj_thread_get_prio_max(pj_thread_this());
	if (max > 0)
	    pj_thread_set_prio(pj_thread_this(), max);
    }

    //printf("%s:------------11--------------\n", THIS_FILE);
    /* Get the first tick */
    pj_get_timestamp(&clock->next_tick);
    clock->next_tick.u64 += clock->interval.u64;


    while (!clock->quitting) {

	pj_get_timestamp(&now);

	/* Wait for the next tick to happen */
	if (now.u64 < clock->next_tick.u64) {
	    unsigned msec;
	    msec = pj_elapsed_msec(&now, &clock->next_tick);
	    pj_thread_sleep(msec);
	}
    //printf("%s:------------12--------------, 0x%02x, %d\n", THIS_FILE, clock, (int)(clock->quitting));
	/* Skip if not running */
	if (!clock->running) {
	    /* Calculate next tick */
	    clock_calc_next_tick(clock, &now);
	    continue;
	}

	pj_lock_acquire(clock->lock);
    //printf("%s:------------13--------------, 0x%02x, %d\n", THIS_FILE, clock, (int)(clock->quitting));
	/* Call callback, if any */
	if (clock->cb)
	    (*clock->cb)(&clock->timestamp, clock->user_data);

	/* Best effort way to detect if we've been destroyed in the callback */
	if (clock->quitting)
	    break;

	/* Increment timestamp */
	clock->timestamp.u64 += clock->timestamp_inc;
    //printf("%s:------------14--------------, 0x%02x, %d\n", THIS_FILE, clock, (int)(clock->quitting));
	/* Calculate next tick */
	clock_calc_next_tick(clock, &now);
    //printf("%s:------------15--------------\n", THIS_FILE);
	pj_lock_release(clock->lock);
    }

    return 0;
}

/*
 * Poll the I/O Queue for completed events.
 */
PJ_DEF(int) pj_ioqueue_poll( pj_ioqueue_t *ioq,
			     const pj_time_val *timeout)
{
    /* Polling is not necessary on UWP, since each socket handles
     * its events already.
     */
    PJ_UNUSED_ARG(ioq);

    pj_thread_sleep(PJ_TIME_VAL_MSEC(*timeout));

    return 0;
}

static pj_status_t wait_play(pjmedia_frame *f)
{
    play_frm_copied = PJ_FALSE;
    play_frm = *f;
    play_frm_ready = PJ_TRUE;
    while (!play_frm_copied) {
	pj_thread_sleep(1);
    }
    play_frm_ready = PJ_FALSE;

    return PJ_SUCCESS;
}

int do_test(void *user) {
    int i;
    pj_pool_t *pool = (pj_pool_t *)user;
    opool_init(&opool, 10, strlen(s)+10, pool);
    for (i = 0; i < 100; i++) {
        opool_item_t *p_item = opool_get(&opool);
        pj_memcpy(p_item->data, s, strlen(s));
        printf("%d: %s\n", i, p_item->data);
        pj_thread_sleep(pj_rand()%100);
        opool_free(&opool, p_item);
    }
    return 0;
}

/* 
 * Handle timer and network events 
 */
static void srv_handle_events(pj_turn_srv *srv, const pj_time_val *max_timeout)
{
    /* timeout is 'out' var. This just to make compiler happy. */
    pj_time_val timeout = { 0, 0};
    unsigned net_event_count = 0;
    int c;

    /* Poll the timer. The timer heap has its own mutex for better 
     * granularity, so we don't need to lock the server. 
     */
    timeout.sec = timeout.msec = 0;
    c = pj_timer_heap_poll( srv->core.timer_heap, &timeout );

    /* timer_heap_poll should never ever returns negative value, or otherwise
     * ioqueue_poll() will block forever!
     */
    pj_assert(timeout.sec >= 0 && timeout.msec >= 0);
    if (timeout.msec >= 1000) timeout.msec = 999;

    /* If caller specifies maximum time to wait, then compare the value with
     * the timeout to wait from timer, and use the minimum value.
     */
    if (max_timeout && PJ_TIME_VAL_GT(timeout, *max_timeout)) {
	timeout = *max_timeout;
    }

    /* Poll ioqueue. 
     * Repeat polling the ioqueue while we have immediate events, because
     * timer heap may process more than one events, so if we only process
     * one network events at a time (such as when IOCP backend is used),
     * the ioqueue may have trouble keeping up with the request rate.
     *
     * For example, for each send() request, one network event will be
     *   reported by ioqueue for the send() completion. If we don't poll
     *   the ioqueue often enough, the send() completion will not be
     *   reported in timely manner.
     */
    do {
	c = pj_ioqueue_poll( srv->core.ioqueue, &timeout);
	if (c < 0) {
	    pj_thread_sleep(PJ_TIME_VAL_MSEC(timeout));
	    return;
	} else if (c == 0) {
	    break;
	} else {
	    net_event_count += c;
	    timeout.sec = timeout.msec = 0;
	}
    } while (c > 0 && net_event_count < MAX_NET_EVENTS);

}

void queue_enqueue(queue_t *queue, void *value)
{
    pthread_mutex_lock(&(queue->mutex));
    while (queue->size == queue->capacity) {
        pj_thread_sleep(2);
        pthread_cond_wait(&(queue->cond_full), &(queue->mutex));
    }
    queue->buffer[queue->in] = value;
    ++ queue->size;
    ++ queue->in;
    queue->in %= queue->capacity;
    pthread_mutex_unlock(&(queue->mutex));
    pthread_cond_broadcast(&(queue->cond_empty));
}

/*
 * Stop stream.
 */
PJ_DEF(pj_status_t) pjmedia_snd_stream_stop(pjmedia_snd_stream *stream)
{
    int i, err = 0;

    stream->quit_flag = 1;
    for (i=0; !stream->rec_thread_exited && i<100; ++i)
	pj_thread_sleep(10);
    for (i=0; !stream->play_thread_exited && i<100; ++i)
	pj_thread_sleep(10);

    pj_thread_sleep(1);

    PJ_LOG(5,(THIS_FILE, "Stopping stream.."));

    if (stream->play_strm)
	err = Pa_StopStream(stream->play_strm);

    if (stream->rec_strm && stream->rec_strm != stream->play_strm)
	err = Pa_StopStream(stream->rec_strm);

    PJ_LOG(5,(THIS_FILE, "Done, status=%d", err));

    return err ? PJMEDIA_ERRNO_FROM_PORTAUDIO(err) : PJ_SUCCESS;
}

void *queue_dequeue(queue_t *queue)
{
    pthread_mutex_lock(&(queue->mutex));
    while (queue->size == 0) {
        pj_thread_sleep(2);
        pthread_cond_wait(&(queue->cond_empty), &(queue->mutex));
    }
    void *value = queue->buffer[queue->out];
    //printf("dequeue %d\n", *(int *)value);
    -- queue->size;
    ++ queue->out;
    queue->out %= queue->capacity;
    pthread_mutex_unlock(&(queue->mutex));
    pthread_cond_broadcast(&(queue->cond_full));
    return value;
}