myvtm/Other/libaudioframework/audiomicspkpulse.c

#include "audiomicspkpulse.h"
#include "audiocontext.h"
#include "audiolog.h"
#include "./other/delaybuf.h"

#include <assert.h>
#include <time.h>
#include <semaphore.h>
#include <pthread.h>
#include <unistd.h>


#define MAX_DELAY			60
#define CLOCK_PERIOD		10
#define AUDIO_CLOCK			8000
#define CAPTURE_AUDIO_CLOCK	8000

/*Audio stream flag*/
#define AUDIO_STRM_ON       1
#define AUDIO_STRM_OFF      0

#ifndef RVC_MAX_AUDIO_BUFFER_LEN
#define RVC_MAX_AUDIO_BUFFER_LEN 1024
#endif

#ifndef RVC_DELAY_AUDIO_LEN
#define RVC_DELAY_AUDIO_LEN 160
#endif

#ifndef RVC_PA_ADJUST_LATENCY_PROTOCOL_VERSION
#define RVC_PA_ADJUST_LATENCY_PROTOCOL_VERSION 13
#endif

static uint32_t latency_ms = 10; // requested initial latency in milisec: 0 use max
static pa_usec_t latency = 0; 	 //real latency in usec (for timestamping)

static pa_usec_t play_latency = 0; 	 //real latency in usec (for timestamping)

//pa_stream* recordstream;	  /* pulse audio stream*/
//pa_context* pa_ctx;			  /* pulse context*/

//pa_stream* playstream;		/* pulse audio stream*/
//pa_context* play_pa_ctx;			  /* pulse context*/

static apr_status_t read_frame(void* self, audioframe_t* frame)
{
	audiomicspkpulse_t* micspk = CONTAINING_RECORD(self, audiomicspkpulse_t, base);

	frame->size = 2 * micspk->capture_frame_samples;
	frame->dtmf = 0;

	delay_buf_get((delay_buf*)micspk->rec_dbuf, (short*)frame->buffer);

	return APR_SUCCESS;
}

static apr_status_t write_frame(void* self, const audioframe_t* frame)
{
	audiomicspkpulse_t* micspk = CONTAINING_RECORD(self, audiomicspkpulse_t, base);

	assert(micspk->play_frame_samples * 2 == frame->size);

	delay_buf_put((delay_buf*)micspk->ply_dbuf, (short*)frame->buffer);

	return APR_SUCCESS;
}

static audiostream_vtbl_t g_stream_vtbl = {
	&read_frame,
	&write_frame,
};

static int get_device_id(audio_context_t* audio_ctx, int indev, const char* key)
{
	assert(NULL != audio_ctx);
	assert(NULL != key);

	int iret = -1;
	int index = 0;
	int device_count = 1;

	audio_device_t* audio_device = NULL;
	
	if (1 == indev){
		device_count = audio_ctx->num_input_dev;
	}
	else{
		device_count = audio_ctx->num_output_dev;
	}

	for (index = 0; index < device_count; ++index) {
		if (indev) {
			audio_device = &audio_ctx->list_input_devices[index];
		}
		else {
			audio_device = &audio_ctx->list_output_devices[index];
		}
		//audio_log_v(AUDIO_LOG_LEVEL_INFO, "%s:%d audio_device(%d) description is %s.", __FUNCTION__, __LINE__, indev, audio_device->description);
		if (audio_device->description && strstr(audio_device->description, key)) {
			iret = index;
		}
	}
	return iret;
}


static void pa_state_cb(pa_context* c, void* data)
{
	pa_context_state_t state;
	int* pa_ready = (int*)data;
	state = pa_context_get_state(c);
	switch (state)
	{
		// These are just here for reference
	case PA_CONTEXT_UNCONNECTED:
		audio_log_v(AUDIO_LOG_LEVEL_INFO, "unconnected");
		break;
	case PA_CONTEXT_CONNECTING:
	case PA_CONTEXT_AUTHORIZING:
	case PA_CONTEXT_SETTING_NAME:
	default:
		audio_log_v(AUDIO_LOG_LEVEL_INFO, "no state");
		break;
	case PA_CONTEXT_FAILED:
	case PA_CONTEXT_TERMINATED:
		*pa_ready = 2;
		audio_log_v(AUDIO_LOG_LEVEL_INFO, "failed");
		break;
	case PA_CONTEXT_READY:
		*pa_ready = 1;
		audio_log_v(AUDIO_LOG_LEVEL_INFO, "ready");
		break;
	}
}


static void pa_sinklist_cb(pa_context* c, const pa_sink_info* l, int eol, void* userdata)
{
	audio_context_t* audio_ctx = (audio_context_t*)userdata;

	/*
	 * If eol is set to a positive number,
	 * you're at the end of the list
	 */
	if (eol > 0){
		return;
	}

	double flatency = 0.0;

	if (flatency <= 0.0)
		flatency = (double)latency_ms / 1000;

	audio_ctx->num_output_dev++;
	/*add device to list*/
	audio_ctx->list_output_devices = realloc(audio_ctx->list_output_devices, audio_ctx->num_output_dev * sizeof(audio_device_t));
	if (audio_ctx->list_output_devices == NULL){
		audio_log_v(AUDIO_LOG_LEVEL_INFO, "memory allocation failure (pa_sinklist_cb): %s", strerror(errno));
		exit(-1);
	}
	/*fill device data*/
	audio_ctx->list_output_devices[audio_ctx->num_output_dev - 1].id = l->index; /*saves dev id*/
	strncpy(audio_ctx->list_output_devices[audio_ctx->num_output_dev - 1].name, l->name, MAX_PATH_EX-1);
	strncpy(audio_ctx->list_output_devices[audio_ctx->num_output_dev - 1].description, l->description, MAX_PATH-1);
	audio_ctx->list_output_devices[audio_ctx->num_output_dev - 1].channels = l->channel_map.channels;
	audio_ctx->list_output_devices[audio_ctx->num_output_dev - 1].samprate = l->sample_spec.rate;
	audio_ctx->list_output_devices[audio_ctx->num_output_dev - 1].low_latency = flatency; /*in seconds*/
	audio_ctx->list_output_devices[audio_ctx->num_output_dev - 1].high_latency = flatency; /*in seconds*/
}

