title: How can I fix "Internal data stream error" generated by the Gstreamer appsrc element?
body: <p>I am trying to use appsrc element of Gstreamer on a trivial example. I am creating a buffer, filling it with dummy data and trying to send it to a fakesink. The code is a watered down version of the tutorial given in link below. It only has two elements, appsrc and fakesink. My code is also given below.</p>
  <p>When I run this code I get &quot;Error: Internal data stream error.&quot; I have searched for a solution and as far as I can tell, the issue is attributed to mismatch of caps between elements. I don't think this is the issue here since I only have two elements.</p>
  <p>I have tried setting &quot;caps&quot; property of appsrc to NULL here, but I have also tried passing the proper &quot;caps&quot; property which was &quot;audio/G729.&quot; Both have failed. Also it seems like the appsrc is fine for the first 4 chunks but then it generates an error. It is always after 4 Chunks. Not sure if that is a clue.</p>
  <p>Also, I am running the code on an embedded system (ARM Cortex-A15) but I don't think that is related. I can succesfully stream a G729 encoded Audio file on this system via following command:</p>
  <pre><code>gst-launch-1.0 -v filesrc location=encodedData.g729 ! 'audio/G729' ! rtpg729pay ! udpsink host=192.168.XX.XX auto-multicast=true port=5004
  </code></pre>
  <p>What could be the reason behind this error? How can I fix this?</p>
  <p>Thanks for all the responses.</p>
  <p>Link: <a href="https://gstreamer.freedesktop.org/documentation/tutorials/basic/short-cutting-the-pipeline.html?gi-language=python" rel="nofollow noreferrer">Link to Tutorial</a></p>
  <p>Code:</p>
  <pre><code>#include &lt;gst/gst.h&gt;
  #include &lt;glib.h&gt;
  #include &quot;glibconfig.h&quot;
  #include &lt;stdio.h&gt;
  
  #define     CHUNK_SIZE          10
  
  typedef struct gstreamstruct {
      GstElement  *pipeline, *app_source, *fakesink;
      guint       sourceid;        /* To control the GSource */
      GMainLoop   *main_loop;  /* GLib's Main Loop */
      guint       sample;
  } gstreamstruct;
  
  static gboolean busCall (GstBus *bus, GstMessage *msg, gpointer data)
  {
      GMainLoop *loop = (GMainLoop *) data;
  
      switch (GST_MESSAGE_TYPE (msg)) {
  
      case GST_MESSAGE_EOS:
          g_print (&quot;End of stream\n&quot;);
          g_main_loop_quit (loop);
          break;
  
      case GST_MESSAGE_ERROR: {
          gchar  *debug;
          GError *error;
  
          gst_message_parse_error (msg, &amp;error, &amp;debug);
          g_free (debug);
  
          g_printerr (&quot;Error: %s\n&quot;, error-&gt;message);
          g_error_free (error);
  
          g_main_loop_quit (loop);
          break;
      }
      default:
          break;
      }
  
      return TRUE;
  }
  
  static gboolean pushData (gstreamstruct *streamer)
  {
      printf(&quot;---&gt;PushData!\n&quot;);
      GstMapInfo map;
      GstBuffer *buffer;
      GstFlowReturn ret;
      guint8 *raw;
      int i;
  
      /* Create a new empty buffer */
      buffer = gst_buffer_new_and_alloc (CHUNK_SIZE);
  
      /* Set its timestamp and duration */
      GST_BUFFER_TIMESTAMP (buffer) = gst_util_uint64_scale (streamer-&gt;sample, GST_SECOND, 1000);
      GST_BUFFER_DURATION (buffer) = gst_util_uint64_scale (CHUNK_SIZE, GST_SECOND, 1000);
  
     //Put some dummy into buffer
      gst_buffer_map (buffer, &amp;map, GST_MAP_WRITE);
      raw = (guint8 *)map.data;
      for(i = 0; i&lt;CHUNK_SIZE; i++)
      {
          raw[0] = 0;
      }
  
      //update sample value
      streamer-&gt;sample += CHUNK_SIZE;
      printf(&quot;currentDuration: %d ms\n&quot;, streamer-&gt;sample);
      gst_buffer_unmap (buffer, &amp;map);
  
      /* Push the buffer into the appsrc */
      g_signal_emit_by_name (streamer-&gt;app_source, &quot;push-buffer&quot;, buffer, &amp;ret);
  
      /* Free the buffer now that we are done with it */
      gst_buffer_unref (buffer);
  
      if (ret != GST_FLOW_OK)
      {
      /* We got some error, stop sending data */
          printf(&quot;Data sending Failed!\n&quot;);
          return FALSE;
      }
  
      return TRUE;
  }
  
  /* This signal callback triggers when appsrc needs data.
   * Here, we add an idle handler to the mainloop to start pushing data into the appsrc
   *
   * Whenever Gstreamer goes idle, it will call this function. Maybe we can utilize this for
   * G729 etc!
   *
   * */
  static void startFeed (GstElement *source, guint size, gstreamstruct *streamer)
  {
      if (streamer-&gt;sourceid == 0)
      {
          g_print (&quot;Start feeding\n&quot;);
          streamer-&gt;sourceid = g_idle_add ((GSourceFunc) pushData, streamer);
      }
  }
  
  /* This callback triggers when appsrc has enough data and we can stop sending.
   * We remove the idle handler from the mainloop */
  static void stopFeed (GstElement *source, gstreamstruct *streamer)
  {
      if (streamer-&gt;sourceid != 0)
      {
          g_print (&quot;Stop feeding\n&quot;);
          g_source_remove (streamer-&gt;sourceid);
          streamer-&gt;sourceid = 0;
      }
  }
  
  void appSrcTest (void* args)
  {
      printf(&quot;---&gt; appSrcTest\n&quot;);
      gstreamstruct my_streamer;
      GstCaps *caps;
      GstBus *bus;
      //GstPad *pad;
      guint bus_watch_id;
  
      memset (&amp;my_streamer, 0, sizeof (gstreamstruct));
  
      gst_init (NULL, NULL);
      my_streamer.main_loop = g_main_loop_new (NULL, FALSE);
  
      printf(&quot;Gst Initialized!\n&quot;);
      my_streamer.sample = 0;
  
      my_streamer.app_source  = gst_element_factory_make(&quot;appsrc&quot;, &quot;appSrc&quot;);
      my_streamer.fakesink     = gst_element_factory_make(&quot;fakesink&quot;, &quot;fakeSink&quot;);
  
      my_streamer.pipeline = gst_pipeline_new (&quot;g729-pipeline&quot;);
  
      if(!my_streamer.app_source || !my_streamer.fakesink || !my_streamer.pipeline)
      {
          g_printerr (&quot;Not all elements could be created.\n&quot;);
          return;
      }
      printf(&quot;Elements Created!\n&quot;);
      caps=NULL;
      /*
      caps = gst_caps_new_simple (&quot;audio/G729&quot;,
                              &quot;channels&quot;, G_TYPE_INT, 1,
                              &quot;rate&quot;, G_TYPE_INT, 8000,
                              NULL);
      */
      //g_object_set (G_OBJECT(my_streamer.app_source), &quot;caps&quot;, caps, &quot;format&quot;, GST_FORMAT_TIME,     NULL);
      g_signal_connect (my_streamer.app_source, &quot;need-data&quot;, G_CALLBACK (startFeed), &amp;my_streamer);
      g_signal_connect (my_streamer.app_source, &quot;enough-data&quot;, G_CALLBACK (stopFeed), &amp;my_streamer);
  
      printf(&quot;Properties Set!\n&quot;);
  
      /* we add a message handler */
      bus = gst_pipeline_get_bus (GST_PIPELINE (my_streamer.pipeline));
      bus_watch_id = gst_bus_add_watch (bus, busCall, my_streamer.main_loop);
      gst_object_unref (bus);
  
      gst_bin_add_many (GST_BIN (my_streamer.pipeline), my_streamer.app_source,     my_streamer.fakesink, NULL);
      printf(&quot;Elements Added!\n&quot;);
  
  
      printf(&quot;Pipeline Starting!\n&quot;);
      gst_element_set_state (my_streamer.pipeline, GST_STATE_PLAYING);
  
      g_main_loop_run (my_streamer.main_loop);
  
      gst_element_set_state (my_streamer.pipeline, GST_STATE_NULL);
      gst_object_unref (my_streamer.pipeline);
      g_source_remove (bus_watch_id);
      g_main_loop_unref (my_streamer.main_loop);
  }
  </code></pre>
  <p>The output from this code is generated as:</p>
  <pre><code>Gst Initialized!
  Elements Created!
  Properties Set!
  Elements Added!
  Pipeline Starting!
  Start feeding
  ---&gt;PushData!
  currentDuration: 10 ms
  ---&gt;PushData!
  currentDuration: 20 ms
  ---&gt;PushData!
  currentDuration: 30 ms
  ---&gt;PushData!
  currentDuration: 40 ms
  Error: Internal data stream error.
  </code></pre>
  <p>Edit: After more trials I have realized that the error is not generated after 4 Chunks consistently. When I reboot the system and call the function the error is generated after 156 Chunks for instance. After a couple more tries, the error starts occur much sooner (like 4 Chunks.) Also I have tried running the code with GST_DEBUG=2 but could not really find anything useful. Below you can find the DEBUG output.
  DEBUG:</p>
  <pre><code>---&gt; appSrcTest
  Gst Initialized!
  Elements Created!
  Properties Set!
  Elements Added!
  Pipeline Starting!
  Start feeding
  ---&gt;PushData!
  currentDuration: 10 ms
  ---&gt;PushData!
  currentDuration: 20 ms
  ---&gt;PushData!
  currentDuration: 30 ms
  ---&gt;PushData!
  currentDuration: 40 ms
  ---&gt;PushData!
  0:00:00.084280528  1344   0x18fa00 WARN                 basesrc gstbasesrc.c:3055:gst_base_src_loop:&lt;appSrc&gt; error: Internal data stream error.
  currentDuration: 50 ms
  ---&gt;PushData!
  0:00:00.084342504  1344   0x18fa00 WARN                 basesrc gstbasesrc.c:3055:gst_base_src_loop:&lt;appSrc&gt; error: streaming stopped, reason not-linked (-1)
  currentDuration: 60 ms
  ---&gt;PushData!
  currentDuration: 70 ms
  Error: Internal data stream error.
  </code></pre>
  <p>Edit 2: After further debugging I have realized that the fakesink element was not linked to appsrc. So I manually linked them via the following line</p>
  <pre><code>gst_element_link_pads (my_streamer.app_source, &quot;src&quot;, my_streamer.fakesink, &quot;sink&quot;);
  </code></pre>
  <p>I think it works fine now, I will come back again after I verify it completely.</p>
  <p>Edit 3: Yeah, I can confirm that was the issue. I forgot to link the elements.</p>