GNUnet  0.20.0
gnunet-helper-audio-playback-gst.c
Go to the documentation of this file.
1 /*
2  This file is part of GNUnet.
3  Copyright (C) 2013 GNUnet e.V.
4 
5  GNUnet is free software: you can redistribute it and/or modify it
6  under the terms of the GNU Affero General Public License as published
7  by the Free Software Foundation, either version 3 of the License,
8  or (at your 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  Affero General Public License for more details.
14 
15  You should have received a copy of the GNU Affero General Public License
16  along with this program. If not, see <http://www.gnu.org/licenses/>.
17 
18  SPDX-License-Identifier: AGPL3.0-or-later
19  */
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 
42 #define MAXLINE 4096
43 
48 #define BUFFER_TIME 1000
49 
54 #define LATENCY_TIME 1000
55 
60 
64 static GstElement *pipeline;
65 
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 
80 static int abort_read;
81 
82 
83 static void
84 sink_child_added (GstChildProxy *child_proxy,
85  GObject *object,
86  gchar *name,
87  gpointer user_data)
88 {
89  if (GST_IS_AUDIO_BASE_SRC (object))
90  g_object_set (object,
91  "buffer-time", (gint64) BUFFER_TIME,
92  "latency-time", (gint64) LATENCY_TIME,
93  NULL);
94 }
95 
96 
97 static void
98 ogg_pad_added (GstElement *element,
99  GstPad *pad,
100  gpointer data)
101 {
102  GstPad *sinkpad;
103  GstElement *decoder = (GstElement *) data;
104 
105  /* We can now link this pad with the opus-decoder sink pad */
106  sinkpad = gst_element_get_static_pad (decoder, "sink");
107 
108  gst_pad_link (pad, sinkpad);
109 
110  gst_element_link_many (decoder, conv, resampler, sink, NULL);
111 
112  gst_object_unref (sinkpad);
113 }
114 
115 
116 static void
118 {
119  if (NULL != source)
120  gst_app_src_end_of_stream (GST_APP_SRC (source));
121  if (NULL != pipeline)
122  gst_element_set_state (pipeline, GST_STATE_NULL);
123  abort_read = 1;
124 }
125 
126 
127 static gboolean
128 bus_call (GstBus *bus, GstMessage *msg, gpointer data)
129 {
131  "Bus message\n");
132  switch (GST_MESSAGE_TYPE (msg))
133  {
134  case GST_MESSAGE_EOS:
136  "End of stream\n");
137  quit ();
138  break;
139 
140  case GST_MESSAGE_ERROR:
141  {
142  gchar *debug;
143  GError *error;
144 
145  gst_message_parse_error (msg, &error, &debug);
146  g_free (debug);
147 
149  "Error: %s\n",
150  error->message);
151  g_error_free (error);
152 
153  quit ();
154  break;
155  }
156 
157  default:
158  break;
159  }
160 
161  return TRUE;
162 }
163 
164 
165 static void
167 {
168  quit ();
169 }
170 
171 
172 static int
173 feed_buffer_to_gst (const char *audio, size_t b_len)
174 {
175  GstBuffer *b;
176  gchar *bufspace;
177  GstFlowReturn flow;
178 
180  "Feeding %u bytes to GStreamer\n",
181  (unsigned int) b_len);
182 
183  bufspace = g_memdup2 (audio, b_len);
184  b = gst_buffer_new_wrapped (bufspace, b_len);
185  if (NULL == b)
186  {
188  "Failed to wrap a buffer\n");
189  g_free (bufspace);
190  return GNUNET_SYSERR;
191  }
192  flow = gst_app_src_push_buffer (GST_APP_SRC (source), b);
193  /* They all return GNUNET_OK, because currently player stops when
194  * data stops coming. This might need to be changed for the player
195  * to also stop when pipeline breaks.
196  */
197  switch (flow)
198  {
199  case GST_FLOW_OK:
201  "Fed %u bytes to the pipeline\n",
202  (unsigned int) b_len);
203  break;
204 
205  case GST_FLOW_FLUSHING:
206  /* buffer was dropped, because pipeline state is not PAUSED or PLAYING */
208  "Dropped a buffer\n");
209  break;
210 
211  case GST_FLOW_EOS:
212  /* end of stream */
214  "EOS\n");
215  break;
216 
217  default:
219  "Unexpected push result\n");
220  break;
221  }
222  return GNUNET_OK;
223 }
224 
225 
234 static int
235 stdin_receiver (void *cls,
236  const struct GNUNET_MessageHeader *msg)
237 {
238  struct AudioMessage *audio;
239  size_t b_len;
240 
241  switch (ntohs (msg->type))
242  {
244  audio = (struct AudioMessage *) msg;
245 
246  b_len = ntohs (audio->header.size) - sizeof(struct AudioMessage);
247  feed_buffer_to_gst ((const char *) &audio[1], b_len);
248  break;
249 
250  default:
251  break;
252  }
253  return GNUNET_OK;
254 }
255 
256 
257 int
258 main (int argc, char **argv)
259 {
260  GstBus *bus;
261  guint bus_watch_id;
262  uint64_t toff;
263 
264  typedef void (*SignalHandlerPointer) (int);
265 
266  SignalHandlerPointer inthandler, termhandler;
267 #ifdef DEBUG_READ_PURE_OGG
268  int read_pure_ogg = getenv ("GNUNET_READ_PURE_OGG") ? 1 : 0;
269 #endif
270 
271  inthandler = signal (SIGINT,
272  &signalhandler);
273  termhandler = signal (SIGTERM,
274  &signalhandler);
275 
276  /* Initialisation */
277  gst_init (&argc, &argv);
278 
280  GNUNET_log_setup ("gnunet-helper-audio-playback-gst",
281  "WARNING",
282  NULL));
283 
285  "Audio sink starts\n");
286 
288  NULL);
289 
290  /* Create gstreamer elements */
291  pipeline = gst_pipeline_new ("audio-player");
292  source = gst_element_factory_make ("appsrc", "audio-input");
293  demuxer = gst_element_factory_make ("oggdemux", "ogg-demuxer");
294  decoder = gst_element_factory_make ("opusdec", "opus-decoder");
295  conv = gst_element_factory_make ("audioconvert", "converter");
296  resampler = gst_element_factory_make ("audioresample", "resampler");
297  sink = gst_element_factory_make ("autoaudiosink", "audiosink");
298 
299  if (! pipeline || ! source || ! conv || ! resampler || ! decoder ||
300  ! demuxer || ! sink)
301  {
303  "One element could not be created. Exiting.\n");
304  return -1;
305  }
306 
307  g_signal_connect (sink,
308  "child-added",
309  G_CALLBACK (sink_child_added),
310  NULL);
311  g_signal_connect (demuxer,
312  "pad-added",
313  G_CALLBACK (ogg_pad_added),
314  decoder);
315 
316  /* Keep a reference to it, we operate on it */
317  gst_object_ref (GST_OBJECT (source));
318 
319  /* Set up the pipeline */
320 
321  /* we feed appsrc as fast as possible, it just blocks when it's full */
322  g_object_set (G_OBJECT (source),
323 /* "format", GST_FORMAT_TIME,*/
324  "block", TRUE,
325  "is-live", TRUE,
326  NULL);
327 
328  g_object_set (G_OBJECT (decoder),
329 /* "plc", FALSE,*/
330 /* "apply-gain", TRUE,*/
331  "use-inband-fec", TRUE,
332  NULL);
333 
334  /* we add a message handler */
335  bus = gst_pipeline_get_bus (GST_PIPELINE (pipeline));
336  bus_watch_id = gst_bus_add_watch (bus, bus_call, pipeline);
337  gst_object_unref (bus);
338 
339  /* we add all elements into the pipeline */
340  /* audio-input | ogg-demuxer | opus-decoder | converter | resampler | audiosink */
341  gst_bin_add_many (GST_BIN (pipeline), source, demuxer, decoder, conv,
342  resampler, sink, NULL);
343 
344  /* we link the elements together */
345  gst_element_link_many (source, demuxer, NULL);
346 
347  /* Set the pipeline to "playing" state*/
348  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "Now playing\n");
349  gst_element_set_state (pipeline, GST_STATE_PLAYING);
350 
351  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "Running...\n");
352  /* Iterate */
353  toff = 0;
354  while (! abort_read)
355  {
356  char readbuf[MAXLINE];
357  int ret;
358 
359  ret = read (0, readbuf, sizeof(readbuf));
360  if (0 > ret)
361  {
363  _ ("Read error from STDIN: %d %s\n"),
364  ret, strerror (errno));
365  break;
366  }
367  toff += ret;
369  "Received %d bytes of audio data (total: %llu)\n",
370  (int) ret,
371  (unsigned long long) toff);
372  if (0 == ret)
373  break;
374 #ifdef DEBUG_READ_PURE_OGG
375  if (read_pure_ogg)
376  {
377  feed_buffer_to_gst (readbuf, ret);
378  }
379  else
380 #endif
382  readbuf,
383  ret,
384  GNUNET_NO,
385  GNUNET_NO);
386  }
388 
389  signal (SIGINT, inthandler);
390  signal (SIGINT, termhandler);
391 
393  "Returned, stopping playback\n");
394  quit ();
395 
397  "Deleting pipeline\n");
398  gst_object_unref (GST_OBJECT (source));
399  source = NULL;
400  gst_object_unref (GST_OBJECT (pipeline));
401  pipeline = NULL;
402  g_source_remove (bus_watch_id);
403 
404  return 0;
405 }
struct GNUNET_MessageHeader * msg
Definition: 005.c:2
constants for network protocols
char * getenv()
static int ret
Return value of the commandline.
Definition: gnunet-abd.c:81
static void signalhandler(int s)
static void sink_child_added(GstChildProxy *child_proxy, GObject *object, gchar *name, gpointer user_data)
static int abort_read
Set to 1 to break the reading loop.
static GstElement * source
Appsrc instance into which we write data for the pipeline.
#define LATENCY_TIME
Min number of microseconds to buffer in audiosink.
static GstElement * pipeline
Main pipeline.
int main(int argc, char **argv)
#define MAXLINE
How much data to read in one go.
static void ogg_pad_added(GstElement *element, GstPad *pad, gpointer data)
static GstElement * decoder
static int stdin_receiver(void *cls, const struct GNUNET_MessageHeader *msg)
Message callback.
static GstElement * resampler
static GstElement * sink
struct GNUNET_MessageStreamTokenizer * stdin_mst
Tokenizer for the data we get from stdin.
static int feed_buffer_to_gst(const char *audio, size_t b_len)
static void quit()
static GstElement * demuxer
static gboolean bus_call(GstBus *bus, GstMessage *msg, gpointer data)
static GstElement * conv
#define BUFFER_TIME
Max number of microseconds to buffer in audiosink.
uint32_t data
The data value.
Core service; the main API for encrypted P2P communications.
#define gst_element_factory_make(element, name)
Definition: gnunet_gst.h:36
Constants for network protocols.
#define GNUNET_log(kind,...)
@ GNUNET_OK
@ GNUNET_NO
@ GNUNET_SYSERR
int GNUNET_log_setup(const char *comp, const char *loglevel, const char *logfile)
Setup logging.
#define GNUNET_assert(cond)
Use this for fatal errors that cannot be handled.
@ GNUNET_ERROR_TYPE_WARNING
@ GNUNET_ERROR_TYPE_ERROR
@ GNUNET_ERROR_TYPE_DEBUG
@ GNUNET_ERROR_TYPE_INFO
#define GNUNET_MESSAGE_TYPE_CONVERSATION_AUDIO
Message to transmit the audio between helper and speaker/microphone library.
enum GNUNET_GenericReturnValue GNUNET_MST_from_buffer(struct GNUNET_MessageStreamTokenizer *mst, const char *buf, size_t size, int purge, int one_shot)
Add incoming data to the receive buffer and call the callback for all complete messages.
Definition: mst.c:101
void GNUNET_MST_destroy(struct GNUNET_MessageStreamTokenizer *mst)
Destroys a tokenizer.
Definition: mst.c:404
struct GNUNET_MessageStreamTokenizer * GNUNET_MST_create(GNUNET_MessageTokenizerCallback cb, void *cb_cls)
Create a message stream tokenizer.
Definition: mst.c:86
#define _(String)
GNU gettext support macro.
Definition: platform.h:178
const char * name
Message to transmit the audio (between client and helpers).
Definition: conversation.h:59
struct GNUNET_MessageHeader header
Type is GNUNET_MESSAGE_TYPE_CONVERSATION_AUDIO.
Definition: conversation.h:63
Header for all communications.
uint16_t type
The type of the message (GNUNET_MESSAGE_TYPE_XXXX), in big-endian format.
uint16_t size
The length of the struct (in bytes, including the length field itself), in big-endian format.
Handle to a message stream tokenizer.
Definition: mst.c:45