static void pa_sourcelist_cb(pa_context* c, const pa_source_info* l, int eol, void* data)
{
	audio_context_t* audio_ctx = (audio_context_t*)data;

	int channels = 1;
	/*
	* If eol is set to a positive number,
	* you're at the end of the list
	*/
	if (eol > 0) {
		return;
	}
		
	if (l->sample_spec.channels < 1)
	{
		channels = 1;
	}
	else {
		channels = l->sample_spec.channels;
	}
		
	double ilatency = 0.0;

	if (ilatency <= 0.0)
		ilatency = (double)latency_ms / 1000;

	if (l->monitor_of_sink == PA_INVALID_INDEX)
	{
		audio_ctx->num_input_dev++;
		/*add device to list*/
		audio_ctx->list_input_devices = realloc(audio_ctx->list_input_devices, audio_ctx->num_input_dev * sizeof(audio_device_t));
		if (audio_ctx->list_input_devices == NULL){
			audio_log_v(AUDIO_LOG_LEVEL_INFO, "memory allocation failure (pa_sourcelist_cb): %s", strerror(errno));
			exit(-1);
		}
		/*fill device data*/
		audio_ctx->list_input_devices[audio_ctx->num_input_dev - 1].id = l->index; /*saves dev id*/
		strncpy(audio_ctx->list_input_devices[audio_ctx->num_input_dev - 1].name, l->name, MAX_PATH_EX-1);
		strncpy(audio_ctx->list_input_devices[audio_ctx->num_input_dev - 1].description, l->description, MAX_PATH-1);
		audio_ctx->list_input_devices[audio_ctx->num_input_dev - 1].channels = channels;
		audio_ctx->list_input_devices[audio_ctx->num_input_dev - 1].samprate = l->sample_spec.rate;
		audio_ctx->list_input_devices[audio_ctx->num_input_dev - 1].low_latency = ilatency; /*in seconds*/
		audio_ctx->list_input_devices[audio_ctx->num_input_dev - 1].high_latency = ilatency; /*in seconds*/
	}
}


void finish(pa_context* rvc_pa_ctx, pa_mainloop* pa_ml)
{
	/* clean up and disconnect */
	pa_context_disconnect(rvc_pa_ctx);
	pa_context_unref(rvc_pa_ctx);
	pa_mainloop_free(pa_ml);
}


int pa_get_devicelist(audio_context_t* audio_ctx)
{
	/*assertions*/
	assert(audio_ctx != NULL);

	/* Define our pulse audio loop and connection variables */
	pa_mainloop* pa_ml;
	pa_mainloop_api* pa_mlapi;
	pa_operation* pa_op = NULL;
	pa_context* pa_ctx;

	/* We'll need these state variables to keep track of our requests */
	int state = 0;
	int pa_ready = 0;

	/* Create a mainloop API and connection to the default server */
	pa_ml = pa_mainloop_new();
	pa_mlapi = pa_mainloop_get_api(pa_ml);
	pa_ctx = pa_context_new(pa_mlapi, "getDevices");

	/* This function connects to the pulse server */
	if (pa_context_connect(pa_ctx, NULL, PA_CONTEXT_NOFLAGS, NULL) < 0)
	{
		audio_log_v(AUDIO_LOG_LEVEL_INFO, "unable to connect to server: pa_context_connect failed");
		finish(pa_ctx, pa_ml);
		return -1;
	}

	/*
	 * This function defines a callback so the server will tell us
	 * it's state.
	 * Our callback will wait for the state to be ready.
	 * The callback will modify the variable to 1 so we know when we
	 * have a connection and it's ready.
	 * If there's an error, the callback will set pa_ready to 2
	 */
	pa_context_set_state_callback(pa_ctx, pa_state_cb, &pa_ready);

	/*
	 * Now we'll enter into an infinite loop until we get the data
	 * we receive or if there's an error
	 */
	for (;;)
	{
		/*
		 * We can't do anything until PA is ready,
		 * so just iterate the mainloop and continue
		 */
		if (pa_ready == 0)
		{
			pa_mainloop_iterate(pa_ml, 1, NULL);
			continue;
		}
		/* We couldn't get a connection to the server, so exit out */
		if (pa_ready == 2)
		{
			finish(pa_ctx, pa_ml);
			return -1;
		}
		/*
		 * At this point, we're connected to the server and ready
		 * to make requests
		 */
		switch (state)
		{
			/* State 0: we haven't done anything yet */
		case 0:
			/*
			 * This sends an operation to the server.
			 * pa_sinklist_cb is our callback function and a pointer
			 * o our devicelist will be passed to the callback
			 * (audio_ctx) The operation ID is stored in the
			 * pa_op variable
			 */
			pa_op = pa_context_get_sink_info_list(
				pa_ctx,
				pa_sinklist_cb,
				(void*)audio_ctx);

			/* Update state for next iteration through the loop */
			state++;
			break;
		case 1:
			/*
			 * Now we wait for our operation to complete.
			 * When it's complete our pa_output_devicelist is
			 * filled out, and we move along to the next state
			 */
			if (pa_operation_get_state(pa_op) == PA_OPERATION_DONE)
			{
				pa_operation_unref(pa_op);

				/*
				 * Now we perform another operation to get the
				 * source(input device) list just like before.
				 * This time we pass a pointer to our input structure
				 */
				pa_op = pa_context_get_source_info_list(
					pa_ctx,
					pa_sourcelist_cb,
					(void*)audio_ctx);
				/* Update the state so we know what to do next */
				state++;
			}
			break;
		case 2:
			if (pa_operation_get_state(pa_op) == PA_OPERATION_DONE)
			{
				/*
				 * Now we're done,
				 * clean up and disconnect and return
				 */
				pa_operation_unref(pa_op);
				finish(pa_ctx, pa_ml);
				return 0;
			}
			break;
		default:
			/* We should never see this state */
			audio_log_v(AUDIO_LOG_LEVEL_INFO, " pulse audio in state %d", state);
			return -1;
		}
		/*
		 * Iterate the main loop and go again.  The second argument is whether
		 * or not the iteration should block until something is ready to be
		 * done.  Set it to zero for non-blocking.
		 */
		pa_mainloop_iterate(pa_ml, 1, NULL);
	}

	return 0;
}

int audio_init_pulseaudio(audio_context_t* audio_ctx)
{
	/*assertions*/
	assert(NULL != audio_ctx);
	if (pa_get_devicelist(audio_ctx) < 0){
		audio_log_v(AUDIO_LOG_LEVEL_INFO, "pulse audio failed to get audio device list from pulse server.");
		return -1;
	}

	return 0;
}

apr_status_t audio_context_create(apr_pool_t* pool, audio_context_t** audio_ctx)
{
	audio_context_t* actx = (audio_context_t*)apr_pcalloc(pool, sizeof(audio_context_t));

	if (NULL == actx) {
		audio_log_v(AUDIO_LOG_LEVEL_INFO, "%s:%d couldn't apr_pcalloc audio context.", __FUNCTION__, __LINE__);
		return APR_EGENERAL;
	}

	actx->paudio_buffer = (char*)malloc(RVC_MAX_AUDIO_BUFFER_LEN*sizeof(char));
	actx->uaudio_len = 0;

	actx->paudio_in = (char*)malloc(RVC_MAX_AUDIO_BUFFER_LEN * sizeof(char));
	actx->uaudio_inlen = 0;

	if (audio_init_pulseaudio(actx)) {
		audio_log_v(AUDIO_LOG_LEVEL_ERROR, "%s:%d audio init pulse audio failed.", __FUNCTION__, __LINE__);
	}
	else{
		audio_log_v(AUDIO_LOG_LEVEL_INFO, "%s:%d audio init pulse audio success.", __FUNCTION__, __LINE__);
	}

	*audio_ctx = actx;

	return APR_SUCCESS;
}

