push data to appsrc

#include <stdio.h>
#include <gst/gst.h>
#include <gst/app/gstappsrc.h>

typedef struct {
 GstPipeline *pipeline;
 GstAppSrc *src;
 GstElement *sink;
 GstElement *decoder;
 GstElement *Audioconvert;
 GstElement *Audioresample;
 GstElement *alsasink;
 GMainLoop *loop;
 guint sourceid;
 FILE *file;
}gst_app_t;
GstCaps SrcCaps;
gst_app_t gst_app;

#define BUFF_SIZE (1024)

static gboolean read_data(gst_app_t *app)
{
 GstBuffer *buffer;
 guint8 *ptr;
 gint size;
 GstFlowReturn ret;

 ptr = g_malloc(BUFF_SIZE);
 g_assert(ptr);

 size = fread(ptr, 1, BUFF_SIZE, app->file);

 if(size == 0){
  ret = gst_app_src_end_of_stream(app->src);
  g_warning("eos returned %d at %d\n", ret, __LINE__);
  return FALSE;
 }

 buffer = gst_buffer_new();
 GST_BUFFER_MALLOCDATA(buffer) = ptr;
 GST_BUFFER_SIZE(buffer) = size;
 GST_BUFFER_DATA(buffer) = GST_BUFFER_MALLOCDATA(buffer);

 ret = gst_app_src_push_buffer(app->src, buffer);

 if(ret !=  GST_FLOW_OK){
  g_warning("push buffer returned %d for %d bytes \n", ret, size);
  return FALSE;
 }

 if(size != BUFF_SIZE){
  ret = gst_app_src_end_of_stream(app->src);

  return FALSE;
 }

 return TRUE;
}

static void start_feed (GstElement * pipeline, guint size, gst_app_t *app)
{
 if (app->sourceid == 0) {
  GST_DEBUG ("start feeding");

  app->sourceid = g_idle_add ((GSourceFunc) read_data, app);
 }
}

static void stop_feed (GstElement * pipeline, gst_app_t *app)
{
 if (app->sourceid != 0) {
  GST_DEBUG ("stop feeding");

  g_source_remove (app->sourceid);
  app->sourceid = 0;
 }
}

static void on_pad_added(GstElement *element, GstPad *pad)
{
 GstCaps *caps;
 GstStructure *str;
 gchar *name;
 GstPad *AudioConvertsink;
 GstPadLinkReturn ret;

 g_debug("pad added");

 caps = gst_pad_get_caps(pad);
 str = gst_caps_get_structure(caps, 0);

 g_assert(str);

 name = (gchar*)gst_structure_get_name(str);

 g_warning("pad name %s", name);

 if(g_strrstr(name, "audio")){
 g_warning("%s, %d",__FUNCTION__, __LINE__);
  AudioConvertsink = gst_element_get_pad(gst_app.Audioconvert, "sink");
  g_assert(AudioConvertsink);
  ret = gst_pad_link(pad, AudioConvertsink);
  g_warning("pad_link returned %d\n", ret);
  gst_object_unref(AudioConvertsink);
 }
 gst_caps_unref(caps);
}

static gboolean bus_callback(GstBus *bus, GstMessage *message, gpointer *ptr)
{
 gst_app_t *app = (gst_app_t*)ptr;

 switch(GST_MESSAGE_TYPE(message)){

 case GST_MESSAGE_ERROR:{
  gchar *debug;
  GError *err;

  gst_message_parse_error(message, &err, &debug);
  g_print("Error %s\n", err->message);
  g_error_free(err);
  g_free(debug);
  g_main_loop_quit(app->loop);
 }
 break;

 case GST_MESSAGE_WARNING:{
  gchar *debug;
  GError *err;
  gchar *name;

  gst_message_parse_warning(message, &err, &debug);
  g_print("Warning %s\nDebug %s\n", err->message, debug);

  name = GST_MESSAGE_SRC_NAME(message);

  g_print("Name of src %s\n", name ? name : "nil");
  g_error_free(err);
  g_free(debug);
 }
 break;

 case GST_MESSAGE_EOS:
  g_print("End of stream\n");
  g_main_loop_quit(app->loop);
  break;

 case GST_MESSAGE_STATE_CHANGED:
  break;

 default:
  g_print("got message %s\n", \
   gst_message_type_get_name (GST_MESSAGE_TYPE (message)));
  break;
 }

 return TRUE;
}

