src/multimedia/dtmf/dtmf.c - T103 - Gitiles

 #include "dtmf.h"

 #define USE_SYSLOG 1
 #if USE_SYSLOG
 #include <log/log.h>
 #endif

 #define RET_OK    0
 #define RET_FAIL -1

 #define CRITICAL 0
 #define DEFAULT  1
 #define INFO     2
 #define DEBUGLEVEL DEFAULT
 #define DTMF_TAG "[LIBDTMF]"

 static guint dtmf_get_time_ms(guint base)
 {
 	guint current = g_get_monotonic_time() / G_TIME_SPAN_MILLISECOND;
 	if (current >= base) {
 		return (current - base);
 	} else {
 		return (G_MAXUINT32 - (base - current));
 	}
 }
 #if USE_SYSLOG
 #undef LOG_TAG
 #define LOG_TAG "[LIBDTMF]"
 #define PRINT_ERR(fmt, args...)			RLOGE(fmt, ##args)
 #define PRINT_DEFAULT(fmt, args...)		RLOGI(fmt, ##args)
 #define PRINT_INFO(fmt, args...)		RLOGD(fmt, ##args)
 #else
 #define app_printf(level, fmt, args...) \
 	do { if ((level) <= DEBUGLEVEL) \
 	{ fprintf (stderr, fmt, ##args); } } while (0)

 #define PRINT_ERR(fmt, args...) \
 	app_printf(CRITICAL, DTMF_TAG"[ERR][L%d][T%d]"fmt, __LINE__, dtmf_get_time_ms(0),##args);

 #define PRINT_DEFAULT(fmt, args...) \
 	app_printf(DEFAULT, DTMF_TAG"[DEF][L%d][T%d]"fmt, __LINE__, dtmf_get_time_ms(0), ##args);

 #define PRINT_INFO(fmt, args...) \
 	app_printf(INFO, DTMF_TAG"[INFO][L%d][T%d]"fmt, __LINE__, dtmf_get_time_ms(0), ##args);
 #endif

 #define FILE_NAME_MAX_LENGTH 64
 #define DTMFSRC_MUTE_LENGTH  100   //unit: ms, from the practice result, the begining 100ms data is 0.

 typedef struct {
 	DTMF_HANDLE handle;
 	gint number;		//the event number: 0-15
 	gint volume;		//db value: 0-36
 	gint time_ms;		//the duration of tone, unit: ms

 	pthread_t thread;
 	GMainLoop *loop;
 	GstElement *pipeline;
 	guint bus_watch_id;
 	GstState gst_cur_state;

 	gint output;	//0: output to pulsesink; 1: output the buffer pointer
 	char *path; 	//while output==1, file path to store the tone data
 	//FILE *fd;		// file handle to the output file
 } DTMF_PARAM_T;

 static int dtmf_stop_full(DTMF_PARAM_T *param)
 {
 	GMainLoop *loop = param->loop;
 	GstElement *pipeline = param->pipeline;
 	guint bus_watch_id = param->bus_watch_id;
 	GstStructure *structure_stop = NULL;
 	GstEvent *event_stop = NULL;

 	PRINT_DEFAULT("%s start, param: 0x%x, handle: 0x%x, number: %d, time_ms: %d, volume: %d \n",
 		__FUNCTION__, param, param->handle, param->number, param->time_ms, param->volume);

 	if (param != param->handle) {
 		PRINT_DEFAULT("invalid handle: %p \n", param->handle);
 		return RET_OK;
 	}

 	structure_stop = gst_structure_new ("dtmf-event",
 				 "type", G_TYPE_INT, 1,
 				 "number", G_TYPE_INT, 1,
 				 "volume", G_TYPE_INT, 25,
 				 "start", G_TYPE_BOOLEAN, FALSE, NULL);
 	event_stop = gst_event_new_custom (GST_EVENT_CUSTOM_UPSTREAM, structure_stop);
 	gst_element_send_event (pipeline, event_stop);

 	PRINT_DEFAULT("gst_element_set_state NULL start \n");
 	gst_element_set_state (pipeline, GST_STATE_NULL);
 	PRINT_DEFAULT("gst_element_set_state NULL end \n");

 	gst_object_unref (GST_OBJECT (pipeline));
 	g_source_remove (bus_watch_id);
 	g_main_loop_unref (loop);

 	param->handle = NULL;
 	free(param);
 	param = NULL;
 	PRINT_DEFAULT("%s end \n", __FUNCTION__);

 	return RET_OK;
 }

 static gboolean dtmf_bus_call (GstBus *bus, GstMessage *msg, gpointer data)
 {
 	DTMF_PARAM_T *param = (DTMF_PARAM_T *)data;
 	GstState oldstate, newstate, pending;
 	gchar  *debug;
 	GError *error;

 	switch (GST_MESSAGE_TYPE (msg)) {
 	case GST_MESSAGE_EOS:
 		PRINT_DEFAULT ("Receive GST_MESSAGE_EOS \n");
 		g_main_loop_quit (param->loop);
 		break;
 	case GST_MESSAGE_STATE_CHANGED:
 		gst_message_parse_state_changed (msg, &oldstate, &newstate, &pending);
 		param->gst_cur_state = newstate;
 		PRINT_INFO ("GST_MESSAGE_STATE_CHANGED, oldstate[%d], newstate[%d], pending[%d] \n",
 			oldstate, newstate, pending);
 		break;
 	case GST_MESSAGE_ERROR:
 		gst_message_parse_error (msg, &error, &debug);
 		g_free (debug);
 		PRINT_ERR ("Error: %s\n", error->message);
 		g_error_free (error);
 		g_main_loop_quit (param->loop);
 		break;
     default:
 		break;
   }

   return TRUE;
 }

 #if 0
 void dtmf_wait_ms(gint ms)
 {
 	gint err;
 	struct timeval wait_tv;
 	wait_tv.tv_sec = ms / 1000;
 	wait_tv.tv_usec = (ms % 1000) * 1000;
 	do {
 		err = select(0, NULL, NULL, NULL, &wait_tv);
 	} while(err<0 && errno == EINTR);
 }

 static GstFlowReturn dtmf_tone_data_output(GstElement *sink, void *data)
 {
 	DTMF_PARAM_T *param = (DTMF_PARAM_T *)data;
 	GstSample *sample;
 	GstBuffer *buffer;
 	GstMapInfo map;

   	sample = gst_app_sink_pull_sample (GST_APP_SINK (sink));
 	if (!sample)
 	{
 		PRINT_ERR("could not get sample. \n");
 		return GST_FLOW_ERROR;
 	}
 	buffer = gst_sample_get_buffer (sample);
 	gst_buffer_map (buffer, &map, GST_MAP_READ);
 	PRINT_INFO ("tone data:  size=%d, data=%p \n", map.size, map.data);
 	fwrite(map.data, 1, map.size, param->fd);

 	gst_buffer_unmap (buffer, &map);
 	gst_sample_unref (sample);

 	return GST_FLOW_OK;
 }
 #endif

 static gint dtmf_check_param(DTMF_PARAM_T *param)
 {
 	if ((param->number < 0) || (param->number > 15)) {
 		PRINT_ERR("invalid number: %d, valid value is [0,15] \n ", param->number);
 		return RET_FAIL;
 	}
 	if ((param->time_ms != 0) && (param->time_ms < DTMF_MIN_LENGTH)) {
 		PRINT_ERR("invalid time_ms: %d \n ", param->time_ms);
 		return RET_FAIL;
 	}
 	if ((param->volume < 0) || (param->volume > 36)) {
 		PRINT_ERR("invalid volume: %d, valid value is [0,36] \n ", param->volume);
 		return RET_FAIL;
 	}
 	if ((param->output < 0) || (param->output > 1)) {
 		PRINT_ERR("invalid output: %d, valid value is [0,1] \n ", param->output);
 		return RET_FAIL;
 	}
 	if ((param->output == 1) && (param->path == NULL)) {
 		PRINT_ERR("path is NULL while output is filesink \n ");
 		return RET_FAIL;
 	}
 	return RET_OK;
 }

 void* dtmf_thread_func(void *arg)
 {
 	DTMF_PARAM_T *param = (DTMF_PARAM_T *)arg;
 	PRINT_DEFAULT("%s start \n", __FUNCTION__);

 	g_main_loop_run (param->loop);
 	PRINT_DEFAULT("g_main_loop_run end \n");

 	dtmf_stop_full(param);

 	PRINT_DEFAULT("%s end \n", __FUNCTION__);
 	return ((void *)0);
 }

 DTMF_HANDLE dtmf_start(gint num, gint time_ms, gint volume, DTMF_EXTRA *extra)
 {
 	GMainLoop *loop;
 	GstElement *pipeline, *source, *sink;
 	GstBus *bus;
 	guint bus_watch_id;
 	pthread_t thread;
 	GstStructure *structure_start = NULL;
 	GstEvent *event_start = NULL;

 	DTMF_HANDLE handle = NULL;
 	DTMF_PARAM_T *param = malloc(sizeof(DTMF_PARAM_T));
 	if (param == NULL) {
 		PRINT_ERR("malloc DTMF_PARAM_T fail \n");
 	}
 	memset(param, 0, sizeof(DTMF_PARAM_T));
 	handle = (DTMF_HANDLE)param;

 	//param->handle = handle;
 	param->number = num;
 	param->time_ms = time_ms;
 	param->volume = volume;

 	PRINT_DEFAULT("%s start, handle: 0x%x, number: %d, time_ms: %d, volume: %d \n ",
 		__FUNCTION__, handle, param->number, param->time_ms, param->volume);

 	if (extra != NULL) {
 		param->output = extra->output;
 		param->path = extra->path;
 		PRINT_DEFAULT("output: %d, path: %s \n", extra->output, extra->path);
 	}

 	if (dtmf_check_param(param) < 0) {
 		PRINT_ERR("input parameter is invalid! \n");
 		return NULL;
 	}

 	gst_init (NULL, NULL);

 	loop = g_main_loop_new (NULL, FALSE);

 	pipeline = gst_pipeline_new ("dtmf-pipeline");
 	source   = gst_element_factory_make ("dtmfsrc", 	 "dtmf-file-source");
 	switch (param->output) {
 		case 0:
 			sink = gst_element_factory_make ("pulsesink",	 "dtmf-audio-output");
 			break;
 		case 1:
 			sink = gst_element_factory_make ("filesink",	 "dtmf-audio-output");
 			g_object_set(sink, "location", extra->path, NULL);
 			break;
 		default:
 			PRINT_ERR("input parameter 'output' is invalid! \n");
 			break;
 	}

     //sink = gst_element_factory_make ("appsink", 	 "audio-output-buffer");
 	//g_object_set(sink, "emit-signals", TRUE, "sync", TRUE, NULL);
 	//g_signal_connect(sink, "new-sample", G_CALLBACK(tone_data_output), (gpointer)param);

 	if (!pipeline || !source || !sink) {
 		PRINT_ERR ("One element creat fail, pipeline: %p, source: %p, sink: %p \n",
 			pipeline, source, sink);
 		free(param);
 		return NULL;
 	}

 	guint interval = 50;
 	gint buffers_num = -1;
 	if (time_ms > 0) {
 		// dtmfsrc plugin will send EOS when the number of buffers reach 'buffers_num' .
 		g_object_get(source, "interval", &interval, NULL);
 		buffers_num = (time_ms + DTMFSRC_MUTE_LENGTH) / interval;
 		PRINT_DEFAULT("dtmfsrc interval: %d ms, set num-buffers: %d \n",
 			interval, buffers_num);
 		g_object_set(source, "num-buffers", buffers_num, NULL);
 	}

 	bus = gst_pipeline_get_bus (GST_PIPELINE (pipeline));
 	bus_watch_id = gst_bus_add_watch (bus, dtmf_bus_call, (gpointer)param);
 	gst_object_unref (bus);

 	param->pipeline = pipeline;
 	param->loop = loop;
 	param->bus_watch_id = bus_watch_id;

 	pthread_create (&thread, NULL, (void *)dtmf_thread_func, param);
 	param->thread = thread;

 	gst_bin_add_many (GST_BIN (pipeline), source, sink, NULL);
 	gst_element_link_many (source, sink, NULL);

 	gst_element_set_state (param->pipeline, GST_STATE_PLAYING);

 	PRINT_INFO("wait GST_STATE_PLAYING start \n");
 	do {
 		if (param->gst_cur_state == GST_STATE_PLAYING)
 			break;
 		else
 			usleep(5000);
 	} while (1);
 	PRINT_INFO("wait GST_STATE_PLAYING end \n");

 	structure_start = gst_structure_new ("dtmf-event",
 				 "type", G_TYPE_INT, 1,
 				 "number", G_TYPE_INT, param->number,
 				 "volume", G_TYPE_INT, param->volume,
 				 "start", G_TYPE_BOOLEAN, TRUE, NULL);

 	//gst_structure_set (structure, "number", G_TYPE_INT, num,
     //          "volume", G_TYPE_INT, volume, NULL);

 	event_start = gst_event_new_custom (GST_EVENT_CUSTOM_UPSTREAM, structure_start);
 	gst_element_send_event (param->pipeline, event_start);

 	PRINT_DEFAULT("%s end, handle: 0x%x, number: %d, time_ms: %d, volume: %d \n",
 		__FUNCTION__, handle, param->number, param->time_ms, param->volume);

 	param->handle = handle;

 	return handle;
 }

 int dtmf_stop(DTMF_HANDLE handle)
 {
 	if (handle == NULL) {
 		PRINT_ERR ("%s, handle is NULL\n", __FUNCTION__);
 		return RET_FAIL;
 	}
 	DTMF_PARAM_T *param = (DTMF_PARAM_T *)handle;

 	PRINT_DEFAULT("%s start, DTMF_HANDLE: 0x%x, handle: 0x%x \n",
 		__FUNCTION__, handle, param->handle);

 	if (handle != param->handle) {
 		PRINT_DEFAULT("invalid handle: %p \n", handle);
 		return RET_OK;
 	}

 	g_main_loop_quit (param->loop);
 	pthread_join(param->thread, NULL);

 	PRINT_DEFAULT("%s end, handle: 0x%x \n", __FUNCTION__, handle);

 	return RET_OK;
 }
	#include "dtmf.h"

	#define USE_SYSLOG 1
	#if USE_SYSLOG
	#include <log/log.h>
	#endif

	#define RET_OK 0
	#define RET_FAIL -1

	#define CRITICAL 0
	#define DEFAULT 1
	#define INFO 2
	#define DEBUGLEVEL DEFAULT
	#define DTMF_TAG "[LIBDTMF]"

	static guint dtmf_get_time_ms(guint base)
	{
	guint current = g_get_monotonic_time() / G_TIME_SPAN_MILLISECOND;
	if (current >= base) {
	return (current - base);
	} else {
	return (G_MAXUINT32 - (base - current));
	}
	}
	#if USE_SYSLOG
	#undef LOG_TAG
	#define LOG_TAG "[LIBDTMF]"
	#define PRINT_ERR(fmt, args...) RLOGE(fmt, ##args)
	#define PRINT_DEFAULT(fmt, args...) RLOGI(fmt, ##args)
	#define PRINT_INFO(fmt, args...) RLOGD(fmt, ##args)
	#else
	#define app_printf(level, fmt, args...) \
	do { if ((level) <= DEBUGLEVEL) \
	{ fprintf (stderr, fmt, ##args); } } while (0)

	#define PRINT_ERR(fmt, args...) \
	app_printf(CRITICAL, DTMF_TAG"[ERR][L%d][T%d]"fmt, __LINE__, dtmf_get_time_ms(0),##args);

	#define PRINT_DEFAULT(fmt, args...) \
	app_printf(DEFAULT, DTMF_TAG"[DEF][L%d][T%d]"fmt, __LINE__, dtmf_get_time_ms(0), ##args);

	#define PRINT_INFO(fmt, args...) \
	app_printf(INFO, DTMF_TAG"[INFO][L%d][T%d]"fmt, __LINE__, dtmf_get_time_ms(0), ##args);
	#endif

	#define FILE_NAME_MAX_LENGTH 64
	#define DTMFSRC_MUTE_LENGTH 100 //unit: ms, from the practice result, the begining 100ms data is 0.

	typedef struct {
	DTMF_HANDLE handle;
	gint number; //the event number: 0-15
	gint volume; //db value: 0-36
	gint time_ms; //the duration of tone, unit: ms

	pthread_t thread;
	GMainLoop *loop;
	GstElement *pipeline;
	guint bus_watch_id;
	GstState gst_cur_state;

	gint output; //0: output to pulsesink; 1: output the buffer pointer
	char *path; //while output==1, file path to store the tone data
	//FILE *fd; // file handle to the output file
	} DTMF_PARAM_T;

	static int dtmf_stop_full(DTMF_PARAM_T *param)
	{
	GMainLoop *loop = param->loop;
	GstElement *pipeline = param->pipeline;
	guint bus_watch_id = param->bus_watch_id;
	GstStructure *structure_stop = NULL;
	GstEvent *event_stop = NULL;

	PRINT_DEFAULT("%s start, param: 0x%x, handle: 0x%x, number: %d, time_ms: %d, volume: %d \n",
	__FUNCTION__, param, param->handle, param->number, param->time_ms, param->volume);

	if (param != param->handle) {
	PRINT_DEFAULT("invalid handle: %p \n", param->handle);
	return RET_OK;
	}

	structure_stop = gst_structure_new ("dtmf-event",
	"type", G_TYPE_INT, 1,
	"number", G_TYPE_INT, 1,
	"volume", G_TYPE_INT, 25,
	"start", G_TYPE_BOOLEAN, FALSE, NULL);
	event_stop = gst_event_new_custom (GST_EVENT_CUSTOM_UPSTREAM, structure_stop);
	gst_element_send_event (pipeline, event_stop);

	PRINT_DEFAULT("gst_element_set_state NULL start \n");
	gst_element_set_state (pipeline, GST_STATE_NULL);
	PRINT_DEFAULT("gst_element_set_state NULL end \n");

	gst_object_unref (GST_OBJECT (pipeline));
	g_source_remove (bus_watch_id);
	g_main_loop_unref (loop);

	param->handle = NULL;
	free(param);
	param = NULL;
	PRINT_DEFAULT("%s end \n", __FUNCTION__);

	return RET_OK;
	}

	static gboolean dtmf_bus_call (GstBus bus, GstMessage msg, gpointer data)
	{
	DTMF_PARAM_T param = (DTMF_PARAM_T )data;
	GstState oldstate, newstate, pending;
	gchar *debug;
	GError *error;

	switch (GST_MESSAGE_TYPE (msg)) {
	case GST_MESSAGE_EOS:
	PRINT_DEFAULT ("Receive GST_MESSAGE_EOS \n");
	g_main_loop_quit (param->loop);
	break;
	case GST_MESSAGE_STATE_CHANGED:
	gst_message_parse_state_changed (msg, &oldstate, &newstate, &pending);
	param->gst_cur_state = newstate;
	PRINT_INFO ("GST_MESSAGE_STATE_CHANGED, oldstate[%d], newstate[%d], pending[%d] \n",
	oldstate, newstate, pending);
	break;
	case GST_MESSAGE_ERROR:
	gst_message_parse_error (msg, &error, &debug);
	g_free (debug);
	PRINT_ERR ("Error: %s\n", error->message);
	g_error_free (error);
	g_main_loop_quit (param->loop);
	break;
	default:
	break;
	}

	return TRUE;
	}

	#if 0
	void dtmf_wait_ms(gint ms)
	{
	gint err;
	struct timeval wait_tv;
	wait_tv.tv_sec = ms / 1000;
	wait_tv.tv_usec = (ms % 1000) * 1000;
	do {
	err = select(0, NULL, NULL, NULL, &wait_tv);
	} while(err<0 && errno == EINTR);
	}

	static GstFlowReturn dtmf_tone_data_output(GstElement sink, void data)
	{
	DTMF_PARAM_T param = (DTMF_PARAM_T )data;
	GstSample *sample;
	GstBuffer *buffer;
	GstMapInfo map;

	sample = gst_app_sink_pull_sample (GST_APP_SINK (sink));
	if (!sample)
	{
	PRINT_ERR("could not get sample. \n");
	return GST_FLOW_ERROR;
	}
	buffer = gst_sample_get_buffer (sample);
	gst_buffer_map (buffer, &map, GST_MAP_READ);
	PRINT_INFO ("tone data: size=%d, data=%p \n", map.size, map.data);
	fwrite(map.data, 1, map.size, param->fd);

	gst_buffer_unmap (buffer, &map);
	gst_sample_unref (sample);

	return GST_FLOW_OK;
	}
	#endif

	static gint dtmf_check_param(DTMF_PARAM_T *param)
	{
	if ((param->number < 0) \|\| (param->number > 15)) {
	PRINT_ERR("invalid number: %d, valid value is [0,15] \n ", param->number);
	return RET_FAIL;
	}
	if ((param->time_ms != 0) && (param->time_ms < DTMF_MIN_LENGTH)) {
	PRINT_ERR("invalid time_ms: %d \n ", param->time_ms);
	return RET_FAIL;
	}
	if ((param->volume < 0) \|\| (param->volume > 36)) {
	PRINT_ERR("invalid volume: %d, valid value is [0,36] \n ", param->volume);
	return RET_FAIL;
	}
	if ((param->output < 0) \|\| (param->output > 1)) {
	PRINT_ERR("invalid output: %d, valid value is [0,1] \n ", param->output);
	return RET_FAIL;
	}
	if ((param->output == 1) && (param->path == NULL)) {
	PRINT_ERR("path is NULL while output is filesink \n ");
	return RET_FAIL;
	}
	return RET_OK;
	}

	void* dtmf_thread_func(void *arg)
	{
	DTMF_PARAM_T param = (DTMF_PARAM_T )arg;
	PRINT_DEFAULT("%s start \n", __FUNCTION__);

	g_main_loop_run (param->loop);
	PRINT_DEFAULT("g_main_loop_run end \n");

	dtmf_stop_full(param);

	PRINT_DEFAULT("%s end \n", __FUNCTION__);
	return ((void *)0);
	}

	DTMF_HANDLE dtmf_start(gint num, gint time_ms, gint volume, DTMF_EXTRA *extra)
	{
	GMainLoop *loop;
	GstElement pipeline, source, *sink;
	GstBus *bus;
	guint bus_watch_id;
	pthread_t thread;
	GstStructure *structure_start = NULL;
	GstEvent *event_start = NULL;

	DTMF_HANDLE handle = NULL;
	DTMF_PARAM_T *param = malloc(sizeof(DTMF_PARAM_T));
	if (param == NULL) {
	PRINT_ERR("malloc DTMF_PARAM_T fail \n");
	}
	memset(param, 0, sizeof(DTMF_PARAM_T));
	handle = (DTMF_HANDLE)param;

	//param->handle = handle;
	param->number = num;
	param->time_ms = time_ms;
	param->volume = volume;

	PRINT_DEFAULT("%s start, handle: 0x%x, number: %d, time_ms: %d, volume: %d \n ",
	__FUNCTION__, handle, param->number, param->time_ms, param->volume);

	if (extra != NULL) {
	param->output = extra->output;
	param->path = extra->path;
	PRINT_DEFAULT("output: %d, path: %s \n", extra->output, extra->path);
	}

	if (dtmf_check_param(param) < 0) {
	PRINT_ERR("input parameter is invalid! \n");
	return NULL;
	}

	gst_init (NULL, NULL);

	loop = g_main_loop_new (NULL, FALSE);

	pipeline = gst_pipeline_new ("dtmf-pipeline");
	source = gst_element_factory_make ("dtmfsrc", "dtmf-file-source");
	switch (param->output) {
	case 0:
	sink = gst_element_factory_make ("pulsesink", "dtmf-audio-output");
	break;
	case 1:
	sink = gst_element_factory_make ("filesink", "dtmf-audio-output");
	g_object_set(sink, "location", extra->path, NULL);
	break;
	default:
	PRINT_ERR("input parameter 'output' is invalid! \n");
	break;
	}

	//sink = gst_element_factory_make ("appsink", "audio-output-buffer");
	//g_object_set(sink, "emit-signals", TRUE, "sync", TRUE, NULL);
	//g_signal_connect(sink, "new-sample", G_CALLBACK(tone_data_output), (gpointer)param);

	if (!pipeline \|\| !source \|\| !sink) {
	PRINT_ERR ("One element creat fail, pipeline: %p, source: %p, sink: %p \n",
	pipeline, source, sink);
	free(param);
	return NULL;
	}

	guint interval = 50;
	gint buffers_num = -1;
	if (time_ms > 0) {
	// dtmfsrc plugin will send EOS when the number of buffers reach 'buffers_num' .
	g_object_get(source, "interval", &interval, NULL);
	buffers_num = (time_ms + DTMFSRC_MUTE_LENGTH) / interval;
	PRINT_DEFAULT("dtmfsrc interval: %d ms, set num-buffers: %d \n",
	interval, buffers_num);
	g_object_set(source, "num-buffers", buffers_num, NULL);
	}

	bus = gst_pipeline_get_bus (GST_PIPELINE (pipeline));
	bus_watch_id = gst_bus_add_watch (bus, dtmf_bus_call, (gpointer)param);
	gst_object_unref (bus);

	param->pipeline = pipeline;
	param->loop = loop;
	param->bus_watch_id = bus_watch_id;

	pthread_create (&thread, NULL, (void *)dtmf_thread_func, param);
	param->thread = thread;

	gst_bin_add_many (GST_BIN (pipeline), source, sink, NULL);
	gst_element_link_many (source, sink, NULL);

	gst_element_set_state (param->pipeline, GST_STATE_PLAYING);

	PRINT_INFO("wait GST_STATE_PLAYING start \n");
	do {
	if (param->gst_cur_state == GST_STATE_PLAYING)
	break;
	else
	usleep(5000);
	} while (1);
	PRINT_INFO("wait GST_STATE_PLAYING end \n");

	structure_start = gst_structure_new ("dtmf-event",
	"type", G_TYPE_INT, 1,
	"number", G_TYPE_INT, param->number,
	"volume", G_TYPE_INT, param->volume,
	"start", G_TYPE_BOOLEAN, TRUE, NULL);

	//gst_structure_set (structure, "number", G_TYPE_INT, num,
	// "volume", G_TYPE_INT, volume, NULL);

	event_start = gst_event_new_custom (GST_EVENT_CUSTOM_UPSTREAM, structure_start);
	gst_element_send_event (param->pipeline, event_start);

	PRINT_DEFAULT("%s end, handle: 0x%x, number: %d, time_ms: %d, volume: %d \n",
	__FUNCTION__, handle, param->number, param->time_ms, param->volume);

	param->handle = handle;

	return handle;
	}

	int dtmf_stop(DTMF_HANDLE handle)
	{
	if (handle == NULL) {
	PRINT_ERR ("%s, handle is NULL\n", __FUNCTION__);
	return RET_FAIL;
	}
	DTMF_PARAM_T param = (DTMF_PARAM_T )handle;

	PRINT_DEFAULT("%s start, DTMF_HANDLE: 0x%x, handle: 0x%x \n",
	__FUNCTION__, handle, param->handle);

	if (handle != param->handle) {
	PRINT_DEFAULT("invalid handle: %p \n", handle);
	return RET_OK;
	}

	g_main_loop_quit (param->loop);
	pthread_join(param->thread, NULL);

	PRINT_DEFAULT("%s end, handle: 0x%x \n", __FUNCTION__, handle);

	return RET_OK;
	}