uint64_t ns_time_monotonic()
{
	struct timespec now;

	if (clock_gettime(CLOCK_MONOTONIC, &now) != 0)
	{
		return 0;
	}

	return ((uint64_t)now.tv_sec * NSEC_PER_SEC + (uint64_t)now.tv_nsec);
}


static void get_latency(pa_stream* s)
{
	pa_usec_t l;
	int negative;
	pa_stream_get_timing_info(s);
	if (pa_stream_get_latency(s, &l, &negative) != 0)
	{
		return;
	}
	latency = l;
}


static void get_play_latency(pa_stream* s)
{
	pa_usec_t l;
	int negative;

	pa_stream_get_timing_info(s);
	if (pa_stream_get_latency(s, &l, &negative) != 0) {
		return;
	}
	play_latency = l;
}



static void  stream_write_request_cb(pa_stream* s, size_t length, void* data)
{
	audio_context_t* audio_ctx = (audio_context_t*)data;

	if (0 == audio_ctx->play_channels || 0 == audio_ctx->play_samprate) {
		return;
	}

	size_t nbytes = 0;
	void* audiodata;
	while ((nbytes = pa_stream_writable_size(s)) != (size_t)-1)
	{
		get_play_latency(s);
		if (0 == nbytes){
			return;
		}

		/*write to stream*/
		if (PA_OK == pa_stream_begin_write(s, &audiodata, &nbytes))
		{
			while (audio_ctx->uaudio_len < nbytes)
			{
				char delaybuffer[RVC_DELAY_AUDIO_LEN] = { 0 };
				if (audio_ctx->bstart_get_flag && audio_ctx->micspkpulse_parent && audio_ctx->play_stream_flag) {
					audiomicspkpulse_t* audio_micspk = (audiomicspkpulse_t*)audio_ctx->micspkpulse_parent;
					if (0 == audio_micspk->ply_buf_cnt) {
						int iget = delay_buf_get((delay_buf*)audio_micspk->ply_dbuf, (short*)delaybuffer);
						if (0 == iget){
							//char audionsbuffer[RVC_DELAY_AUDIO_LEN] = { 0 };
							if (NULL != audio_micspk->on_audio_playing) {
								audio_micspk->on_audio_playing((void*)delaybuffer, RVC_DELAY_AUDIO_LEN, audio_micspk->user_data);
							}
							else {
								audio_log_v(AUDIO_LOG_LEVEL_INFO, "%s:%d on_audio_playing is NULL.", __FUNCTION__, __LINE__);
							}
							//if (0 == audio_micspk->on_audio_play_ns(audionsbuffer, RVC_DELAY_AUDIO_LEN, delaybuffer, RVC_DELAY_AUDIO_LEN, audio_micspk->user_data)) {
								if (audio_ctx->uaudio_len + RVC_DELAY_AUDIO_LEN < RVC_MAX_AUDIO_BUFFER_LEN) {
									memcpy(audio_ctx->paudio_buffer + audio_ctx->uaudio_len, delaybuffer, RVC_DELAY_AUDIO_LEN);
									audio_ctx->uaudio_len += RVC_DELAY_AUDIO_LEN;
								}
								else {
									memcpy(audio_ctx->paudio_buffer + audio_ctx->uaudio_len, delaybuffer, RVC_MAX_AUDIO_BUFFER_LEN - audio_ctx->uaudio_len);
									audio_ctx->uaudio_len = nbytes = RVC_MAX_AUDIO_BUFFER_LEN;									
									break;
								}
							//}
						}
					}
				}
			}

			int ileft = audio_ctx->uaudio_len - nbytes;
			memcpy(audiodata, audio_ctx->paudio_buffer, nbytes);
			audio_ctx->uaudio_len = ileft;
			if (ileft > 0) {
				memcpy(audio_ctx->paudio_buffer, audio_ctx->paudio_buffer + nbytes, ileft);
			}

			if (PA_OK != pa_stream_write(s, audiodata, nbytes, NULL, 0, PA_SEEK_RELATIVE)) {
				//audio_log_v(AUDIO_LOG_LEVEL_INFO, "%s:%d pa_stream_write failed.", __FUNCTION__, __LINE__);
				break;
			}
			//else {
			//	audio_log_v(AUDIO_LOG_LEVEL_INFO, "%s:%d pa_stream_write success.", __FUNCTION__, __LINE__);
			//}
		}
		//else {
		//	audio_log_v(AUDIO_LOG_LEVEL_INFO, "%s:%d pa_stream_begin_write failed for %s.", __FUNCTION__, __LINE__, pa_strerror(pa_context_errno(play_pa_ctx)));
		//}
	}
}

static void stream_request_cb(pa_stream* s, size_t length, void* data)
{
	audio_context_t* audio_ctx = (audio_context_t*)data;
	if (0 == audio_ctx->channels || 0 == audio_ctx->samprate){
		return;
	}

	int64_t ts = 0;
	while (pa_stream_readable_size(s) > 0){
		const void* inputBuffer;
		size_t length = 0;
		int icount = 0;
		bool bhasput = false;
		size_t ucopy = 0;
		size_t uleft = 0;
		/*read from stream*/
		if (pa_stream_peek(s, &inputBuffer, &length) < 0){
			audio_log_v(AUDIO_LOG_LEVEL_INFO, "AUDIO: pulse audio pa_stream_peek failed.");
			return;
		}
		//else {
		//	audio_log_v(AUDIO_LOG_LEVEL_INFO, "%s:%d pa_stream_peek audio length is %d.", __FUNCTION__, __LINE__, length);
		//}

		if (length == 0){
			//audio_log_v(AUDIO_LOG_LEVEL_INFO, "AUDIO: (pulse audio) empty buffer!");
			return; /*buffer is empty*/
		}
		get_latency(s);
		ts = ns_time_monotonic() - (latency * 1000);

		if (audio_ctx->last_ts <= 0) {
			audio_ctx->last_ts = ts;
		}

		if (audio_ctx->uaudio_inlen + length <= RVC_MAX_AUDIO_BUFFER_LEN){
			memcpy(audio_ctx->paudio_in + audio_ctx->uaudio_inlen, inputBuffer, length);
			audio_ctx->uaudio_inlen += length;
		}
		else{
			if (RVC_MAX_AUDIO_BUFFER_LEN >= audio_ctx->uaudio_inlen){
				ucopy = RVC_MAX_AUDIO_BUFFER_LEN - audio_ctx->uaudio_inlen;
				memcpy(audio_ctx->paudio_in + audio_ctx->uaudio_inlen, inputBuffer, ucopy);
				audio_ctx->uaudio_inlen = RVC_MAX_AUDIO_BUFFER_LEN;
				uleft = length - ucopy;
			}
			else{
				audio_ctx->uaudio_inlen = 0;
			}
		}

		icount = 0;
		while (audio_ctx->uaudio_inlen > RVC_DELAY_AUDIO_LEN) {
			if (audio_ctx->bstart_put_flag && audio_ctx->micspkpulse_parent && audio_ctx->stream_flag) {
				audiomicspkpulse_t* audio_micspk = audio_ctx->micspkpulse_parent;
				char paudions[RVC_DELAY_AUDIO_LEN] = { 0 };
				if (0 == audio_micspk->on_audio_ns(paudions, RVC_DELAY_AUDIO_LEN, (short*)audio_ctx->paudio_in + icount * RVC_DELAY_AUDIO_LEN / sizeof(short), RVC_DELAY_AUDIO_LEN, audio_micspk->user_data)){
					delay_buf_put((delay_buf*)audio_micspk->rec_dbuf, paudions);
				}
				icount++;
				audio_ctx->uaudio_inlen -= RVC_DELAY_AUDIO_LEN;
				bhasput = true;
			}
			else{
				break;
			}
		}
			
		if (bhasput && audio_ctx->uaudio_inlen > 0){
			if (icount * RVC_DELAY_AUDIO_LEN < RVC_MAX_AUDIO_BUFFER_LEN){
				memcpy(audio_ctx->paudio_in, (short*)audio_ctx->paudio_in + icount * RVC_DELAY_AUDIO_LEN / sizeof(short), audio_ctx->uaudio_inlen);
			}
		}

		if (uleft > 0 && audio_ctx->uaudio_inlen >= 0){
			if (audio_ctx->uaudio_inlen + uleft <= RVC_MAX_AUDIO_BUFFER_LEN){
				memcpy(audio_ctx->paudio_in + audio_ctx->uaudio_inlen, inputBuffer+ ucopy, uleft);
				audio_ctx->uaudio_inlen += uleft;
			}
		}
		pa_stream_drop(s); /*clean the samples*/
	}
}


