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

   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 */
  20 /**
  21  * @file conversation/gnunet-helper-audio-playback.c
  22  * @brief program to playback audio data to the speaker
  23  * @author Siomon Dieterle
  24  * @author Andreas Fuchs
  25  * @author Christian Grothoff
  26  */
  27 #include "platform.h"
  28 #include "gnunet_util_lib.h"
  29 #include "gnunet_protocols.h"
  30 #include "conversation.h"
  31 #include "gnunet_constants.h"
  32 #include "gnunet_core_service.h"
  33
  34 #include <pulse/simple.h>
  35 #include <pulse/error.h>
  36 #include <pulse/rtclock.h>
  37
  38 #include <pulse/pulseaudio.h>
  39 #include <opus/opus.h>
  40 #include <opus/opus_types.h>
  41 #include <ogg/ogg.h>
  42
  43 #define DEBUG_READ_PURE_OGG 1
  44 #define DEBUG_DUMP_DECODED_OGG 1
  45
  46 #define MAXLINE 4096
  47
  48 #define SAMPLING_RATE 48000
  49
  50 #define CHANNELS 1
  51
  52 /* 120ms at 48000 */
  53 #define MAX_FRAME_SIZE (960 * 6)
  54
  55 /**
  56  * Pulseaudio specification. May change in the future.
  57  */
  58 static pa_sample_spec sample_spec = {
  59   .format = PA_SAMPLE_FLOAT32LE,
  60   .rate = SAMPLING_RATE,
  61   .channels = CHANNELS
  62 };
  63
  64 #ifdef DEBUG_DUMP_DECODED_OGG
  65 static int dump_to_stdout;
  66 #endif
  67
  68 /**
  69  * Pulseaudio mainloop api
  70  */
  71 static pa_mainloop_api *mainloop_api;
  72
  73 /**
  74  * Pulseaudio threaded mainloop
  75  */
  76 static pa_threaded_mainloop *m;
  77
  78 /**
  79  * Pulseaudio context
  80  */
  81 static pa_context *context;
  82
  83 /**
  84  * Pulseaudio output stream
  85  */
  86 static pa_stream *stream_out;
  87
  88 /**
  89  * OPUS decoder
  90  */
  91 static OpusDecoder *dec;
  92
  93 /**
  94  * PCM data buffer
  95  */
  96 static float *pcm_buffer;
  97
  98 /**
  99  * Number of samples for one frame
 100  */
 101 static int frame_size;
 102
 103 /**
 104  * Pipe we use to signal the main loop that we are ready to receive.
 105  */
 106 static int ready_pipe[2];
 107
 108 /**
 109  * Ogg I/O state.
 110  */
 111 static ogg_sync_state oy;
 112
 113 /**
 114  * Ogg stream state.
 115  */
 116 static ogg_stream_state os;
 117
 118 static int channels;
 119
 120 static int preskip;
 121
 122 static float gain;
 123
 124 GNUNET_NETWORK_STRUCT_BEGIN
 125
 126 /* OggOpus spec says the numbers must be in little-endian order */
 127 struct OpusHeadPacket
 128 {
 129   uint8_t magic[8];
 130   uint8_t version;
 131   uint8_t channels;
 132   uint16_t preskip GNUNET_PACKED;
 133   uint32_t sampling_rate GNUNET_PACKED;
 134   uint16_t gain GNUNET_PACKED;
 135   uint8_t channel_mapping;
 136 };
 137
 138 GNUNET_NETWORK_STRUCT_END
 139
 140 /**
 141  * Process an Opus header and setup the opus decoder based on it.
 142  * It takes several pointers for header values which are needed
 143  * elsewhere in the code.
 144  */
 145 static OpusDecoder *
 146 process_header (ogg_packet *op)
 147 {
 148   int err;
 149   OpusDecoder *dec;
 150   struct OpusHeadPacket header;
 151
 152   if ( ((unsigned int) op->bytes) < sizeof (header))
 153     return NULL;
 154   GNUNET_memcpy (&header,
 155                  op->packet,
 156                  sizeof (header));
 157   header.preskip = GNUNET_le16toh (header.preskip);
 158   header.sampling_rate = GNUNET_le32toh (header.sampling_rate);
 159   header.gain = GNUNET_le16toh (header.gain);
 160
 161   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 162               "Header: v%u, %u-ch, skip %u, %uHz, %u gain\n",
 163               header.version,
 164               header.channels,
 165               header.preskip,
 166               header.sampling_rate,
 167               header.gain);
 168   channels = header.channels;
 169   preskip = header.preskip;
 170
 171   if (header.channel_mapping != 0)
 172   {
 173     fprintf (stderr,
 174              "This implementation does not support non-mono streams\n");
 175     return NULL;
 176   }
 177
 178   dec = opus_decoder_create (SAMPLING_RATE, channels, &err);
 179   if (OPUS_OK != err)
 180   {
 181     fprintf (stderr,
 182              "Cannot create encoder: %s\n",
 183              opus_strerror (err));
 184     return NULL;
 185   }
 186   if (! dec)
 187   {
 188     fprintf (stderr,
 189              "Decoder initialization failed: %s\n",
 190              opus_strerror (err));
 191     return NULL;
 192   }
 193
 194   if (0 != header.gain)
 195   {
 196     /*Gain API added in a newer libopus version, if we don't have it
 197       we apply the gain ourselves. We also add in a user provided
 198       manual gain at the same time.*/
 199     int gainadj = (int) header.gain;
 200     err = opus_decoder_ctl (dec, OPUS_SET_GAIN (gainadj));
 201     if(OPUS_UNIMPLEMENTED == err)
 202     {
 203       gain = pow (10.0, gainadj / 5120.0);
 204     }
 205     else if (OPUS_OK != err)
 206     {
 207       fprintf (stderr, "Error setting gain: %s\n", opus_strerror (err));
 208       return NULL;
 209     }
 210   }
 211
 212   return dec;
 213 }
 214
 215
 216 #ifdef DEBUG_DUMP_DECODED_OGG
 217 static size_t
 218 fwrite_le32(opus_int32 i32, FILE *file)
 219 {
 220    unsigned char buf[4];
 221    buf[0]=(unsigned char)(i32&0xFF);
 222    buf[1]=(unsigned char)(i32>>8&0xFF);
 223    buf[2]=(unsigned char)(i32>>16&0xFF);
 224    buf[3]=(unsigned char)(i32>>24&0xFF);
 225    return fwrite(buf,4,1,file);
 226 }
 227
 228
 229 static size_t
 230 fwrite_le16(int i16, FILE *file)
 231 {
 232    unsigned char buf[2];
 233    buf[0]=(unsigned char)(i16&0xFF);
 234    buf[1]=(unsigned char)(i16>>8&0xFF);
 235    return fwrite(buf,2,1,file);
 236 }
 237
 238
 239 static int
 240 write_wav_header()
 241 {
 242    int ret;
 243    FILE *file = stdout;
 244
 245    ret = fprintf (file, "RIFF") >= 0;
 246    ret &= fwrite_le32 (0x7fffffff, file);
 247
 248    ret &= fprintf (file, "WAVEfmt ") >= 0;
 249    ret &= fwrite_le32 (16, file);
 250    ret &= fwrite_le16 (1, file);
 251    ret &= fwrite_le16 (channels, file);
 252    ret &= fwrite_le32 (SAMPLING_RATE, file);
 253    ret &= fwrite_le32 (2*channels*SAMPLING_RATE, file);
 254    ret &= fwrite_le16 (2*channels, file);
 255    ret &= fwrite_le16 (16, file);
 256
 257    ret &= fprintf (file, "data") >= 0;
 258    ret &= fwrite_le32 (0x7fffffff, file);
 259
 260    return !ret ? -1 : 16;
 261 }
 262
 263 #endif
 264
 265
 266 static int64_t
 267 audio_write (int64_t maxout)
 268 {
 269   int64_t sampout = 0;
 270   int tmp_skip;
 271   unsigned out_len;
 272   unsigned to_write;
 273   float *output;
 274 #ifdef DEBUG_DUMP_DECODED_OGG
 275   static int wrote_wav_header;
 276
 277   if (dump_to_stdout && !wrote_wav_header)
 278   {
 279     write_wav_header ();
 280     wrote_wav_header = 1;
 281   }
 282 #endif
 283   maxout = 0 > maxout ? 0 : maxout;
 284   do
 285   {
 286     tmp_skip = (preskip > frame_size) ? (int) frame_size : preskip;
 287     preskip -= tmp_skip;
 288     output = pcm_buffer + channels * tmp_skip;
 289     out_len = frame_size - tmp_skip;
 290     if (out_len > MAX_FRAME_SIZE)
 291       exit (6);
 292     frame_size = 0;
 293
 294     to_write = out_len < maxout ? out_len : (unsigned) maxout;
 295     if (0 < maxout)
 296     {
 297       int64_t wrote = 0;
 298       wrote = to_write;
 299       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 300                   "Writing %u * %u * %u = %llu bytes into PA\n",
 301                   to_write,
 302                   channels,
 303                   (unsigned int) sizeof (float),
 304                   (unsigned long long) (to_write * channels * sizeof (float)));
 305 #ifdef DEBUG_DUMP_DECODED_OGG
 306       if (dump_to_stdout)
 307       {
 308 # define fminf(_x,_y) ((_x)<(_y)?(_x):(_y))
 309 # define fmaxf(_x,_y) ((_x)>(_y)?(_x):(_y))
 310 # define float2int(flt) ((int)(floor(.5+flt)))
 311         int i;
 312         int16_t *out = alloca(sizeof(short)*MAX_FRAME_SIZE*channels);
 313         for (i=0;i<(int)out_len*channels;i++)
 314           out[i]=(short)float2int(fmaxf(-32768,fminf(output[i]*32768.f,32767)));
 315
 316         fwrite (out, 2 * channels, out_len<maxout?out_len:maxout, stdout);
 317       }
 318       else
 319 #endif
 320       if (pa_stream_write
 321           (stream_out, output, to_write * channels * sizeof (float), NULL, 0,
 322           PA_SEEK_RELATIVE) < 0)
 323       {
 324         GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 325                     _("pa_stream_write() failed: %s\n"),
 326                     pa_strerror (pa_context_errno (context)));
 327       }
 328       sampout += wrote;
 329       maxout -= wrote;
 330     }
 331   } while (0 < frame_size && 0 < maxout);
 332
 333   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 334               "Wrote %" PRId64 " samples\n",
 335               sampout);
 336   return sampout;
 337 }
 338
 339
 340 /**
 341  * Pulseaudio shutdown task
 342  */
 343 static void
 344 quit (int ret)
 345 {
 346   mainloop_api->quit (mainloop_api,
 347                       ret);
 348   exit (ret);
 349 }
 350
 351
 352 static void
 353 ogg_demux_and_decode ()
 354 {
 355   ogg_page og;
 356   static int stream_init;
 357   int64_t page_granule = 0;
 358   ogg_packet op;
 359   static int has_opus_stream;
 360   static int has_tags_packet;
 361   static int32_t opus_serialno;
 362   static int64_t link_out;
 363   static int64_t packet_count;
 364   int eos = 0;
 365   static int total_links;
 366   static int gran_offset;
 367
 368   while (1 == ogg_sync_pageout (&oy, &og))
 369   {
 370     if (0 == stream_init)
 371     {
 372       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 373                   "Initialized the stream\n");
 374       ogg_stream_init (&os, ogg_page_serialno (&og));
 375       stream_init = 1;
 376     }
 377     if (ogg_page_serialno (&og) != os.serialno)
 378     {
 379       /* so all streams are read. */
 380       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 381                   "Re-set serial number\n");
 382       ogg_stream_reset_serialno (&os, ogg_page_serialno (&og));
 383     }
 384     /*Add page to the bitstream*/
 385     ogg_stream_pagein (&os, &og);
 386     page_granule = ogg_page_granulepos (&og);
 387     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 388                 "Reading page that ends at %" PRId64 "\n",
 389                 page_granule);
 390     /*Extract all available packets*/
 391     while (1 == ogg_stream_packetout (&os, &op))
 392     {
 393       /*OggOpus streams are identified by a magic string in the initial
 394         stream header.*/
 395       if (op.b_o_s && op.bytes >= 8 && !memcmp (op.packet, "OpusHead", 8))
 396       {
 397         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 398                     "Got Opus Header\n");
 399         if (has_opus_stream && has_tags_packet)
 400         {
 401           /*If we're seeing another BOS OpusHead now it means
 402             the stream is chained without an EOS.
 403             This can easily happen if record helper is terminated unexpectedly.
 404            */
 405           has_opus_stream = 0;
 406           if (dec)
 407             opus_decoder_destroy (dec);
 408           dec = NULL;
 409           fprintf (stderr, "\nWarning: stream %" PRId64 " ended without EOS and a new stream began.\n", (int64_t) os.serialno);
 410         }
 411         if (!has_opus_stream)
 412         {
 413           if (packet_count > 0 && opus_serialno == os.serialno)
 414           {
 415             fprintf (stderr, "\nError: Apparent chaining without changing serial number (%" PRId64 "==%" PRId64 ").\n",
 416               (int64_t) opus_serialno, (int64_t) os.serialno);
 417             quit(1);
 418           }
 419           opus_serialno = os.serialno;
 420           has_opus_stream = 1;
 421           has_tags_packet = 0;
 422           link_out = 0;
 423           packet_count = 0;
 424           eos = 0;
 425           total_links++;
 426           GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 427                       "Got header for stream %" PRId64 ", this is %dth link\n",
 428                       (int64_t) opus_serialno, total_links);
 429         }
 430         else
 431         {
 432           fprintf (stderr, "\nWarning: ignoring opus stream %" PRId64 "\n", (int64_t) os.serialno);
 433         }
 434       }
 435       if (!has_opus_stream || os.serialno != opus_serialno)
 436       {
 437         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 438                     "breaking out\n");
 439         break;
 440       }
 441       /*If first packet in a logical stream, process the Opus header*/
 442       if (0 == packet_count)
 443       {
 444         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 445                     "Decoding header\n");
 446         dec = process_header (&op);
 447         if (!dec)
 448            quit (1);
 449
 450         if (0 != ogg_stream_packetout (&os, &op) || 255 == og.header[og.header_len - 1])
 451         {
 452           /*The format specifies that the initial header and tags packets are on their
 453             own pages. To aid implementors in discovering that their files are wrong
 454             we reject them explicitly here. In some player designs files like this would
 455             fail even without an explicit test.*/
 456           fprintf (stderr, "Extra packets on initial header page. Invalid stream.\n");
 457           quit (1);
 458         }
 459
 460         /*Remember how many samples at the front we were told to skip
 461           so that we can adjust the timestamp counting.*/
 462         gran_offset = preskip;
 463
 464         if (! pcm_buffer)
 465         {
 466           GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 467                       "Allocating %u * %u * %u = %llu bytes of buffer space\n",
 468                       MAX_FRAME_SIZE,
 469                       channels,
 470                       (unsigned int) sizeof (float),
 471                       (unsigned long long) (MAX_FRAME_SIZE * channels * sizeof (float)));
 472           pcm_buffer = pa_xmalloc (sizeof (float) * MAX_FRAME_SIZE * channels);
 473         }
 474       }
 475       else if (1 == packet_count)
 476       {
 477         has_tags_packet = 1;
 478         if (0 != ogg_stream_packetout (&os, &op) || 255 == og.header[og.header_len - 1])
 479         {
 480           fprintf (stderr, "Extra packets on initial tags page. Invalid stream.\n");
 481           quit (1);
 482         }
 483       }
 484       else
 485       {
 486         int ret;
 487         int64_t maxout;
 488         int64_t outsamp;
 489
 490         /*End of stream condition*/
 491         if (op.e_o_s && os.serialno == opus_serialno)
 492         {
 493           GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 494                       "Got EOS\n");
 495           eos = 1; /* don't care for anything except opus eos */
 496         }
 497
 498         /*Decode Opus packet*/
 499         ret = opus_decode_float (dec,
 500                                  (const unsigned char *) op.packet,
 501                                  op.bytes,
 502                                  pcm_buffer,
 503                                  MAX_FRAME_SIZE, 0);
 504
 505         /*If the decoder returned less than zero, we have an error.*/
 506         if (0 > ret)
 507         {
 508           fprintf (stderr, "Decoding error: %s\n", opus_strerror (ret));
 509           break;
 510         }
 511         frame_size = ret;
 512         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 513                     "Decoded %d bytes/channel (%d bytes) from %u compressed bytes\n",
 514                     ret,
 515                     ret * channels,
 516                     (unsigned int) op.bytes);
 517
 518         /*Apply header gain, if we're not using an opus library new
 519           enough to do this internally.*/
 520         if (0 != gain)
 521         {
 522           int i;
 523           GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 524                       "Applying gain %f\n",
 525                       gain);
 526           for (i = 0; i < frame_size * channels; i++)
 527             pcm_buffer[i] *= gain;
 528         }
 529
 530         /*This handles making sure that our output duration respects
 531           the final end-trim by not letting the output sample count
 532           get ahead of the granpos indicated value.*/
 533         maxout = ((page_granule - gran_offset) * SAMPLING_RATE / 48000) - link_out;
 534         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 535                     "Writing audio packet %" PRId64 ", at most %" PRId64 " samples\n",
 536                     packet_count, maxout);
 537
 538         outsamp = audio_write (0 > maxout ? 0 : maxout);
 539         link_out += outsamp;
 540       }
 541       packet_count++;
 542     }
 543     if (eos)
 544     {
 545       has_opus_stream = 0;
 546       if (dec)
 547         opus_decoder_destroy (dec);
 548       dec = NULL;
 549     }
 550   }
 551 }
 552
 553
 554 /**
 555  * Message callback
 556  *
 557  * @param msg message we received.
 558  * @return #GNUNET_OK on success,
 559  *     #GNUNET_NO to stop further processing due to disconnect (no error)
 560  *     #GNUNET_SYSERR to stop further processing due to error
 561  */
 562 static int
 563 stdin_receiver (void *cls,
 564                 const struct GNUNET_MessageHeader *msg)
 565 {
 566   struct AudioMessage *audio;
 567   char *data;
 568   size_t payload_len;
 569
 570   (void) cls;
 571   switch (ntohs (msg->type))
 572   {
 573   case GNUNET_MESSAGE_TYPE_CONVERSATION_AUDIO:
 574     audio = (struct AudioMessage *) msg;
 575     payload_len = ntohs (audio->header.size) - sizeof (struct AudioMessage);
 576
 577     /*Get the ogg buffer for writing*/
 578     data = ogg_sync_buffer (&oy, payload_len);
 579     /*Read bitstream from input file*/
 580     GNUNET_memcpy (data, (const unsigned char *) &audio[1], payload_len);
 581     ogg_sync_wrote (&oy, payload_len);
 582
 583     ogg_demux_and_decode ();
 584     break;
 585   default:
 586     break;
 587   }
 588   return GNUNET_OK;
 589 }
 590
 591
 592 /**
 593  * Callback when data is there for playback
 594  */
 595 static void
 596 stream_write_callback (pa_stream *s,
 597                        size_t length,
 598                        void *userdata)
 599 {
 600   /* unblock 'main' */
 601   (void) userdata;
 602   (void) length;
 603   (void) s;
 604   if (-1 != ready_pipe[1])
 605   {
 606     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 607                 "Unblocking main loop!\n");
 608     (void) write (ready_pipe[1], "r", 1);
 609   }
 610 }
 611
 612
 613 /**
 614  * Exit callback for SIGTERM and SIGINT
 615  */
 616 static void
 617 exit_signal_callback (pa_mainloop_api *m,
 618                       pa_signal_event *e,
 619                       int sig,
 620                       void *userdata)
 621 {
 622   (void) m;
 623   (void) e;
 624   (void) sig;
 625   (void) userdata;
 626   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 627               _("gnunet-helper-audio-playback - Got signal, exiting\n"));
 628   quit (1);
 629 }
 630
 631
 632 /**
 633  * Pulseaudio stream state callback
 634  */
 635 static void
 636 context_state_callback (pa_context *c,
 637                         void *userdata)
 638 {
 639   int p;
 640
 641   (void) userdata;
 642   GNUNET_assert (NULL != c);
 643   switch (pa_context_get_state (c))
 644   {
 645   case PA_CONTEXT_CONNECTING:
 646   case PA_CONTEXT_AUTHORIZING:
 647   case PA_CONTEXT_SETTING_NAME:
 648     break;
 649   case PA_CONTEXT_READY:
 650   {
 651     GNUNET_assert (! stream_out);
 652     GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 653                 _("Connection established.\n"));
 654     if (! (stream_out =
 655            pa_stream_new (c, "GNUNET VoIP playback", &sample_spec, NULL)))
 656     {
 657       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 658                   _("pa_stream_new() failed: %s\n"),
 659                   pa_strerror (pa_context_errno (c)));
 660       goto fail;
 661     }
 662     pa_stream_set_write_callback (stream_out,
 663                                   &stream_write_callback,
 664                                   NULL);
 665     if ((p =
 666          pa_stream_connect_playback (stream_out, NULL,
 667                                      NULL,
 668                                      PA_STREAM_ADJUST_LATENCY | PA_STREAM_INTERPOLATE_TIMING | PA_STREAM_AUTO_TIMING_UPDATE,
 669                                      NULL,  NULL)) < 0)
 670     {
 671       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 672                   _("pa_stream_connect_playback() failed: %s\n"),
 673                   pa_strerror (pa_context_errno (c)));
 674       goto fail;
 675     }
 676     break;
 677   }
 678   case PA_CONTEXT_TERMINATED:
 679     quit (0);
 680     break;
 681
 682   case PA_CONTEXT_FAILED:
 683   default:
 684     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 685                 _("Connection failure: %s\n"),
 686                 pa_strerror (pa_context_errno (c)));
 687     goto fail;
 688   }
 689   return;
 690  fail:
 691   quit (1);
 692 }
 693
 694
 695 /**
 696  * Pulseaudio initialization
 697  */
 698 static void
 699 pa_init ()
 700 {
 701   int r;
 702
 703   if (!pa_sample_spec_valid (&sample_spec))
 704   {
 705     GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 706                 _("Wrong Spec\n"));
 707   }
 708   /* set up threaded playback mainloop */
 709   if (!(m = pa_threaded_mainloop_new ()))
 710   {
 711     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 712                 _("pa_mainloop_new() failed.\n"));
 713   }
 714   mainloop_api = pa_threaded_mainloop_get_api (m);
 715   /* listen to signals */
 716   r = pa_signal_init (mainloop_api);
 717   GNUNET_assert (r == 0);
 718   pa_signal_new (SIGINT, exit_signal_callback, NULL);
 719   pa_signal_new (SIGTERM, exit_signal_callback, NULL);
 720
 721
 722   /* connect to the main pulseaudio context */
 723   if (!(context = pa_context_new (mainloop_api, "GNUnet VoIP")))
 724   {
 725     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 726                 _("pa_context_new() failed.\n"));
 727   }
 728   pa_context_set_state_callback (context, context_state_callback, NULL);
 729
 730   if (pa_context_connect (context, NULL, 0, NULL) < 0)
 731   {
 732     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 733                 _("pa_context_connect() failed: %s\n"),
 734                 pa_strerror (pa_context_errno (context)));
 735   }
 736   if (pa_threaded_mainloop_start (m) < 0)
 737   {
 738     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 739                 _("pa_mainloop_run() failed.\n"));
 740   }
 741 }
 742
 743
 744 static void
 745 ogg_init ()
 746 {
 747   ogg_sync_init (&oy);
 748 }
 749
 750
 751 static void
 752 drain_callback (pa_stream*s, int success, void *userdata)
 753 {
 754   (void) s;
 755   (void) success;
 756   (void) userdata;
 757   pa_threaded_mainloop_signal (m,
 758                                0);
 759 }
 760
 761
 762 /**
 763  * The main function for the playback helper.
 764  *
 765  * @param argc number of arguments from the command line
 766  * @param argv command line arguments
 767  * @return 0 ok, 1 on error
 768  */
 769 int
 770 main (int argc, char *argv[])
 771 {
 772   static unsigned long long toff;
 773   char readbuf[MAXLINE];
 774   struct GNUNET_MessageStreamTokenizer *stdin_mst;
 775   char c;
 776   ssize_t ret;
 777 #ifdef DEBUG_READ_PURE_OGG
 778   int read_pure_ogg = getenv ("GNUNET_READ_PURE_OGG") ? 1 : 0;
 779 #endif
 780
 781   (void) argc;
 782   (void) argv;
 783   GNUNET_assert (GNUNET_OK ==
 784                  GNUNET_log_setup ("gnunet-helper-audio-playback",
 785                                    "WARNING",
 786                                    NULL));
 787   if (0 != pipe (ready_pipe))
 788   {
 789     GNUNET_log_strerror (GNUNET_ERROR_TYPE_ERROR, "pipe");
 790     return 1;
 791   }
 792   stdin_mst = GNUNET_MST_create (&stdin_receiver, NULL);
 793   ogg_init ();
 794   pa_init ();
 795   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 796               "Waiting for PulseAudio to be ready.\n");
 797   GNUNET_assert (1 == read (ready_pipe[0], &c, 1));
 798   close (ready_pipe[0]);
 799   close (ready_pipe[1]);
 800   ready_pipe[0] = -1;
 801   ready_pipe[1] = -1;
 802 #ifdef DEBUG_DUMP_DECODED_OGG
 803   dump_to_stdout = getenv ("GNUNET_DUMP_DECODED_OGG") ? 1 : 0;
 804 #endif
 805   while (1)
 806   {
 807     ret = read (STDIN_FILENO,
 808                 readbuf,
 809                 sizeof (readbuf));
 810     toff += ret;
 811     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 812                 "Received %d bytes of audio data (total: %llu)\n",
 813                 (int) ret,
 814                 toff);
 815     if (0 > ret)
 816     {
 817       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 818                   _("Read error from STDIN: %s\n"),
 819                   strerror (errno));
 820       break;
 821     }
 822     if (0 == ret)
 823       break;
 824 #ifdef DEBUG_READ_PURE_OGG
 825     if (read_pure_ogg)
 826     {
 827       char *data = ogg_sync_buffer (&oy, ret);
 828       GNUNET_memcpy (data, readbuf, ret);
 829       ogg_sync_wrote (&oy, ret);
 830       ogg_demux_and_decode ();
 831     }
 832     else
 833 #endif
 834     GNUNET_MST_from_buffer (stdin_mst,
 835                             readbuf, ret,
 836                             GNUNET_NO, GNUNET_NO);
 837   }
 838   GNUNET_MST_destroy (stdin_mst);
 839   if (stream_out)
 840   {
 841     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 842                 "Locking\n");
 843     pa_threaded_mainloop_lock (m);
 844     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 845                 "Draining\n");
 846     pa_operation *o = pa_stream_drain (stream_out, drain_callback, NULL);
 847     while (pa_operation_get_state (o) == PA_OPERATION_RUNNING)
 848     {
 849       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 850                   "Waiting\n");
 851       pa_threaded_mainloop_wait (m);
 852     }
 853     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 854                 "Unreffing\n");
 855     pa_operation_unref (o);
 856     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 857                 "Unlocking\n");
 858     pa_threaded_mainloop_unlock (m);
 859   }
 860   return 0;
 861 }