int put_payload_to_gst(const char * payload, int sz)
{
	GstBuffer *buffer;
	GstFlowReturn ret;
	gst_app_t *app = &gst_app;

	 if(sz == 0){
  		ret = gst_app_src_end_of_stream(app->src);
  		printf("eos returned %d at %d\n", ret, __LINE__);
  		return 0;
 	}

	buffer = gst_buffer_new();
 	GST_BUFFER_MALLOCDATA(buffer) = payload;
 	GST_BUFFER_SIZE(buffer) = sz;
 	GST_BUFFER_DATA(buffer) = GST_BUFFER_MALLOCDATA(buffer);

  	ret = gst_app_src_push_buffer(app->src, buffer);
	if(ret !=  GST_FLOW_OK){
  		g_warning("push buffer returned %d for %d bytes \n", ret, sz);
  		return 0;
 	}

	return 1;

}

int CreateGstPipe(void)
{
 gst_app_t *app = &gst_app;
 GstBus *bus;
 GstStateChangeReturn state_ret;
 GstCaps *pSrcCaps;

 gst_init(NULL, NULL);


 app->pipeline = (GstPipeline*)gst_pipeline_new("mypipeline");
 bus = gst_pipeline_get_bus(app->pipeline);
 gst_bus_add_watch(bus, (GstBusFunc)bus_callback, app);
 gst_object_unref(bus);

 app->src = (GstAppSrc*)gst_element_factory_make("appsrc", "mysrc");
 app->decoder = gst_element_factory_make("decodebin2", "mydecoder");
 app->Audioconvert = gst_element_factory_make ("audioconvert", "audioconvert1");
 app->Audioresample = gst_element_factory_make ("audioresample", "audioresample1");
 app->alsasink = gst_element_factory_make("alsasink", "myasink");

 g_assert(app->src);
 g_assert(app->decoder);
 g_assert(app->Audioconvert);
 g_assert(app->Audioresample);
 g_assert(app->alsasink);




 gst_app_src_set_caps(app->src, pSrcCaps);
 g_object_set (app->src, "is-live", FALSE, NULL);

 g_object_set (app->src, "do-timestamp", FALSE, NULL);

 gst_app_src_set_stream_type(app->src, GST_APP_STREAM_TYPE_SEEKABLE);

 g_signal_connect(app->decoder, "pad-added", G_CALLBACK(on_pad_added), app->decoder);

 gst_bin_add_many(GST_BIN(app->pipeline), (GstElement*)app->src,
  app->decoder, app->Audioconvert, app->Audioresample, app->alsasink, NULL);

 if(!gst_element_link((GstElement*)app->src, app->decoder)){
  g_warning("failed to link src anbd decoder");
 }

 if(!gst_element_link(app->Audioconvert, app->Audioresample)){
  g_warning("failed to link Audioconvert and Audioresample");
 }

 if(!gst_element_link(app->Audioresample, app->alsasink)){
  g_warning("failed to link Audioresample and alsasink");
 }


 state_ret = gst_element_set_state((GstElement*)app->pipeline, GST_STATE_PLAYING);

 g_warning("set state returned %d\n", state_ret);




 return 0;
}

/////////////////////////////////////////////////////////
push data here
///////////////////////////////////////////////////////////
#define ONE_PKT_SIZE		4*1024
char onepkt[ONE_PKT_SIZE];
GstFlowReturn ret;
GstBuffer *buffer;
gst_app_t *app = &gst_app;
	while(1)//////////////////////////////////GET Audio Data
	{
        iRet = Read((char*)(onepkt), ONE_PKT_SIZE, 0);
		if(iRet > 0)
		{
            		iTotalBytesRead += iRet;
            		write(localFile, onepkt, iRet);

            		buffer = gst_buffer_new_and_alloc (iRet);
			GST_BUFFER_SIZE(buffer) = iRet;
			GST_BUFFER_DATA(buffer) = GST_BUFFER_MALLOCDATA(buffer);
			memcpy(GST_BUFFER_DATA(buffer), onepkt,iRet);
			ret = gst_app_src_push_buffer(app->src, buffer);
			if(ret !=  GST_FLOW_OK){
				g_warning("push buffer returned %d for %d bytes \n", ret, iRet);

			}

            		if(iTotalBytesRead != iReadBytes)
			{
				printf("Read iTotalBytesRead=%d\n",iTotalBytesRead);
				continue;
			}

		}else{

			printf("Read END:iTotalBytesRead=%d\n",iTotalBytesRead);
			break;
		}
	}