void* pulse_read_audio(void* data)
{
	audio_context_t* audio_ctx = (audio_context_t*)data;
	/*assertions*/
	assert(audio_ctx != NULL);

	pa_mainloop* pa_ml;
	pa_mainloop_api* pa_mlapi;
	pa_context* pa_ctx;
	pa_buffer_attr bufattr;
	pa_sample_spec ss;
	pa_stream_flags_t flags = PA_STREAM_NOFLAGS;
	int32_t pastream_flag = (int32_t)PA_STREAM_NOFLAGS;
	int r;
	int pa_ready = 0;
	char* dev = NULL;

	/* Create a mainloop API and connection to the default server */
	pa_ml = pa_mainloop_new();
	pa_mlapi = pa_mainloop_get_api(pa_ml);
	pa_ctx = pa_context_new(pa_mlapi, "rvc pulse api");

	if (pa_context_connect(pa_ctx, NULL, PA_CONTEXT_NOFLAGS, NULL) < 0){
		audio_log_v(AUDIO_LOG_LEVEL_INFO,"AUDIO: PULSE - unable to connect to server: pa_context_connect failed");
		finish(pa_ctx, pa_ml);
		return ((void*)-1);
	}

	/*
	 * This function defines a callback so the server will tell us it's state.
	 * Our callback will wait for the state to be ready.  The callback will
	 * modify the variable to 1 so we know when we have a connection and it's
	 * ready.
	 * If there's an error, the callback will set pa_ready to 2
	 */
	pa_context_set_state_callback(pa_ctx, pa_state_cb, &pa_ready);

	/*
	 * This function defines a time event callback (called every TIME_EVENT_USEC)
	 */
	 //pa_context_rttime_new(pa_ctx, pa_rtclock_now() + TIME_EVENT_USEC, time_event_callback, NULL);

	 /*
	  * We can't do anything until PA is ready, so just iterate the mainloop
	  * and continue
	  */
	while (pa_ready == 0){
		pa_mainloop_iterate(pa_ml, 1, NULL);
	}
	if (pa_ready == 2){
		finish(pa_ctx, pa_ml);
		return ((void*)-1);
	}

	/* set the sample spec (frame rate, channels and format) */
	ss.rate = audio_ctx->samprate;
	ss.channels = audio_ctx->channels;
	ss.format = audio_ctx->eformat; /*for PCM -> PA_SAMPLE_S16LE*/

	pa_stream* recordstream = pa_stream_new(pa_ctx, "Record", &ss, NULL);
	if (!recordstream){
		audio_log_v(AUDIO_LOG_LEVEL_INFO, "pulse audio pa_stream_new failed (chan:%d rate:%d)", ss.channels, ss.rate);
	}

	/* define the callbacks */
	pa_stream_set_read_callback(recordstream, stream_request_cb, (void*)audio_ctx);
	// Set properties of the record buffer
	pa_zero(bufattr);
	/* optimal value for all is (uint32_t)-1   ~= 2 sec */
	bufattr.maxlength = (uint32_t)-1;
	bufattr.prebuf = (uint32_t)-1;
	bufattr.minreq = (uint32_t)-1;

	if (audio_ctx->latency > 0) {
		bufattr.fragsize = bufattr.tlength = pa_usec_to_bytes((audio_ctx->latency * 1000) * PA_USEC_PER_MSEC, &ss);
		pastream_flag |= PA_STREAM_ADJUST_LATENCY;
	}
	else {
		bufattr.fragsize = bufattr.tlength = (uint32_t)-1;
	}

	pastream_flag |= PA_STREAM_INTERPOLATE_TIMING;
	pastream_flag |= PA_STREAM_AUTO_TIMING_UPDATE;

	dev = audio_ctx->list_input_devices[audio_ctx->device].name;
	
	audio_log_v(AUDIO_LOG_LEVEL_INFO,"pulse audio connecting to device %s (channels %d rate %d)",dev, ss.channels, ss.rate);

	r = pa_stream_connect_record(recordstream, dev, &bufattr, (pa_stream_flags_t)pastream_flag);
	if (r < 0){
		audio_log_v(AUDIO_LOG_LEVEL_INFO, "AUDIO: (pulse audio) skip latency adjustment");
		/*
		 * Old pulse audio servers don't like the ADJUST_LATENCY flag,
		 * so retry without that
		 */
		r = pa_stream_connect_record(recordstream, dev, &bufattr, ((int32_t)PA_STREAM_INTERPOLATE_TIMING | (int32_t)PA_STREAM_AUTO_TIMING_UPDATE));
	}

	if (r < 0){
		audio_log_v(AUDIO_LOG_LEVEL_INFO, "AUDIO: (pulse audio) pa_stream_connect_record failed for %d.", pa_context_errno(pa_ctx));
		finish(pa_ctx, pa_ml);
		return ((void*)-1);
	}

	get_latency(recordstream);

	/*
	 * Iterate the main loop while streaming.  The second argument is whether
	 * or not the iteration should block until something is ready to be
	 * done.  Set it to zero for non-blocking.
	 */

	while (audio_ctx->stream_flag == AUDIO_STRM_ON){
		pa_mainloop_iterate(pa_ml, 1, NULL);
	}

	usleep(10000);
	pa_stream_set_read_callback(recordstream, NULL, NULL);
	audio_log_v(AUDIO_LOG_LEVEL_INFO, "AUDIO: pulse audio stream terminated(%i)", audio_ctx->stream_flag);

	pa_stream_disconnect(recordstream);
	pa_stream_unref(recordstream);
	finish(pa_ctx, pa_ml);

	return ((void*)0);
}


