src/conversation/gnunet-helper-audio-playback-gst.c

   1 /*
   2      This file is part of GNUnet.
   3      (C) 2013 Christian Grothoff (and other contributing authors)
   4
   5      GNUnet is free software; you can redistribute it and/or modify
   6      it under the terms of the GNU General Public License as published
   7      by the Free Software Foundation; either version 3, or (at your
   8      option) any later version.
   9
  10      GNUnet is distributed in the hope that it will be useful, but
  11      WITHOUT ANY WARRANTY; without even the implied warranty of
  12      MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13      General Public License for more details.
  14
  15      You should have received a copy of the GNU General Public License
  16      along with GNUnet; see the file COPYING.  If not, write to the
  17      Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  18      Boston, MA 02111-1307, USA.
  19 */
  20 /**
  21  * @file conversation/gnunet-helper-audio-playback-gst.c
  22  * @brief program to playback audio data to the speaker (GStreamer version)
  23  * @author LRN
  24  */
  25 #include "platform.h"
  26 #include "gnunet_util_lib.h"
  27 #include "gnunet_protocols.h"
  28 #include "conversation.h"
  29 #include "gnunet_constants.h"
  30 #include "gnunet_core_service.h"
  31
  32 #include <gst/gst.h>
  33 #include <gst/audio/gstaudiobasesrc.h>
  34 #include <gst/app/gstappsrc.h>
  35 #include <glib.h>
  36
  37 #define DEBUG_READ_PURE_OGG 1
  38
  39 /**
  40  * How much data to read in one go
  41  */
  42 #define MAXLINE 4096
  43
  44 /**
  45  * Max number of microseconds to buffer in audiosink.
  46  * Default is 200000
  47  */
  48 #define BUFFER_TIME 1000
  49
  50 /**
  51  * Min number of microseconds to buffer in audiosink.
  52  * Default is 10000
  53  */
  54 #define LATENCY_TIME 1000
  55
  56 /**
  57  * Tokenizer for the data we get from stdin
  58  */
  59 struct GNUNET_SERVER_MessageStreamTokenizer *stdin_mst;
  60
  61 /**
  62  * Main pipeline.
  63  */
  64 static GstElement *pipeline;
  65
  66 /**
  67  * Appsrc instance into which we write data for the pipeline.
  68  */
  69 static GstElement *source;
  70
  71 static GstElement *demuxer;
  72 static GstElement *decoder;
  73 static GstElement *conv;
  74 static GstElement *resampler;
  75 static GstElement *sink;
  76
  77 /**
  78  * Set to 1 to break the reading loop
  79  */
  80 static int abort_read;
  81
  82 static void
  83 sink_child_added (GstChildProxy *child_proxy, GObject *object, gchar *name, gpointer user_data)
  84 {
  85   if (GST_IS_AUDIO_BASE_SRC (object))
  86     g_object_set (object, "buffer-time", (gint64) BUFFER_TIME, "latency-time", (gint64) LATENCY_TIME, NULL);
  87 }
  88
  89 static void
  90 ogg_pad_added (GstElement *element, GstPad *pad, gpointer data)
  91 {
  92   GstPad *sinkpad;
  93   GstElement *decoder = (GstElement *) data;
  94
  95   /* We can now link this pad with the opus-decoder sink pad */
  96   sinkpad = gst_element_get_static_pad (decoder, "sink");
  97
  98   gst_pad_link (pad, sinkpad);
  99
 100   gst_element_link_many (decoder, conv, resampler, sink, NULL);
 101
 102   gst_object_unref (sinkpad);
 103 }
 104
 105 static void
 106 quit ()
 107 {
 108   if (NULL != source)
 109     gst_app_src_end_of_stream (GST_APP_SRC (source));
 110   if (NULL != pipeline)
 111     gst_element_set_state (pipeline, GST_STATE_NULL);
 112   abort_read = 1;
 113 }
 114
 115 static gboolean
 116 bus_call (GstBus *bus, GstMessage *msg, gpointer data)
 117 {
 118   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Bus message\n");
 119   switch (GST_MESSAGE_TYPE (msg))
 120   {
 121   case GST_MESSAGE_EOS:
 122     GNUNET_log (GNUNET_ERROR_TYPE_INFO, "End of stream\n");
 123     quit ();
 124     break;
 125
 126   case GST_MESSAGE_ERROR:
 127     {
 128       gchar  *debug;
 129       GError *error;
 130
 131       gst_message_parse_error (msg, &error, &debug);
 132       g_free (debug);
 133
 134       GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "Error: %s\n", error->message);
 135       g_error_free (error);
 136
 137       quit ();
 138       break;
 139     }
 140   default:
 141     break;
 142   }
 143
 144   return TRUE;
 145 }
 146
 147
 148 static void
 149 signalhandler (int s)
 150 {
 151   quit ();
 152 }
 153
 154 static int
 155 feed_buffer_to_gst (const char *audio, size_t b_len)
 156 {
 157   GstBuffer *b;
 158   gchar *bufspace;
 159   GstFlowReturn flow;
 160
 161   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 162       "Feeding %u bytes to GStreamer\n",
 163       (unsigned int) b_len);
 164
 165   bufspace = g_memdup (audio, b_len);
 166   b = gst_buffer_new_wrapped (bufspace, b_len);
 167   if (NULL == b)
 168   {
 169     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Failed to wrap a buffer\n");
 170     g_free (bufspace);
 171     return GNUNET_SYSERR;
 172   }
 173   flow = gst_app_src_push_buffer (GST_APP_SRC (source), b);
 174   /* They all return GNUNET_OK, because currently player stops when
 175    * data stops coming. This might need to be changed for the player
 176    * to also stop when pipeline breaks.
 177    */
 178   switch (flow)
 179   {
 180   case GST_FLOW_OK:
 181     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Fed %u bytes to the pipeline\n",
 182         (unsigned int) b_len);
 183     break;
 184   case GST_FLOW_FLUSHING:
 185     /* buffer was dropped, because pipeline state is not PAUSED or PLAYING */
 186     GNUNET_log (GNUNET_ERROR_TYPE_INFO, "Dropped a buffer\n");
 187     break;
 188   case GST_FLOW_EOS:
 189     /* end of stream */
 190     GNUNET_log (GNUNET_ERROR_TYPE_INFO, "EOS\n");
 191     break;
 192   default:
 193     GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "Unexpected push result\n");
 194     break;
 195   }
 196   return GNUNET_OK;
 197 }
 198
 199 /**
 200  * Message callback
 201  */
 202 static int
 203 stdin_receiver (void *cls,
 204                 void *client,
 205                 const struct GNUNET_MessageHeader *msg)
 206 {
 207   struct AudioMessage *audio;
 208   size_t b_len;
 209
 210   switch (ntohs (msg->type))
 211   {
 212   case GNUNET_MESSAGE_TYPE_CONVERSATION_AUDIO:
 213     audio = (struct AudioMessage *) msg;
 214
 215     b_len = ntohs (audio->header.size) - sizeof (struct AudioMessage);
 216     feed_buffer_to_gst ((const char *) &audio[1], b_len);
 217     break;
 218   default:
 219     break;
 220   }
 221   return GNUNET_OK;
 222 }
 223
 224
 225 int
 226 main (int argc, char **argv)
 227 {
 228   GstBus *bus;
 229   guint bus_watch_id;
 230   uint64_t toff;
 231
 232   typedef void (*SignalHandlerPointer) (int);
 233
 234   SignalHandlerPointer inthandler, termhandler;
 235 #ifdef DEBUG_READ_PURE_OGG
 236   int read_pure_ogg = getenv ("GNUNET_READ_PURE_OGG") ? 1 : 0;
 237 #endif
 238
 239   inthandler = signal (SIGINT, signalhandler);
 240   termhandler = signal (SIGTERM, signalhandler);
 241
 242 #ifdef WINDOWS
 243   setmode (0, _O_BINARY);
 244 #endif
 245
 246   /* Initialisation */
 247   gst_init (&argc, &argv);
 248
 249   GNUNET_assert (GNUNET_OK ==
 250                  GNUNET_log_setup ("gnunet-helper-audio-playback",
 251                                    "WARNING",
 252                                    NULL));
 253
 254   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 255               "Audio sink starts\n");
 256
 257   stdin_mst = GNUNET_SERVER_mst_create (&stdin_receiver, NULL);
 258
 259   /* Create gstreamer elements */
 260   pipeline = gst_pipeline_new ("audio-player");
 261   source   = gst_element_factory_make ("appsrc",        "audio-input");
 262   demuxer  = gst_element_factory_make ("oggdemux",      "ogg-demuxer");
 263   decoder  = gst_element_factory_make ("opusdec",       "opus-decoder");
 264   conv     = gst_element_factory_make ("audioconvert",  "converter");
 265   resampler= gst_element_factory_make ("audioresample", "resampler");
 266   sink     = gst_element_factory_make ("autoaudiosink", "audiosink");
 267
 268   if (!pipeline || !source || !conv || !resampler || !decoder || !demuxer || !sink)
 269   {
 270     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 271         "One element could not be created. Exiting.\n");
 272     return -1;
 273   }
 274
 275   g_signal_connect (sink, "child-added", G_CALLBACK (sink_child_added), NULL);
 276   g_signal_connect (demuxer, "pad-added", G_CALLBACK (ogg_pad_added), decoder);
 277
 278   /* Keep a reference to it, we operate on it */
 279   gst_object_ref (GST_OBJECT (source));
 280
 281   /* Set up the pipeline */
 282
 283   /* we feed appsrc as fast as possible, it just blocks when it's full */
 284   g_object_set (G_OBJECT (source),
 285 /*      "format", GST_FORMAT_TIME,*/
 286       "block", TRUE,
 287       "is-live", TRUE,
 288       NULL);
 289
 290   g_object_set (G_OBJECT (decoder),
 291 /*      "plc", FALSE,*/
 292 /*      "apply-gain", TRUE,*/
 293       "use-inband-fec", TRUE,
 294       NULL);
 295
 296   /* we add a message handler */
 297   bus = gst_pipeline_get_bus (GST_PIPELINE (pipeline));
 298   bus_watch_id = gst_bus_add_watch (bus, bus_call, pipeline);
 299   gst_object_unref (bus);
 300
 301   /* we add all elements into the pipeline */
 302   /* audio-input | ogg-demuxer | opus-decoder | converter | resampler | audiosink */
 303   gst_bin_add_many (GST_BIN (pipeline), source, demuxer, decoder, conv,
 304       resampler, sink, NULL);
 305
 306   /* we link the elements together */
 307   gst_element_link_many (source, demuxer, NULL);
 308
 309   /* Set the pipeline to "playing" state*/
 310   GNUNET_log (GNUNET_ERROR_TYPE_INFO, "Now playing\n");
 311   gst_element_set_state (pipeline, GST_STATE_PLAYING);
 312
 313   GNUNET_log (GNUNET_ERROR_TYPE_INFO, "Running...\n");
 314   /* Iterate */
 315   toff = 0;
 316   while (!abort_read)
 317   {
 318     char readbuf[MAXLINE];
 319     int ret;
 320
 321     ret = read (0, readbuf, sizeof (readbuf));
 322     if (0 > ret)
 323     {
 324       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 325                   _("Read error from STDIN: %d %s\n"),
 326                   ret, strerror (errno));
 327       break;
 328     }
 329     toff += ret;
 330     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 331                 "Received %d bytes of audio data (total: %llu)\n",
 332                 (int) ret,
 333                 toff);
 334     if (0 == ret)
 335       break;
 336 #ifdef DEBUG_READ_PURE_OGG
 337     if (read_pure_ogg)
 338     {
 339       feed_buffer_to_gst (readbuf, ret);
 340     }
 341     else
 342 #endif
 343     GNUNET_SERVER_mst_receive (stdin_mst, NULL,
 344                                readbuf, ret,
 345                                GNUNET_NO, GNUNET_NO);
 346   }
 347   GNUNET_SERVER_mst_destroy (stdin_mst);
 348
 349   signal (SIGINT, inthandler);
 350   signal (SIGINT, termhandler);
 351
 352   GNUNET_log (GNUNET_ERROR_TYPE_INFO, "Returned, stopping playback\n");
 353   quit ();
 354
 355   GNUNET_log (GNUNET_ERROR_TYPE_INFO, "Deleting pipeline\n");
 356   gst_object_unref (GST_OBJECT (source));
 357   source = NULL;
 358   gst_object_unref (GST_OBJECT (pipeline));
 359   pipeline = NULL;
 360   g_source_remove (bus_watch_id);
 361
 362   return 0;
 363 }