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/app/gstappsrc.h>
  34 #include <gst/audio/gstaudiobasesrc.h>
  35 #include <glib.h>
  36
  37 #include <opus/opus.h>
  38 #include <opus/opus_types.h>
  39
  40 /**
  41  * How much data to read in one go
  42  */
  43 #define MAXLINE 4096
  44
  45 #define SAMPLING_RATE 48000
  46
  47 #define CHANNELS 1
  48
  49 #define FRAME_SIZE (SAMPLING_RATE / 50)
  50
  51 #define PCM_LENGTH (FRAME_SIZE * CHANNELS * sizeof (int16_t))
  52
  53 /**
  54  * Max number of microseconds to buffer in audiosink.
  55  * Default is 200000
  56  */
  57 #define BUFFER_TIME 1000
  58
  59 /**
  60  * Min number of microseconds to buffer in audiosink.
  61  * Default is 10000
  62  */
  63 #define LATENCY_TIME 1000
  64
  65 /**
  66  * Tokenizer for the data we get from stdin
  67  */
  68 struct GNUNET_SERVER_MessageStreamTokenizer *stdin_mst;
  69
  70 /**
  71  * Main pipeline.
  72  */
  73 static GstElement *pipeline;
  74
  75 /**
  76  * Appsrc instance into which we write data for the pipeline.
  77  */
  78 static GstElement *source;
  79
  80 /**
  81  * OPUS decoder
  82  */
  83 static OpusDecoder *dec;
  84
  85
  86 /**
  87  * Set to 1 to break the reading loop
  88  */
  89 static int abort_read;
  90
  91
  92 /**
  93  * OPUS initialization
  94  */
  95 static void
  96 opus_init ()
  97 {
  98   int err;
  99   int channels = 1;
 100
 101   dec = opus_decoder_create (SAMPLING_RATE, channels, &err);
 102 }
 103
 104 void
 105 sink_child_added (GstChildProxy *child_proxy, GObject *object, gchar *name, gpointer user_data)
 106 {
 107   if (GST_IS_AUDIO_BASE_SRC (object))
 108     g_object_set (object, "buffer-time", (gint64) BUFFER_TIME, "latency-time", (gint64) LATENCY_TIME, NULL);
 109 }
 110
 111 static void
 112 quit ()
 113 {
 114   if (NULL != source)
 115     gst_app_src_end_of_stream (GST_APP_SRC (source));
 116   if (NULL != pipeline)
 117     gst_element_set_state (pipeline, GST_STATE_NULL);
 118   abort_read = 1;
 119 }
 120
 121 static gboolean
 122 bus_call (GstBus *bus, GstMessage *msg, gpointer data)
 123 {
 124   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Bus message\n");
 125   switch (GST_MESSAGE_TYPE (msg))
 126   {
 127   case GST_MESSAGE_EOS:
 128     GNUNET_log (GNUNET_ERROR_TYPE_INFO, "End of stream\n");
 129     quit ();
 130     break;
 131
 132   case GST_MESSAGE_ERROR:
 133     {
 134       gchar  *debug;
 135       GError *error;
 136
 137       gst_message_parse_error (msg, &error, &debug);
 138       g_free (debug);
 139
 140       GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "Error: %s\n", error->message);
 141       g_error_free (error);
 142
 143       quit ();
 144       break;
 145     }
 146   default:
 147     break;
 148   }
 149
 150   return TRUE;
 151 }
 152
 153
 154 static void
 155 signalhandler (int s)
 156 {
 157   quit ();
 158 }
 159
 160
 161 /**
 162  * Message callback
 163  */
 164 static int
 165 stdin_receiver (void *cls,
 166                 void *client,
 167                 const struct GNUNET_MessageHeader *msg)
 168 {
 169   struct AudioMessage *audio;
 170   GstBuffer *b;
 171   int16_t *bufspace;
 172   GstFlowReturn flow;
 173   int ret;
 174
 175   switch (ntohs (msg->type))
 176   {
 177   case GNUNET_MESSAGE_TYPE_CONVERSATION_AUDIO:
 178     audio = (struct AudioMessage *) msg;
 179
 180     bufspace = (int16_t *) g_malloc (PCM_LENGTH);
 181
 182     ret = opus_decode (dec,
 183                        (const unsigned char *) &audio[1],
 184                        ntohs (audio->header.size) - sizeof (struct AudioMessage),
 185                        bufspace,
 186                        FRAME_SIZE, 0);
 187     if (ret < 0)
 188     {
 189       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 190                   "Opus decoding failed: %d\n",
 191                   ret);
 192       g_free (bufspace);
 193       return GNUNET_OK;
 194     }
 195     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 196                 "Decoded frame with %u bytes\n",
 197                 ntohs (audio->header.size));
 198
 199     b = gst_buffer_new_wrapped (bufspace, ret * sizeof (int16_t));
 200     if (NULL == b)
 201     {
 202       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Failed to wrap a buffer\n");
 203       g_free (bufspace);
 204       return GNUNET_SYSERR;
 205     }
 206
 207     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "pushing...\n");
 208     flow = gst_app_src_push_buffer (GST_APP_SRC (source), b);
 209     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "pushed!\n");
 210     /* They all return GNUNET_OK, because currently player stops when
 211      * data stops coming. This might need to be changed for the player
 212      * to also stop when pipeline breaks.
 213      */
 214     switch (flow)
 215     {
 216     case GST_FLOW_OK:
 217       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Fed %u bytes to the pipeline\n",
 218           (unsigned int) ret * sizeof (int16_t));
 219       break;
 220     case GST_FLOW_FLUSHING:
 221       /* buffer was dropped, because pipeline state is not PAUSED or PLAYING */
 222       GNUNET_log (GNUNET_ERROR_TYPE_INFO, "Dropped a buffer\n");
 223       break;
 224     case GST_FLOW_EOS:
 225       /* end of stream */
 226       GNUNET_log (GNUNET_ERROR_TYPE_INFO, "EOS\n");
 227       break;
 228     default:
 229       GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "Unexpected push result\n");
 230       break;
 231     }
 232     break;
 233   default:
 234     break;
 235   }
 236   return GNUNET_OK;
 237 }
 238
 239
 240 int
 241 main (int argc, char **argv)
 242 {
 243   GstElement *conv, *resampler, *sink;
 244   GstBus *bus;
 245   GstCaps *caps;
 246   guint bus_watch_id;
 247   uint64_t toff;
 248
 249   typedef void (*SignalHandlerPointer) (int);
 250
 251   SignalHandlerPointer inthandler, termhandler;
 252
 253   inthandler = signal (SIGINT, signalhandler);
 254   termhandler = signal (SIGTERM, signalhandler);
 255
 256 #ifdef WINDOWS
 257   setmode (0, _O_BINARY);
 258 #endif
 259
 260   opus_init ();
 261
 262   /* Initialisation */
 263   gst_init (&argc, &argv);
 264
 265   GNUNET_assert (GNUNET_OK ==
 266                  GNUNET_log_setup ("gnunet-helper-audio-playback",
 267                                    "WARNING",
 268                                    NULL));
 269
 270   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 271               "Audio sink starts\n");
 272
 273   stdin_mst = GNUNET_SERVER_mst_create (&stdin_receiver, NULL);
 274
 275   /* Create gstreamer elements */
 276   pipeline = gst_pipeline_new ("audio-player");
 277   source   = gst_element_factory_make ("appsrc",        "audio-input");
 278   conv     = gst_element_factory_make ("audioconvert",  "converter");
 279   resampler= gst_element_factory_make ("audioresample", "resampler");
 280   sink     = gst_element_factory_make ("autoaudiosink", "audiosink");
 281
 282   if (!pipeline || !source || !conv || !resampler || !sink)
 283   {
 284     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 285         "One element could not be created. Exiting.\n");
 286     return -1;
 287   }
 288
 289   g_signal_connect (sink, "child-added", G_CALLBACK (sink_child_added), NULL);
 290
 291   caps = gst_caps_new_simple ("audio/x-raw",
 292     "format", G_TYPE_STRING, "S16LE",
 293     "rate", G_TYPE_INT, SAMPLING_RATE,
 294     "channels", G_TYPE_INT, CHANNELS,
 295     "layout", G_TYPE_STRING, "interleaved",
 296      NULL);
 297   gst_app_src_set_caps (GST_APP_SRC (source), caps);
 298   gst_caps_unref (caps);
 299
 300   /* Keep a reference to it, we operate on it */
 301   gst_object_ref (GST_OBJECT (source));
 302
 303   /* Set up the pipeline */
 304
 305   /* we feed appsrc as fast as possible, it just blocks when it's full */
 306   g_object_set (G_OBJECT (source),
 307       "format", GST_FORMAT_TIME,
 308       "block", TRUE,
 309       "is-live", TRUE,
 310       NULL);
 311
 312   /* we add a message handler */
 313   bus = gst_pipeline_get_bus (GST_PIPELINE (pipeline));
 314   bus_watch_id = gst_bus_add_watch (bus, bus_call, pipeline);
 315   gst_object_unref (bus);
 316
 317   /* we add all elements into the pipeline */
 318   /* audio-input | converter | resampler | audiosink */
 319   gst_bin_add_many (GST_BIN (pipeline), source, conv,
 320       resampler, sink, NULL);
 321
 322   /* we link the elements together */
 323   gst_element_link_many (source, conv, resampler, sink, NULL);
 324
 325   /* Set the pipeline to "playing" state*/
 326   GNUNET_log (GNUNET_ERROR_TYPE_INFO, "Now playing\n");
 327   gst_element_set_state (pipeline, GST_STATE_PLAYING);
 328
 329   GNUNET_log (GNUNET_ERROR_TYPE_INFO, "Running...\n");
 330   /* Iterate */
 331   toff = 0;
 332   while (!abort_read)
 333   {
 334     char readbuf[MAXLINE];
 335     int ret;
 336
 337     ret = read (0, readbuf, sizeof (readbuf));
 338     if (0 > ret)
 339     {
 340       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 341                   _("Read error from STDIN: %d %s\n"),
 342                   ret, strerror (errno));
 343       break;
 344     }
 345     toff += ret;
 346     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 347                 "Received %d bytes of audio data (total: %llu)\n",
 348                 (int) ret,
 349                 toff);
 350     if (0 == ret)
 351       break;
 352     GNUNET_SERVER_mst_receive (stdin_mst, NULL,
 353                                readbuf, ret,
 354                                GNUNET_NO, GNUNET_NO);
 355   }
 356   GNUNET_SERVER_mst_destroy (stdin_mst);
 357
 358   signal (SIGINT, inthandler);
 359   signal (SIGINT, termhandler);
 360
 361   GNUNET_log (GNUNET_ERROR_TYPE_INFO, "Returned, stopping playback\n");
 362   quit ();
 363
 364   GNUNET_log (GNUNET_ERROR_TYPE_INFO, "Deleting pipeline\n");
 365   gst_object_unref (GST_OBJECT (source));
 366   source = NULL;
 367   gst_object_unref (GST_OBJECT (pipeline));
 368   pipeline = NULL;
 369   g_source_remove (bus_watch_id);
 370
 371   return 0;
 372 }