int audio_start_pulseaudio(audio_context_t* audio_ctx)
{
	/*assertions*/
	assert(audio_ctx != NULL);

	audio_ctx->stream_flag = AUDIO_STRM_ON;

	/* start audio capture thread */
	if (pthread_create(&audio_ctx->readthreadid, NULL, pulse_read_audio, (void*)audio_ctx)) {
		audio_log_v(AUDIO_LOG_LEVEL_INFO, "AUDIO: (pulse audio) read thread creation failed.");
		audio_ctx->stream_flag = AUDIO_STRM_OFF;
		return (-1);
	}
	else {
		audio_log_v(AUDIO_LOG_LEVEL_INFO, "AUDIO: (pulse audio) read thread create success, and thread id is %u.", audio_ctx->readthreadid);
	}

	return 0;
}


static void stream_latency_cb(pa_stream* p, void* userdata) 
{
	pa_operation* o;
	o = pa_stream_update_timing_info(p, NULL, NULL);
	pa_operation_unref(o);
}

void* pulse_write_audio(void* data)
{
	audio_context_t* audio_ctx = (audio_context_t*)data;
	/*assertions*/
	assert(audio_ctx != NULL);

	pa_mainloop* pa_ml;
	pa_mainloop_api* pa_mlapi;
	pa_buffer_attr bufattr;
	pa_sample_spec ss;
	pa_stream_flags_t flags = PA_STREAM_NOFLAGS;
	int32_t pastream_flag = (int32_t)PA_STREAM_NOFLAGS;

	int r;
	int pa_ready = 0;
	char* dev = NULL;

	/* Create a mainloop API and connection to the default server */
	pa_ml = pa_mainloop_new();
	pa_mlapi = pa_mainloop_get_api(pa_ml);
	pa_context* play_pa_ctx = pa_context_new(pa_mlapi, "rvc play api");

	if (PA_OK != pa_context_connect(play_pa_ctx, NULL, PA_CONTEXT_NOFLAGS, NULL)) {
		audio_log_v(AUDIO_LOG_LEVEL_INFO, "AUDIO: PULSE - unable to connect to server: pa_context_connect failed for %d.", pa_context_errno(play_pa_ctx));
		finish(play_pa_ctx, pa_ml);
		return ((void*)-1);
	}

	/*
	 * This function defines a callback so the server will tell us it's state.
	 * Our callback will wait for the state to be ready.  The callback will
	 * modify the variable to 1 so we know when we have a connection and it's
	 * ready.
	 * If there's an error, the callback will set pa_ready to 2
	 */
	pa_context_set_state_callback(play_pa_ctx, pa_state_cb, &pa_ready);
	 /*
	  * We can't do anything until PA is ready, so just iterate the mainloop
	  * and continue
	  */

	while (pa_ready == 0) {
		pa_mainloop_iterate(pa_ml, 1, NULL);
	}
	if (pa_ready == 2) {
		finish(play_pa_ctx, pa_ml);
		return ((void*)-1);
	}

	/* set the sample spec (frame rate, channels and format) */
	ss.rate = audio_ctx->play_samprate;
	ss.channels = audio_ctx->play_channels;
	ss.format = audio_ctx->play_eformat;

	pa_stream* playstream = pa_stream_new(play_pa_ctx, "playStream", &ss, NULL);
	if (!playstream) {
		audio_log_v(AUDIO_LOG_LEVEL_INFO, "play audio pa_stream_new failed (chan:%d rate:%d) for %d.", ss.channels, ss.rate, pa_context_errno(play_pa_ctx));
	}

	audio_log_v(AUDIO_LOG_LEVEL_INFO, "play audio stream state is %d.", pa_stream_get_state(playstream));
	/* define the callbacks */
	pa_stream_set_write_callback(playstream, stream_write_request_cb, (void*)audio_ctx);

	//pa_stream_set_latency_update_callback(playstream, stream_latency_cb, NULL);

	// Set properties of the record buffer
	pa_zero(bufattr);
	/* optimal value for all is (uint32_t)-1   ~= 2 sec */
	bufattr.maxlength = (uint32_t)-1;
	bufattr.prebuf = (uint32_t)-1;
	bufattr.minreq = (uint32_t)-1;

	if (audio_ctx->play_latency > 0) {
		bufattr.fragsize = bufattr.tlength = pa_usec_to_bytes((audio_ctx->play_latency * 1000) * PA_USEC_PER_MSEC, &ss);
		pastream_flag |= PA_STREAM_ADJUST_LATENCY;
	}
	else{
		bufattr.fragsize = bufattr.tlength = (uint32_t)-1;
	}
	
	pastream_flag |= PA_STREAM_INTERPOLATE_TIMING;
	pastream_flag |= PA_STREAM_AUTO_TIMING_UPDATE;

	dev = audio_ctx->list_output_devices[audio_ctx->play_device].name;

	audio_log_v(AUDIO_LOG_LEVEL_INFO, "play audio connecting to device %s (channels %d rate %d buf frag size %d buf length %d)", dev, ss.channels, ss.rate, bufattr.fragsize, bufattr.tlength);

	// Connect the stream to a sink
	r = pa_stream_connect_playback(playstream, dev, &bufattr, (pa_stream_flags_t)pastream_flag, NULL, NULL);
	if (PA_OK != r)
	{
		audio_log_v(AUDIO_LOG_LEVEL_INFO, "play stream connected failed for %d.", pa_context_errno(play_pa_ctx));
		finish(play_pa_ctx, pa_ml);
		return ((void*)-1);
	}
	else {
		audio_log_v(AUDIO_LOG_LEVEL_INFO, "play stream connected.");
		const pa_sample_spec* spec = pa_stream_get_sample_spec(playstream);
		audio_log_v(AUDIO_LOG_LEVEL_INFO, "play stream spec->format = %d, spec->channels = %d, spec->rate = %d.", spec->format, spec->channels, spec->rate);
	}

	get_play_latency(playstream);

	/*
	 * Iterate the main loop while streaming.  The second argument is whether
	 * or not the iteration should block until something is ready to be
	 * done.  Set it to zero for non-blocking.
	 */

	while (audio_ctx->play_stream_flag == AUDIO_STRM_ON) {
		pa_mainloop_iterate(pa_ml, 1, NULL);
	}

	usleep(10000);
	pa_stream_set_write_callback(playstream, NULL, NULL);
	audio_log_v(AUDIO_LOG_LEVEL_INFO, "AUDIO: play audio stream terminated(%i)", audio_ctx->play_stream_flag);

	pa_stream_disconnect(playstream);
	pa_stream_unref(playstream);
	finish(play_pa_ctx, pa_ml);

	return ((void*)0);
}


int audio_start_audioplay(audio_context_t* audio_ctx)
{
	/*assertions*/
	assert(audio_ctx != NULL);

	audio_ctx->play_stream_flag = AUDIO_STRM_ON;

	/* start audio capture thread */
	if (pthread_create(&audio_ctx->writethreadid, NULL, pulse_write_audio, (void*)audio_ctx)) {
		audio_log_v(AUDIO_LOG_LEVEL_INFO, "AUDIO: (pulse audio) write thread creation failed.");
		audio_ctx->play_stream_flag = AUDIO_STRM_OFF;
		return -1;
	}
	else {
		audio_log_v(AUDIO_LOG_LEVEL_INFO, "AUDIO: (pulse audio) write thread create success, and thread id is %u.", audio_ctx->writethreadid);
	}

	return 0;
}


int audio_stop_playaudio(audio_context_t* audio_ctx)
{
	/*assertions*/
	assert(audio_ctx != NULL);
	if (AUDIO_STRM_ON == audio_ctx->play_stream_flag){
		audio_ctx->play_stream_flag = AUDIO_STRM_OFF;
		if (0 != audio_ctx->writethreadid) {
			if (0 == pthread_join(audio_ctx->writethreadid, NULL)) {
			//struct timespec ts;
			//clock_gettime(CLOCK_REALTIME, &ts);
			//long unsec = ts.tv_nsec + (1000 * 1000 * 1000);
			//ts.tv_sec += (unsec / 1000000000);
			//ts.tv_nsec = (unsec % 1000000000);
			//if (0 == pthread_timedjoin_np(audio_ctx->writethreadid, NULL, &ts)) {
				audio_log_v(AUDIO_LOG_LEVEL_INFO, "%s:%d pulse audio write thread %u joined success.", __FUNCTION__, __LINE__, audio_ctx->writethreadid);
				audio_ctx->writethreadid = 0;
			}
			else {
				audio_log_v(AUDIO_LOG_LEVEL_INFO, "%s:%d pulse audio write thread joined failed for %s.", __FUNCTION__, __LINE__, strerror(errno));
			}
		}
	}

	return 0;
}

int audio_stop_pulseaudio(audio_context_t* audio_ctx)
{
	/*assertions*/
	assert(audio_ctx != NULL);
	if (AUDIO_STRM_ON == audio_ctx->stream_flag){
		audio_ctx->stream_flag = AUDIO_STRM_OFF;
		if (0 != audio_ctx->readthreadid){
			if (0 == pthread_join(audio_ctx->readthreadid, NULL)) {
			//struct timespec ts;
			//clock_gettime(CLOCK_REALTIME, &ts);
			//long unsec = ts.tv_nsec + (1000 * 1000 * 1000);
			//ts.tv_sec += (unsec / 1000000000);
			//ts.tv_nsec = (unsec % 1000000000);
			//if (0 == pthread_timedjoin_np(audio_ctx->readthreadid, NULL, &ts)) {
				audio_log_v(AUDIO_LOG_LEVEL_INFO, "%s:%d pulse audio read thread %u joined success.", __FUNCTION__, __LINE__, audio_ctx->readthreadid);
				audio_ctx->readthreadid = 0;
			}
			else {
				audio_log_v(AUDIO_LOG_LEVEL_INFO, "%s:%d pulse audio read thread joined failed for %s.", __FUNCTION__, __LINE__, strerror(errno));
			}
		}
	}

	return 0;
}


void audio_close_pulseaudio(audio_context_t* audio_ctx)
{
	if (audio_ctx == NULL) {
		return;
	}	
	
	if (audio_ctx->play_stream_flag == AUDIO_STRM_ON) {
		audio_stop_playaudio(audio_ctx);
	}

	if (audio_ctx->stream_flag == AUDIO_STRM_ON) {
		audio_stop_pulseaudio(audio_ctx);
	}

	if (NULL != audio_ctx->list_input_devices) {
		free(audio_ctx->list_input_devices);
		audio_ctx->list_input_devices = NULL;
	}

	if (NULL != audio_ctx->list_output_devices) {
		free(audio_ctx->list_output_devices);
		audio_ctx->list_output_devices = NULL;
	}
	
	if (NULL != audio_ctx->paudio_buffer){
		free(audio_ctx->paudio_buffer);
		audio_ctx->paudio_buffer = NULL;
		audio_ctx->uaudio_len = 0;
	}

	if (NULL != audio_ctx->paudio_in) {
		free(audio_ctx->paudio_in);
		audio_ctx->paudio_in = NULL;
		audio_ctx->uaudio_inlen = 0;
	}
}


void audio_set_latency(audio_context_t* audio_ctx, double latency)
{
	/*assertions*/
	assert(audio_ctx != NULL);
	audio_ctx->latency = latency;
}


void audio_set_play_latency(audio_context_t* audio_ctx, double latency)
{
	/*assertions*/
	assert(audio_ctx != NULL);
	audio_ctx->play_latency = latency;
}

void audio_set_samprate(audio_context_t* audio_ctx, int samprate)
{
	/*assertions*/
	assert(audio_ctx != NULL);
	audio_ctx->samprate = samprate;
}

void audio_set_play_samprate(audio_context_t* audio_ctx, int samprate)
{
	/*assertions*/
	assert(audio_ctx != NULL);
	audio_ctx->play_samprate = samprate;
}

void audio_set_channels(audio_context_t* audio_ctx, int channels)
{
	/*assertions*/
	assert(audio_ctx != NULL);
	audio_ctx->channels = channels;
}

void audio_set_play_channels(audio_context_t* audio_ctx, int channels)
{
	/*assertions*/
	assert(audio_ctx != NULL);
	audio_ctx->play_channels = channels;
}

void audio_set_capformat(audio_context_t* audio_ctx, pa_sample_format_t eformat)
{
	/*assertions*/
	assert(audio_ctx != NULL);
	audio_ctx->eformat = eformat;
}

void audio_set_playformat(audio_context_t* audio_ctx, pa_sample_format_t eformat)
{
	/*assertions*/
	assert(audio_ctx != NULL);
	audio_ctx->play_eformat = eformat;
}

void audio_set_capdeviceid(audio_context_t* audio_ctx, int ideviceid)
{
	/*assertions*/
	assert(audio_ctx != NULL);
	audio_ctx->device = ideviceid;
}


void audio_set_playdeviceid(audio_context_t* audio_ctx, int ideviceid)
{
	/*assertions*/
	assert(audio_ctx != NULL);
	audio_ctx->play_device = ideviceid;
}

static int initialize_speaker(audiomicspkpulse_t* micspk)
{
	int iret = -1;
	int ply_dev_id = micspk->ply_dev_id;

	if (-1 == ply_dev_id) {
		audio_log_v(AUDIO_LOG_LEVEL_ERROR, "audio speaker create error, cannot find output device.");
		return APR_EGENERAL;
	}

	audio_set_playdeviceid(micspk->audio_ctx, ply_dev_id);
	//audio_set_play_latency(micspk->audio_ctx, 0.01325);
	audio_set_play_latency(micspk->audio_ctx, 0.02);
	audio_set_play_samprate(micspk->audio_ctx, 8000);
	audio_set_play_channels(micspk->audio_ctx, 1);
	audio_set_playformat(micspk->audio_ctx, PA_SAMPLE_S16LE);

	if (0 == audio_start_audioplay(micspk->audio_ctx)) {
		audio_log_v(AUDIO_LOG_LEVEL_INFO, "audio speaker create success, audio output device start play audio success!");
		iret = 0;
	}
	else {
		audio_log_v(AUDIO_LOG_LEVEL_INFO, "audio speaker create success, audio output device start play audio failed!");
	}

	return iret;
}



static int initialize_micro(audiomicspkpulse_t* micspk)
{
	int iret = -1;
	int micro_dev_id = micspk->rec_dev_id;

	if (-1 == micro_dev_id) {
		audio_log_v(AUDIO_LOG_LEVEL_ERROR, "audio micro create error, cannot find input device.");
		return APR_EGENERAL;
	}

	audio_set_capdeviceid(micspk->audio_ctx, micro_dev_id);
	audio_set_latency(micspk->audio_ctx, 0.01);
	audio_set_samprate(micspk->audio_ctx, 8000);
	audio_set_channels(micspk->audio_ctx, 1);
	audio_set_capformat(micspk->audio_ctx, PA_SAMPLE_S16LE);

	if (0 == audio_start_pulseaudio(micspk->audio_ctx)) {
		audio_log_v(AUDIO_LOG_LEVEL_INFO, "audio micro create success, audio input device start pulse audio success!");
		iret = 0;
	}
	else {
		audio_log_v(AUDIO_LOG_LEVEL_INFO, "audio micro create success, audio input device start pulse audio failed!");
	}

	return iret;
}

static void uninitialize_speaker(audiomicspkpulse_t* micspk)
{
	if (micspk->baudio_device_started_flag) {
		if (micspk->audio_ctx) {
			audio_stop_playaudio(micspk->audio_ctx);
		}
	}

	if (micspk->ply_dbuf) {
		delay_buf_destroy((delay_buf*)micspk->ply_dbuf);
		micspk->ply_dbuf = NULL;
	}

	audio_log_v(AUDIO_LOG_LEVEL_INFO, "uninitialize_speaker success!");
}


static void uninitialize_micro(audiomicspkpulse_t* micspk)
{
	if (micspk->baudio_device_started_flag) {
		if (micspk->audio_ctx) {
			audio_stop_pulseaudio(micspk->audio_ctx);
		}
	}

	if (micspk->rec_dbuf) {
		delay_buf_destroy((delay_buf*)micspk->rec_dbuf);
		micspk->rec_dbuf = NULL;
	}
	audio_log_v(AUDIO_LOG_LEVEL_INFO, "uninitialize_micro success!");
}


void* APR_THREAD_FUNC* audiowork_proc(apr_thread_t* threadhandle, void* param)
{
	audiomicspkpulse_t* micspk = (audiomicspkpulse_t*)param;

	int rc;
	//
	// record need play because of AEC, so 
	// record <---> record and play
	// play   <---> play
	// record and play <---> record and play
	//
	audio_log_v(AUDIO_LOG_LEVEL_INFO, "%s:%d micspk addr is 0x%08x, current sem addr is 0x%08x.started flag is %s.", __FUNCTION__, __LINE__, param, micspk->audio_device_started_sem, micspk->baudio_device_started_flag ? "true" : "false");
	audio_log_v(AUDIO_LOG_LEVEL_INFO, "%s:%d micspk->opt = %d.", __FUNCTION__, __LINE__, micspk->opt);

	if (micspk->opt & AMS_OPT_RECPLAY) {
		rc = initialize_speaker(micspk);
		if (rc != 0) {
			micspk->on_audio_device_event(true, -1, false, micspk->dev_type, "recplay mode initialize speaker louder param failed, goto error.", micspk);
			goto on_error;
		}
		else {
			
			micspk->on_audio_device_event(true, 0, false, micspk->dev_type, "recplay mode initialize speaker louder param success!", micspk);
		}

		usleep(100 * 1000); // play before record
		rc = initialize_micro(micspk);
		if (0 != rc) {
			micspk->on_audio_device_event(true, -1, true, micspk->dev_type, "recplay mode initialize micro capture param failed, goto error.", micspk);
			goto on_error;
		}
		else {
			micspk->on_audio_device_event(true, 0, true, micspk->dev_type, "recplay mode initialize micro capture param success!", micspk);
		}
	}
	else if (micspk->opt & AMS_OPT_PLAY) {
		rc = initialize_speaker(micspk);
		if (rc != 0) {
			micspk->on_audio_device_event(true, -1, false, micspk->dev_type, "play mode initialize speaker louder param failed, goto error!", micspk);
			goto on_error;
		}
		else {
			micspk->on_audio_device_event(true, 0, false, micspk->dev_type, "play mode initialize speaker louder param success!", micspk);
		}
	}
	else if (micspk->opt & AMS_OPT_RECORD)
	{
		rc = initialize_micro(micspk);
		if (0 != rc) {
			micspk->on_audio_device_event(true, -1, true, micspk->dev_type, "record mode initialize micro capture param failed, goto error!", micspk);
			goto on_error;
		}
		else {
			micspk->on_audio_device_event(true, 0, true, micspk->dev_type, "record mode initialize micro capture param success!", micspk);
		}
	}

	micspk->baudio_device_started_flag = true;
	audio_log_v(AUDIO_LOG_LEVEL_INFO, "%s:%d micspk addr is 0x%08x, current sem addr is 0x%08x.", __FUNCTION__, __LINE__, micspk, micspk->audio_device_started_sem);

	sem_wait(micspk->audio_device_started_sem);
	audio_log_v(AUDIO_LOG_LEVEL_INFO, "%s:%d after post audio_device_started_sem.", __FUNCTION__, __LINE__);

on_error:
	if (micspk->opt & AMS_OPT_RECPLAY) {
		uninitialize_micro(micspk);
		micspk->on_audio_device_event(false, 0, true, micspk->dev_type, "recplay mode uninitialize micro!", micspk);
		uninitialize_speaker(micspk);
		micspk->on_audio_device_event(false, 0, false, micspk->dev_type, "recplay mode uninitialize speaker!", micspk);
	}
	else if (micspk->opt & AMS_OPT_PLAY) {
		uninitialize_speaker(micspk);
		micspk->on_audio_device_event(false, 0, false, micspk->dev_type, "play mode uninitialize speaker!", micspk);
	}
	else if (micspk->opt & AMS_OPT_RECORD){
		uninitialize_micro(micspk);
		micspk->on_audio_device_event(false, 0, true, micspk->dev_type, "record mode uninitialize micro!", micspk);
	}
	audio_log_v(AUDIO_LOG_LEVEL_INFO, "audiowork_proc exit.");

	return 0;
}


apr_status_t audiomicspkpulse_create(apr_pool_t* pool,
	audioengine_t* engine,
	int opt,
	int clock,
	const char* rec_dev_key,
	const char* ply_dev_key,
	int idev_type,
	lpfn_audio_device_event lpevent,
	audiomicspkpulse_t** p_micspk)
{
	audiomicspkpulse_t* micspk;
	unsigned long play_frame_samples;
	unsigned long capture_frame_samples;

	micspk = (audiomicspkpulse_t*)apr_palloc(pool, sizeof(audiomicspkpulse_t));
	memset(micspk, 0, sizeof(audiomicspkpulse_t));
	micspk->audio_device_started_sem = (sem_t*)apr_palloc(pool, sizeof(sem_t));

	if (APR_SUCCESS != audio_context_create(pool, &micspk->audio_ctx)){
		return APR_EGENERAL;
	}

	micspk->rec_dev_id = get_device_id(micspk->audio_ctx, 1, rec_dev_key);
	micspk->ply_dev_id = get_device_id(micspk->audio_ctx, 0, ply_dev_key);
	micspk->on_audio_device_event = lpevent;
	micspk->dev_type = idev_type;

	if (-1 == micspk->rec_dev_id || -1 == micspk->ply_dev_id) {
		audio_log_v(AUDIO_LOG_LEVEL_INFO, "%s:%d, get device id failed!", __FUNCTION__, __LINE__);
		return APR_EGENERAL;
	}
	else {
		audio_log_v(AUDIO_LOG_LEVEL_INFO, "%s:%d, rec_dev_id is %d, ply_dev_id is %d.", __FUNCTION__, __LINE__, micspk->rec_dev_id, micspk->ply_dev_id);
	}

	play_frame_samples = FRAME_TIME * clock / 1000;
	capture_frame_samples = FRAME_TIME * CAPTURE_AUDIO_CLOCK / 1000;

	micspk->opt = opt;
	micspk->play_frame_samples = play_frame_samples;
	micspk->capture_frame_samples = capture_frame_samples;

	audiostream_init(engine, &g_stream_vtbl, &micspk->base);
	micspk->base.direction = 0;

	if (opt & AMS_OPT_PLAY) {
		micspk->base.direction |= STREAM_DIR_WRITE;
		delay_buf_create(clock, play_frame_samples, 1, MAX_DELAY, 0, (delay_buf * *)& micspk->ply_dbuf);
		micspk->ply_buf = (short*)apr_palloc(pool, play_frame_samples << 1);
		micspk->ply_buf_cnt = 0;
	}
	if (opt & AMS_OPT_RECORD) {
		micspk->base.direction |= STREAM_DIR_READ;
		delay_buf_create(clock, play_frame_samples, 1, MAX_DELAY, 0, (delay_buf * *)& micspk->rec_dbuf);
		micspk->rec_buf = (short*)apr_palloc(pool, capture_frame_samples << 1);
		micspk->rec_buf_cnt = 0;
	}

	micspk->baudio_device_started_flag = false;
	sem_init(micspk->audio_device_started_sem, 0, 0);
	audio_log_v(AUDIO_LOG_LEVEL_INFO, "%s:%d micspk addr is 0x%08x, current sem addr is 0x%08x.", __FUNCTION__, __LINE__, micspk, micspk->audio_device_started_sem);

	apr_status_t err = apr_thread_create(&micspk->audio_work_thread, NULL, &audiowork_proc, micspk, pool);
	if (APR_SUCCESS == err) {
		bool baudio_work_thread_exit = false;
		audio_log_v(AUDIO_LOG_LEVEL_INFO, "%s:%d audio_work_thread id is %u.", __FUNCTION__, __LINE__, micspk->audio_work_thread);

		do {
			struct timespec ts;
			int ivalue = -1;
			clock_gettime(CLOCK_REALTIME, &ts);
			long unsec = ts.tv_nsec + (1000 * 1000 * 10);
			ts.tv_sec += (unsec / 1000000000);
			ts.tv_nsec = (unsec % 1000000000);
			sem_getvalue(micspk->audio_device_started_sem, &ivalue);
			//audio_log_v(AUDIO_LOG_LEVEL_INFO, "%s:%d current sem value is %d.", __FUNCTION__, __LINE__, ivalue);
			if (-1 == sem_timedwait(micspk->audio_device_started_sem, &ts)) {
				if (ETIMEDOUT == errno) {
					if (micspk->baudio_device_started_flag) {
						audio_log_v(AUDIO_LOG_LEVEL_INFO, "%s:%d audio device is already started.", __FUNCTION__, __LINE__);
						break;
					}
				}
			}
			else {
				audio_log_v(AUDIO_LOG_LEVEL_INFO, "%s:%d audio device work thread has exit.", __FUNCTION__, __LINE__);
				baudio_work_thread_exit = true;
			}
		} while (!baudio_work_thread_exit);

		if (baudio_work_thread_exit) {
			audiomicspkpulse_destroy(micspk);
			audio_log_v(AUDIO_LOG_LEVEL_INFO, "audio work thread has exit, audiomicspk_destroy success!");
			return APR_EGENERAL;
		}
	}
	else {
		audiomicspkpulse_destroy(micspk);
		audio_log_v(AUDIO_LOG_LEVEL_INFO, "create audio micspk work thread failed, audiomicspk_destroy success!");
		return APR_EGENERAL;
	}

	micspk->audio_ctx->micspkpulse_parent = (void*)micspk;
	micspk->audio_ctx->bstart_put_flag = true;
	micspk->audio_ctx->bstart_get_flag = true;
	*p_micspk = micspk;

	return APR_SUCCESS;
}


void audiomicspkpulse_destroy(audiomicspkpulse_t* micspk)
{
	assert(NULL != micspk);
	
	if (micspk->audio_ctx){
		audio_close_pulseaudio(micspk->audio_ctx);
	}

	sem_post(micspk->audio_device_started_sem);

	if (NULL != micspk->audio_work_thread){
		apr_status_t status;
		apr_thread_join(&status, micspk->audio_work_thread);
		micspk->audio_work_thread = NULL;
	}

	if (micspk->ply_dbuf) {
		delay_buf_destroy((delay_buf*)micspk->ply_dbuf);
		micspk->ply_dbuf = NULL;
	}
	
	if (micspk->rec_dbuf) {
		delay_buf_destroy((delay_buf*)micspk->rec_dbuf);
		micspk->rec_dbuf = NULL;
	}

	if (micspk->opt & AMS_OPT_AS_ENGINE) {
		//DeleteCriticalSection(&micspk->engine_lock);
	}
	sem_destroy(micspk->audio_device_started_sem);
}